FrankChen021 commented on code in PR #20092:
URL: https://github.com/apache/druid/pull/20092#discussion_r3821776160
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4446,13 +4446,10 @@ private void createNewTasks() throws
JsonProcessingException
// If any partitions need a reset, issue a single batch reset.
if (!partitionsToReset.isEmpty()) {
+ log.makeAlert(
+ "Previous sequenceNumbers are no longer available - automatically
resetting sequences"
+ ).addData("partitions", partitionsToReset).emit();
resetInternal(createDataSourceMetaDataForReset(ioConfig.getStream(),
partitionsToReset));
Review 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.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -2213,7 +2213,7 @@ public void resetInternal(DataSourceMetadata
dataSourceMetadata)
if (currentMetadata == null) {
metadataUpdateSuccess = true;
} else {
- final DataSourceMetadata newMetadata =
currentMetadata.minus(resetMetadata);
+ final DataSourceMetadata newMetadata =
currentMetadata.plus(resetMetadata);
Review 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.
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskRunner.java:
##########
@@ -87,14 +81,8 @@ protected
List<OrderedPartitionableRecord<KafkaTopicPartition, Long, KafkaRecord
return recordSupplier.poll(task.getIOConfig().getPollTimeout());
}
catch (OffsetOutOfRangeException e) {
- //
- // Handles OffsetOutOfRangeException, which is thrown if the seeked-to
- // offset is not present in the topic-partition. This can happen if
we're asking a task to read from data
- // that has not been written yet (which is totally legitimate). So let's
wait for it to show up
- //
log.warn("OffsetOutOfRangeException with message [%s]", e.getMessage());
- possiblyResetOffsetsOrWait(e.offsetOutOfRangePartitions(),
recordSupplier, toolbox);
- return Collections.emptyList();
+ throw e;
Review 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.
--
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]
