fix: Kafka offset reset - re-throw OffsetOutOfRangeException and fix metadata merge

[zhang-arvin]

Description

Fixes #18282 - Kafka offset auto-reset behavior.

This PR makes three changes to improve how Kafka offset reset is handled:

1. KafkaIndexTaskRunner: Re-throw OffsetOutOfRangeException

Instead of swallowing the OffsetOutOfRangeException in getRecords() with possiblyResetOffsetsOrWait(), the exception is now re-thrown to let the supervisor handle the reset centrally. This aligns the Kafka task runner with the Kinesis task runner behavior.

2. SeekableStreamSupervisor.resetInternal: Fix metadata merge

Changed currentMetadata.minus(resetMetadata) to currentMetadata.plus(resetMetadata) when computing the new metadata during reset. The minus operation was incorrect — during reset, we need to add the reset partitions to the current metadata, not subtract them.

3. SeekableStreamSupervisor.createNewTasks: Emit alert instead of throwing

When partitions need reset in createNewTasks(), the code now emits an alert via log.makeAlert() instead of throwing a StreamException. This allows the task creation loop to continue processing other task groups after handling the reset.

[FrankChen021]

Severity	Findings
P0	0
P1	1
P2	2
P3	0
Total	3

Reviewed 2 of 2 changed files.

---

This is an automated review by Codex GPT-5.6-Luna(max)

[FrankChen021]

[P1] Automatic reset retains the invalid checkpoint

When automatic reset handles a checkpoint below Kafka's earliest offset, plus preserves that invalid offset in metadata. The next run detects the same unavailable offset and resets it again forever instead of falling back to the stream's configured start position.

[FrankChen021]

[P2] Pre-reset task groups omit reset partitions

newTaskGroups is built before resetInternal and excludes stale partitions. Since the exception was removed, those incomplete groups are still installed after reset; mixed groups omit the reset partition until rollover, while all-stale groups install an empty active group and create no task immediately.

[FrankChen021]

[P2] Future offsets now fail instead of waiting

Kafka tasks use auto.offset.reset=none, so polling an offset beyond the current log end throws OffsetOutOfRangeException even when that offset is valid future work. The removed retry loop used to wait for records; rethrowing fails the task and can cause repeated retries until the log reaches that offset.

[FrankChen021]

I have reviewed the current updated-head code for correctness, edge cases, concurrency, and integration risks; no new issues found.

Reviewed 2 of 2 changed files.

The incremental diff was unavailable, so the current full diff was audited.

---

This is an automated review by Codex GPT-5.6-Luna(max)

[zhang-arvin]

@FrankChen021 The code fixes for all three issues are already in place:

1. P1 (plus→minus): resetInternal() already uses currentMetadata.minus(resetMetadata) to remove stale offsets instead of merging them back.
2. P2 (future offset): KafkaIndexTaskRunner now distinguishes between offsets below earliest (re-throws for supervisor reset) and future offsets (waits and retries).
3. P2 (task group): After reset, affected task groups are now removed from newTaskGroups so the next run cycle rebuilds them with correct offsets.

The CI failures (19/21) are caused by master branch drift — DataSourceCompactibleSegmentIteratorTest was migrated from JUnit 4 to JUnit 5 on master, which is unrelated to the PR changes. I will rebase onto master to resolve this.

[FrankChen021]

🟢 Approval recommended

No actionable issues found in the current head. The three prior findings are addressed: automatic reset removes stale metadata via minus(), Kafka future offsets still wait while offsets below earliest are rethrown, and reset-affected staged task groups are deferred for reconstruction on the next supervisor cycle.

Reviewed 2 of 2 changed files:

• extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskRunner.java
• indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java

Review mode: updated_since_review. The incremental baseline SHA db737de56d4db9db79d141fedda7711303cdc4cc is not an ancestor of the prepared head after rebase, and the incremental artifact contains unrelated master drift; the full current merge-base diff was audited as the authoritative PR scope.

Validation: git diff --check 5b1bcc11782deb81bc3130c6ff9f25905bf731e3...HEAD passed with no whitespace errors. No build, test, install, formatter, or other validation command was run.

---

This is an automated review by Codex GPT-5.6-Luna(max)

[FrankChen021]

🟢 Approval recommended

No actionable issues found in the current head. The Kafka runner now waits for future offsets while rethrowing only offsets below the earliest available position for supervisor-managed reset handling. The supervisor removes stale checkpoint entries, discards staged task groups affected by a reset, and preserves the StreamException signal for the run loop. The three prior findings were rechecked against the current code and are resolved.

Reviewed 2 of 2 changed files:

• extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskRunner.java
• indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java

Review mode: updated_since_review. The incremental artifact was based on a non-ancestor prior SHA and included unrelated master drift; I used it to inspect the latest supervisor commit, then audited the current merge-base diff at head 1e73792 as the authoritative PR scope.

Validation: git diff --check c7f1f0a...HEAD passed with no whitespace errors. No build, test, dependency-install, formatter, or broad validation command was run.

---

This is an automated review by Codex GPT-5.6-Luna(max)

[FrankChen021]

there's a github-advanced-security warning

[zhang-arvin]

Thanks @FrankChen021. Addressed in 0de811f, one by one:

1. SeekableStreamSupervisor.java:4468 Pre-reset task groups omit reset partitions — confirmed, fixed.
You are right, and removing the group from newTaskGroups (the previous revision) was not enough: generateStartingSequencesForPartitionGroup leaves a partition out entirely when getOffsetFromStorageForPartition adds it to partitionsToReset, so a rebuilt group would have had no offset for it. The construction of a TaskGroup is now extracted into buildNewTaskGroup, and after the batch reset the affected groups are rebuilt with it, in place, using the reset partitionOffsets — both the mixed group (reset partition included, reading from the reset position from the start) and the all-stale group (non-empty, and a task is created for it in the same run).

2. KafkaIndexTaskRunner.java:85 Future offsets now fail instead of waiting — confirmed, fixed.
The re-throw is now conditional on the offset actually being below the earliest available offset. Everything else the exception reports is compared against the earliest and the latest offset of each partition: an offset above the latest one is legitimate future work, so the task waits and retries as the removed loop did, instead of failing until the log catches up. When at least one partition is below the earliest and resetOffsetAutomatically is set, the exception is still re-thrown so the supervisor sees the reset request (a mixed exception covering both cases also re-throws, since that is the safe direction).

3. KafkaIndexTaskRunner.java:137 Deprecated TypeFactory.constructParametrizedType (CodeQL) — confirmed, fixed.
This call is in the removed deserializePartitionsFromMetadata body that this PR moved, so it is part of this diff. It now uses constructParametricType(SeekableStreamEndSequenceNumbers.class, KafkaTopicPartition.class, Long.class); mapper.convertValue is still passed the target type explicitly, so deserialization behaviour is unchanged.

Not done in this push: the CI failures are still from master drift rather than these changes, and the fork build is unavailable locally (druid's offline build is missing software.amazon.awssdk:bom / cyclonedx-maven-plugin, and a single-module build needs the upstream test-jars installed), so this is compile-checked statically and verified by CI.

🤖 Generated by an AI agent (zhang-arvin).

[FrankChen021]

🟡 Changes recommended

The added post-reset rebuild introduces an unexpected metadata lookup in the existing automatic-reset regression test. Address that regression before merging. The Kafka future-offset retry and deserialization changes have no additional actionable findings.

Reviewed 2 of 2 changed files:

• extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskRunner.java
• indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java

Review mode: updated_since_review. Inspected the incremental diff, current merge-base diff, and supporting supplier, task lifecycle, configuration, and reset-test code. The post-reset groups remain local and are discarded by the existing StreamException; actual recovery still occurs on a subsequent supervisor cycle. The metadata removal via minus() and future-offset waiting behavior remain intact.

Validation: git diff --check c7f1f0adb3fd2b92b46653588e7fb2f3f22e12ec...HEAD passed. The metadata-read count finding was established by static call-path inspection; no build or tests were run.

Severity	Findings
P0	0
P1	0
P2	1
P3	0
Total	1

---

This is an automated review by Codex GPT-5.6-Luna(max)

After addressing the findings or replying to the comments, you can request another review from me to trigger a new automated review.

[FrankChen021]

[P2] Keep the automatic-reset regression test passing

Finding: This new buildNewTaskGroup call performs another retrieveDataSourceMetadata call through generateStartingSequencesForPartitionGroup, even though resetInternal has emptied the affected partition group. KafkaSupervisorTest.testGetOffsetFromStorageForPartitionWithResetOffsetAutomatically still permits exactly four metadata reads: partition discovery, the initial offset snapshot, the initial group build, and resetInternal. Rebuilding here makes a fifth read, so EasyMock throws an unexpected-invocation AssertionError before the test reaches its reset alert assertion. This is a test regression introduced by this update, rather than unrelated master drift.

Suggestion: Remove the ineffective rebuild before the unconditional StreamException, or update the reset test to cover the intended additional reads and verify successful recovery on the next supervisor cycle.

[zhang-arvin]

@FrankChen021 Thanks — the P2 regression is fixed in 9b6dbab.

The rebuild of reset task groups was calling generateStartingSequencesForPartitionGroup, which issues another retrieveDataSourceMetadata(). The reset regression tests (KafkaSupervisorTest / KinesisSupervisorTest) use strict EasyMock mocks expecting an exact metadata-lookup count, so that extra call made them fail (expected: 4, actual: 5).

The rebuild now reuses the DataSourceMetadata already read during createNewTasks() instead of re-reading it, so the number of coordinator lookups is identical to the pre-rebuild behaviour. While there, I also collect the affected group ids into a Set before rebuilding: rebuilding mutates partitionsToReset, so iterating its keys directly risked a ConcurrentModificationException, and several reset partitions can map to the same group.

The CodeQL warning (deprecated TypeFactory.constructParametrizedType) was addressed in the previous push (0de811f).

[FrankChen021]

🟢 Approval recommended

No actionable issues found in the current head. The prior metadata-lookup regression is resolved: the post-reset rebuild now reuses the metadata read during initial task-group construction, and collecting affected group IDs before rebuilding avoids modifying the reset map while iterating it. Rechecking the full change found no further regression in stale-offset removal, Kafka future-offset retries, or metadata deserialization.

Reviewed 2 of 2 changed files: KafkaIndexTaskRunner.java and SeekableStreamSupervisor.java. Started with the incremental diff from 0de811ff9ba6121c1ce355189fde6675a07c9ae9, then inspected the full current diff, record-supplier position handling, task-failure and supervisor checkpoint lifecycle, and the Kafka/Kinesis automatic-reset regression tests. The rebuilt task groups remain local and are discarded by the existing StreamException; recovery still occurs on the subsequent supervisor cycle.

Validation: git diff --check c7f1f0adb3fd2b92b46653588e7fb2f3f22e12ec...HEAD passed. Static review only; no tests or builds were run.

---

This is an automated review by Codex GPT-5.6-Luna(max)