Segment publishing order should be preserved in kafka indexing service

[jihoonson]

In Kafka indexing service, the overlord does a sanity check that the start offsets of partitions of current publishing segments are same with the ones stored in metastore, so that it guarantees that all segments are published in order. Because of this check, some tasks might fail in this scenario.

1. The supervisor created a task (T1) with a start offset O1.
2. Somehow, the supervisor couldn't send an endOffset O2 to T1 in taskDuration. Instead, it sent an endOffset O3 to T1 after taskDuration * 10. (In our case, supervisor couldn't send because of too frequent HTTP connection refused errors.)
3. T1 started to merge, push, and publish segments.
4. The supervisor created a new task, T2, with a start offset O3.
5. After taskDuration, it sent an endOffset O4 to T2.
6. T2 started to merge, push, and publish segments.
7. Since T1 had run for a much longer time, it had much more segments to publish than T2. As a result, T2 tried to publish before T1 complete publishing.
8. T2 failed to publish because of the sanity check when updating metastore.

So, I think the supervisor should be able to guarantee segment publishing order across all running tasks like below.

T1: indexing ===> publishing ===> handoff
              T2: indexing ===> publishing ===> handoff
                            T3: indexing ===> publishing ===> handoff
                                          ...

[jihoonson]

I think this isn't a blocking issue for 0.13.0 release since this bug happens pretty rarely. Let's fix in another release.

[hellobabygogo]

In Kafka indexing service, the overlord does a sanity check that the start offsets of partitions of current publishing segments are same with the ones stored in metastore, so that it guarantees that all segments are published in order. Because of this check, some tasks might fail in this scenario.

1. The supervisor created a task (T1) with a start offset O1.
2. Somehow, the supervisor couldn't send an endOffset O2 to T1 in taskDuration. Instead, it sent an endOffset O3 to T1 after taskDuration * 10. (In our case, supervisor couldn't send because of too frequent HTTP connection refused errors.)
3. T1 started to merge, push, and publish segments.
4. The supervisor created a new task, T2, with a start offset O3.
5. After taskDuration, it sent an endOffset O4 to T2.
6. T2 started to merge, push, and publish segments.
7. Since T1 had run for a much longer time, it had much more segments to publish than T2. As a result, T2 tried to publish before T1 complete publishing.
8. T2 failed to publish because of the sanity check when updating metastore.

So, I think the supervisor should be able to guarantee segment publishing order across all running tasks like below.

T1: indexing ===> pushing ===> publishing ===> handoff
              T2: indexing ===> pushing ===> publishing ===> handoff
                            T3: indexing ===> pushing ===> publishing ===> handoff
                                          ...

To do so, I suppose the supervisor should be able to send pushing signals to kafka tasks as well as publishing signals.

good idea!

[gianm]

Would it work to hook into the checkpointing logic? As in, have the supervisor not tell tasks in a group to resume until after any earlier groups (offset-wise) have completed. This would have the effect of having the T2 tasks stay paused until the T1 tasks have finished.

[jihoonson]

@gianm that sounds a possible option, but, I don't want to block the stream ingestion if possible. I think it's possible to not block if we can add a new signal which the supervisor can send to tasks to start publishing segments. The changed algorithm I'm thinking is like

1. Checkpointing is initialized by the task or the supervisor
2. The supervisor sends pause request to all tasks in the same taskGroup.
3. Tasks send their current offsets to the supervisor.
4. The supervisor finds the max and sends it to all tasks in the taskGroup.
5. Tasks resume and read up to the updated endOffsets. They generate and push segments, but wouldn't publish them immediately. Instead, they add the set of segments to a queue and send a publish request to the supervisor.
6. The supervisor adds the publish request to a queue. This publish request contains a taskGroup ID.
7. If it's the first request of the previous publish has finished, the supervisor pops a request from the queue and sends a signal to tasks to start publishing.
8. Tasks pops a set of segments from its queue, publishes them, and tells the supervisor that publish is finished.

So, 1-4 steps are same with the current, but 5-8 is to guarantee the publish order across taskGroups. What do you think?

[jihoonson]

Probably we need a way for throttling to prevent the queues in the task or the supervisor from becoming huge. I think the task can pause itself if its queue is too large. The supervisor can reject the publish request if its queue is too large.

[gianm]

Hmm, what about having the supervisor publish the segments itself, so steps (7) and (8) are not necessary? Nothing really says that tasks need to publish their own segments.

[jihoonson]

i was thinking about it in #5492, but ended up not doing it because publishing segments is currently allowed for only tasks but the supervisor is not a task.

[github-actions]

This issue has been marked as stale due to 280 days of inactivity.
It will be closed in 4 weeks if no further activity occurs. If this issue is still
relevant, please simply write any comment. Even if closed, you can still revive the
issue at any time or discuss it on the dev@druid.apache.org list.
Thank you for your contributions.

[github-actions]

This issue has been closed due to lack of activity. If you think that
is incorrect, or the issue requires additional review, you can revive the issue at
any time.