[Kafka Indexing Service] Decouple Druid segment partitioning from Kafka partitioning

[pjain1]

Currently Kafka indexing task creates a segment partition corresponding to a Kafka partition. However, in many cases Kafka partition does not directly indicate the amount of data that ends up in Druid segments and as a result many times the tasks end up creating a lot of small segments.

Therefore, this proposal is to decouple the segment partitioning from Kafka partitioning and thus allowing to tune Kafka partitions and Druid segment partitions independently. Thus, a Kafka indexing task will create a single segment per segment interval (still consuming from multiple Kafka partitions) and number of tasks running concurrently will decide the number of segment partitions.

Any thoughts ?

Edit:
After having discussion with @cheddar and @himanshug we came up with few ideas -

1. As soon as a task hits the maxRowsInSegment limit it should signal Supervisor about it and then Supervisor essentially pauses all replica tasks, gets the highest offset for all partitions and sets the new offset on all replicas; asks them to publish and shutdown and start new set of tasks for consuming further offsets. In this case it is a little hard to reason about the number of concurrently running tasks and task duration. Also, the taskDuration config will be ignored.

2. As soon as a task hits the maxRowsInSegment limit it should signal Supervisor about it and then Supervisor essentially pauses all replica tasks, gets the highest offset for all partitions and sends these partition offset tuples as a check point to all tasks. If any task fetches a record with offset for any partition which is beyond this check point then the task will create a new appenderator driver with a new appenderator and hands out this record to the new appenderator for consumption. For each record pulled from Kafka the task will check the offset and hands out the record to relevant appenderator based on the check points it has. Once the task has consumed past all the offsets for all partitions for an individual check point then it can close the relevant appenderator and will hand off the segments handled by this appenderator. When the taskDuration is reached for a task then it will publish (or wait for) hand off of segments for each check point in FIFO manner before shutdown.

Second approach looks better and I will try to implement it if it sounds good.

[gianm]

The task does a separate segment per partition to make replicated ingestion work, see #2703 and #2844 for description of what can go wrong if multiple Kafka partitions go into the same Druid segment.

It is a big pain point though so maybe there's a creative solution to this problem…

[pjain1]

@gianm If ordering is a problem then polled records can be ordered by partition first. So, all records for a single partition are consumed and then records for next partitions and so on. Or records can be sorted on combination of (partition + offset) and then consumed. This way ordering can be maintained independently at each replica. Does this make sense ?

[gianm]

I think that wouldn't work in practice when Druid is caught up to the Kafka stream (the normal case for real time indexing), it would mean that it won't read later partitions until very late in the task.

[pjain1]

I meant ordering events after they have been polled basically after each consumer.poll() call. I did not meant that the task should consume events from one partition and then gets assigned to next partition and consume that and so on, this will not work.

However, I can see a problem in ordering after each poll call as well, if poll returns different number of records for partitions (or some partition records are missing) on replica tasks then also the segments can go out of sync.

[gianm]

Ah I see. In that case, I think you're right, there's a problem since there's no guarantee that different replicas will get the exact same batches from poll.

[himanshug]

one idea that floated was to let KafkaIndexTask create segments per kafka partition, eventually when time comes to publish those segments ... merge all "per partition" segments into a final segment and publish that one instead.

[gianm]

@himanshug hmm that sounds like it could work. The final segment would need to have a higher version number than the earlier ones but that should be doable. There might also be more than one final segment if the task handled a lot of data.

[himanshug]

@gianm actually final segment should have a higher version as it should overshadow the "per partition" segments served from peons

[gianm]

yeah, that makes sense.

[himanshug]

ok, had a short chat with @pjain1 and it appears merging at the end is not enough as the original problem faced by customer is related to slower queries due to many small "realtime" segments which would exist even when we do merging at the end.

[gianm]

Presumably it would still be useful though to your customer, since it would mean a lot fewer segments on historicals after handoff.

[himanshug]

another thing we briefly chatted about removing "maxRowsPerSegment" setting and in that world segment size would be controlled by number of of kafka partitions per task and segment granularity.

[himanshug]

Presumably it would still be useful though to your customer, since it would mean a lot fewer segments on historicals after handoff.

they run batch reindexing to do the merging already.