fix: Kafka offset reset - re-throw OffsetOutOfRangeException and fix metadata merge - #20092
fix: Kafka offset reset - re-throw OffsetOutOfRangeException and fix metadata merge#20092zhang-arvin wants to merge 1 commit into
Conversation
FrankChen021
left a comment
FrankChen021
left a comment
There was a problem hiding this comment.
| 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)
| metadataUpdateSuccess = true; | ||
| } else { | ||
| final DataSourceMetadata newMetadata = currentMetadata.minus(resetMetadata); | ||
| final DataSourceMetadata newMetadata = currentMetadata.plus(resetMetadata); |
There was a problem hiding this comment.
[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.
| log.makeAlert( | ||
| "Previous sequenceNumbers are no longer available - automatically resetting sequences" | ||
| ).addData("partitions", partitionsToReset).emit(); | ||
| resetInternal(createDataSourceMetaDataForReset(ioConfig.getStream(), partitionsToReset)); |
There was a problem hiding this comment.
[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.
| log.warn("OffsetOutOfRangeException with message [%s]", e.getMessage()); | ||
| possiblyResetOffsetsOrWait(e.offsetOutOfRangePartitions(), recordSupplier, toolbox); | ||
| return Collections.emptyList(); | ||
| throw e; |
There was a problem hiding this comment.
[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.
…metadata merge - KafkaIndexTaskRunner: re-throw OffsetOutOfRangeException instead of swallowing it with possiblyResetOffsetsOrWait, letting the supervisor handle the reset centrally - SeekableStreamSupervisor.resetInternal: use plus() instead of minus() when merging reset metadata with current metadata - SeekableStreamSupervisor.createNewTasks: emit alert instead of throwing StreamException when partitions need reset
f800295 to
db737de
Compare
2 participants
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
OffsetOutOfRangeExceptioningetRecords()withpossiblyResetOffsetsOrWait(), 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)tocurrentMetadata.plus(resetMetadata)when computing the new metadata during reset. Theminusoperation 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 vialog.makeAlert()instead of throwing aStreamException. This allows the task creation loop to continue processing other task groups after handling the reset.