maytasm opened a new pull request, #20741:
URL: https://github.com/apache/druid/pull/20741
### Description
When a pending completion task group fails (no task succeeds within
`completionTimeout`, or all of its tasks fail),
`SeekableStreamSupervisor.checkPendingCompletionTasks` resets the group's
partitions, kills the actively reading task group with the **same groupId**,
and stops every later pending completion task group in the same list.
The groupId only identifies the same partitions while `taskCount` is
unchanged. After a task count change (for example, by the lag-based
autoscaler), the reading task groups that actually started from the failed
group's unpublished end offsets have other groupIds. They keep running, and
their tasks fail to publish with `The new start metadata state[...] is ahead of
the last committed end state[...]`. The next rollover copies their end offsets
into `partitionOffsets`, so the gap is carried forward and the supervisor does
not recover on its own. Resetting the supervisor recovers it, but skips the
unpublished data.
We saw this in production when one task's segment push failed with a
transient error at the same moment as an autoscaler scale action.
#### Kill the reading task groups by partition
The new `resetPartitionsAndKillReadingTaskGroups` resets the partitions of
the failed task group and kills every actively reading task group that:
- reads any of these partitions from an offset other than the committed
offset in metadata storage, or
- is assigned one of these partitions in `partitionGroups` but is not
reading it, such as a closed Kinesis shard whose end offset was never
published. Otherwise, that partition would not be re-read until the task group
rolls over.
Reading task groups that started from the committed offsets are kept. For
example, when a pending completion group published its offsets but timed out
waiting for handoff, the readers that started from its end offsets keep
running, whereas before, the reader with the same groupId was killed.
The reset and the kills now happen once per pending completion list, after
all groups in the list are checked and before `onCompleted()`. This way, the
killed tasks' publish waits resolve as not ready before the completion
listeners would let them publish.
#### Stop only the overlapping later pending completion groups
Later pending completion task groups in the same list are now stopped only
if they overlap the partitions of a failed or stopped group, instead of all of
them. Their partitions are reset as well.
For single-topic Kafka, every task group in pending completion list `k`
contains the partition with rank `k`, so this part has no effect there. It
matters for Kinesis and multi-topic Kafka, where task groups with the same
groupId can have disjoint partitions.
#### Known limitation
The failed partitions are still tracked per pending completion list. A
pending completion group in a *different* list that started from the failed
group's unpublished end offsets is not stopped, and is released when the failed
group completes. Its publish then fails, and its own failure goes through the
fixed path, so the supervisor recovers with one extra restart. This behavior is
unchanged by this PR. Fixing it properly needs an offset-ordered check across
all lists, similar to `getOlderTaskGroupsPublishingToPartitions`, which I'd
prefer to do in a follow-up.
#### Release note
Fixed Kafka and Kinesis supervisors getting stuck in repeated task failures
with "The new start metadata state ... is ahead of the last committed end
state" after a publishing task group failed around a task count change, such as
during autoscaling.
<hr>
##### Key changed/added classes in this PR
* `SeekableStreamSupervisor`
* `SeekableStreamSupervisorStateTest`
<hr>
This PR has:
- [x] been self-reviewed.
- [x] a release note entry in the PR description.
- [x] added Javadocs for most classes and all non-trivial methods. Linked
related entities via Javadoc links.
- [x] added comments explaining the "why" and the intent of the code
wherever would not be obvious for an unfamiliar reader.
- [x] added unit tests or modified existing tests to cover new code paths,
ensuring the threshold for [code
coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md)
is met.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
