perf: batch reset invalid stream offsets in stream supervisor

[jtuglu1]

Description

Problem

I've noticed that recent Druid can be extremely slow to recover from a lagging supervisor (where offsets are invalid and tuningConfig.resetOffsetAutomatically=true). I have observed the following issue occurs:

1. Kafka supervisor falls behind, offsets are invalid
2. Supervisor starts creating a new task group
3. Supervisor runs into an invalid offset for a particular partition, throwing an exception. This exception propagates up to the supervisor, forcing us to wait until another round of runInternal(), which can take O(XXs) depending on the supervisor run loop configuration. Since these partitions are evaluated serially, if multiple partitions are invalid offsets, this means we keep looping for every invalid offset for every partition for every taskGroup, leading to many minutes of downtime. This issue is exacerbated when ingesting from large Kafka topics with many partitions.

Solution

For each taskGroup, identify + reset all invalid partition offsets in one go. Once we have established the set of invalid offsets, perform an internal reset, then throw the exception. This reduces the time-to-recovery of a fatally-lagged supervisor from N runInternal() calls to 1 runInternal() call, where N is the # of invalid partition offsets.

Release note

Batch reset invalid stream offsets in stream supervisor to speed up stream supervisor recovery.

---

This PR has:

• been self-reviewed.
  • using the concurrency checklist (Remove this item if the PR doesn't have any relation to concurrency.)
• added documentation for new or modified features or behaviors.
• a release note entry in the PR description.
• added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
• added or updated version, license, or notice information in licenses.yaml
• added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
• added integration tests.
• been tested in a test Druid cluster.

[jtuglu1]

LMK what you think about this – I think this shouldn't impact task groups without invalid offsets as currently they'd just be added to the activelyReadingTaskGroups then be delayed in processing (see few lines down) since generateStartingSequencesForPartitionGroup would just throw an exception (and force a new runInternal()).