[Proposal] Add batching support in kafka-indexing-service

[himanshug]

It would benefit some users if kafka-indexing service supported the capability to let users put multiple druid InputRow inside single kafka record.
This allows users to do batching while still using kafka sync producer which allows only one kafka record at a time.

I would imagine adding following method to InputRowParser interface.

  default List<InputRow> parseBatch(T input)
  {
    return ImmutableList.of(parse(input));
  }

current InputRow parse(T input) method would be deprecated and all of druid code would be adjusted to use parseBatch(input) instead.

kafka-indexing service will need to persist <offset,row-number-in-record> pair instead of just offset to support exactly once. I believe, this will end up being the biggest change really.

This approach would add batching support in general and not just for kafka-indexing .

[gianm]

See also #3439.

[himanshug]

We had few conversations internally and it appears tracking record-offset within kafka record introduces additional metadata to be tracked and increases code complexity for the Kafka Indexing Service related code. So, for now the proposal is to not track record-offset within kafka record and that means if InputRowParser.parseBatch(kafka-record-bytes) returns more than one InputRow then exactly-once behavior is not guaranteed but best effort. This behavior could be changed in future if necessary.
So, for now, a user who requires exactly-once guarantee would use batching support provided by kafka so that each kafka record stays single Druid row.

@gianm does that sound ok?

[pjain1]

The parse batch thing in KafkaIndexTask can be put behind a config, which when set explicitly will enable parsing of batch. This config can be documented with the warning that, when enabled task does not support exactly once and can also be printed as warning message in the task logs.

[gianm]

@himanshug In principle, it sounds ok to me to have some non-exactly-once-features as long as they're disabled by default and clearly marked. I think by default we should be exactly once and users should need to change configs somehow if they want to break that.

However there is a difficulty with dropping exactly-once behavior: the segment-allocate stuff (pending segments etc) assumes that replica tasks will do exactly the same thing given the same set of inputs. If they don't do exactly the same thing then they can get out of sync with regard to which messages go in which segments, and that can cause the segment sets not to line up, and that can cause allocation errors.

Instead what do you think about keeping exactly-once behavior by basically forcing an entire batch to be processed all-or-nothing? In practice it would mean adding an addAll method to Appenderator, and not doing any persists or publishes in the middle of a batch. To handle the situation where a limit (maxRowsPerSegment or something) could be hit in the middle of a batch, we could either persisting early (before a batch) or late (by allowing to going a little past the limit).

I think this might be the cleanest approach.

[himanshug]

@gianm sure, having explicit config to enable batching support would be there ( that is what @pjain1 explained in #4373 (comment) ) . so if that config is not explicitly enabled and we got multiple elements in list returned by InputRowParser.parseBatch(kafka-record-bytes) then task would just loudly fail.

I think you mean adding addAll(..) method to AppenderatorDriver as all rows in the batch might not belong to same segment. In any case, we would have something like that to ensure replica consistency or else would add more comments here.

[gianm]

Ah yeah I did mean addAll on AppenderatorDriver.

If you do that then I think we don't need to worry about breaking exactly-onceness. As long as AppenderatorDriver can promise to either index the whole batch, or none of it (and I think it can promise that) then exactly-onceness is preserved.

What do you think?

[himanshug]

@gianm yes adding AppenderatorDriver.addAll(..) is fine. However regarding promising indexing of whole batch or none, do you mean to ignore "maxRowsInMemory" so that you index whole batch ? or do you mean adding some batch add capability to IncrementalIndex as well where it either accepts or rejects whole batch?

[gianm]

I was thinking of the Appenderator having an addAll method too, and it should take care of making sure that the following are true:

1. a commit must not happen between two rows from the same batch
2. all rows in the same batch must end up in the same segment once the merge happens

(1) might require ignoring "maxRowsInMemory" by going slightly under or over it. If it's going over, it will need some changes to IncrementalIndex. If it's going under, it will probably require implementing a "persist without committing" method in Appenderator such that the earlier sink can be persisted with part of a batch, and a new sink can be created to hold the rest of the batch. The "without committing" is important since it would be a violation to persist+commit a partial batch.

(2) might require ignoring "maxRowsPerSegment" by going slightly under or over it.

I think this all can be done in a deterministic way, such that replicas still make the same decisions as each other and we can still enable exactly-once ingestion without needing to track offsets-within-batches.

[himanshug]

@gianm I and @pjain1 talked about it a little and it appears that things stay simple by going a bit over for both maxRowInMemory and maxRowsPerSegment. So, let us update IncrementalIndex so that it can optionally go over max-size when asked. this would maintain exactly-once. (sidenote: we are already going "over" for maxRowsInSegment limit in #4815 to maintain consistency of checkpoints across replicas )
above is probably simpler than the going "under" and "persist without commit" approach where we might have to keep disk state inconsistent (persisted data having more events than what is stored in commit metadata) for prolonged periods or persisting small sinks depending on implementation.

[gianm]

That sounds good to me. I agree that going a little bit over maxRowsInMemory and maxRowsPerSegment will lead to the simplest solution.