amaechler commented on code in PR #20092:
URL: https://github.com/apache/druid/pull/20092#discussion_r4260716815
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskRunner.java:
##########
@@ -87,75 +84,51 @@ protected
List<OrderedPartitionableRecord<KafkaTopicPartition, Long, KafkaRecord
return recordSupplier.poll(task.getIOConfig().getPollTimeout());
}
catch (OffsetOutOfRangeException e) {
+ // OffsetOutOfRangeException means one of two things:
//
- // 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
- //
+ // 1) the seeked-to offset is below the earliest available offset of the
partition. If
+ // resetOffsetAutomatically is set, re-throw so the supervisor can
reset the offsets. Note that
+ // we must not simply fail the task here: only the supervisor can
update the metadata, and it
+ // recognizes the reset from the exception.
+ // 2) the seeked-to offset is beyond the current end of the partition.
This is legitimate - a task
+ // can be asked to read data that has not been written yet - so wait
for it to show up. Kafka
+ // tasks use auto.offset.reset=none, so polling such an offset raises
this exception instead of
+ // silently repositioning the consumer.
log.warn("OffsetOutOfRangeException with message [%s]", e.getMessage());
- possiblyResetOffsetsOrWait(e.offsetOutOfRangePartitions(),
recordSupplier, toolbox);
- return Collections.emptyList();
- }
- }
- @Override
- protected SeekableStreamEndSequenceNumbers<KafkaTopicPartition, Long>
deserializePartitionsFromMetadata(
- ObjectMapper mapper,
- Object object
- )
- {
- return mapper.convertValue(object,
mapper.getTypeFactory().constructParametrizedType(
- SeekableStreamEndSequenceNumbers.class,
- SeekableStreamEndSequenceNumbers.class,
- KafkaTopicPartition.class,
- Long.class
- ));
- }
+ boolean belowEarliestOffset = false;
+ boolean futureOffset = false;
+ final String stream =
task.getIOConfig().getStartSequenceNumbers().getStream();
+ final boolean isMultiTopic = task.getIOConfig().isMultiTopic();
- private void possiblyResetOffsetsOrWait(
- Map<TopicPartition, Long> outOfRangePartitions,
- RecordSupplier<KafkaTopicPartition, Long, KafkaRecordEntity>
recordSupplier,
- TaskToolbox taskToolbox
- ) throws InterruptedException, IOException
- {
- final String stream =
task.getIOConfig().getStartSequenceNumbers().getStream();
- final boolean isMultiTopic = task.getIOConfig().isMultiTopic();
- final Map<TopicPartition, Long> resetPartitions = new HashMap<>();
- boolean doReset = false;
- if (task.getTuningConfig().isResetOffsetAutomatically()) {
- for (Map.Entry<TopicPartition, Long> outOfRangePartition :
outOfRangePartitions.entrySet()) {
- final TopicPartition topicPartition = outOfRangePartition.getKey();
- final long nextOffset = outOfRangePartition.getValue();
- // seek to the beginning to get the least available offset
- StreamPartition<KafkaTopicPartition> streamPartition =
StreamPartition.of(
+ for (Map.Entry<TopicPartition, Long> entry :
e.offsetOutOfRangePartitions().entrySet()) {
+ final TopicPartition topicPartition = entry.getKey();
+ final StreamPartition<KafkaTopicPartition> streamPartition =
StreamPartition.of(
stream,
new KafkaTopicPartition(isMultiTopic, topicPartition.topic(),
topicPartition.partition())
);
- final Long leastAvailableOffset =
recordSupplier.getEarliestSequenceNumber(streamPartition);
- if (leastAvailableOffset == null) {
- throw new ISE(
- "got null sequence number for partition[%s] when fetching from
kafka!",
- topicPartition.partition()
- );
- }
- // reset the seek
- recordSupplier.seek(streamPartition, nextOffset);
- // Reset consumer offset if resetOffsetAutomatically is set to true
- // and the current message offset in the kafka partition is more than
the
- // next message offset that we are trying to fetch
- if (leastAvailableOffset > nextOffset) {
- doReset = true;
- resetPartitions.put(topicPartition, nextOffset);
+ final Long earliestOffset =
recordSupplier.getEarliestSequenceNumber(streamPartition);
+ final Long latestOffset =
recordSupplier.getLatestSequenceNumber(streamPartition);
+
+ if (earliestOffset != null && earliestOffset > entry.getValue()) {
+ belowEarliestOffset = true;
+ } else if (latestOffset != null && latestOffset < entry.getValue()) {
+ // Not written yet. Waiting for it to appear is the correct
behavior; resetting would skip data.
+ futureOffset = true;
}
}
- }
- if (doReset) {
- sendResetRequestAndWait(CollectionUtils.mapKeys(resetPartitions,
topicPartition -> StreamPartition.of(
- stream,
- new KafkaTopicPartition(isMultiTopic, topicPartition.topic(),
topicPartition.partition())
- )), taskToolbox);
- } else {
+ if (belowEarliestOffset &&
task.getTuningConfig().isResetOffsetAutomatically()) {
+ throw e;
Review Comment:
With `replicas >= 2`, I think this can bring back the symptom #18282 set out
to remove ("new tasks are continuously forked but all fail"). When one replica
fails here while another is still running, `checkCurrentTaskState` only removes
the failed task, and `createNewTasks` tops the group up through
`createTasksForGroup` from the group's original `startingSequences`, which are
already below the earliest offset. Each replacement then fails the same way
until the surviving replica also fails or reaches `taskDuration`.
I reproduced this in a `KafkaSupervisorTest` on this branch: with
`replicas=2`, after `deleteRecords` moved partition 0's earliest offset to 2,
four consecutive replacements all started at offset 0, no reset was issued, and
the supervisor reported `UNHEALTHY_TASKS` from the third failure. On master,
the failing replica sends a `ResetDataSourceMetadataAction` instead, which
avoids the churn but also kills the healthy replica.
Options I can see:
1. Skip replica top-up while the group's starting offsets are unavailable,
next to the bounded-mode guard in the same loop. The surviving replica keeps
running with its data. If it fails, the group empties and goes through the
existing reset path; if it finishes first, the next group starts from its end
offsets. A local sketch of this stopped the churn, and `KafkaSupervisorTest`
still passes:
```java
if (!taskGroup.tasks.isEmpty()
&& taskTuningConfig.isResetOffsetAutomatically()
&& !taskTuningConfig.isSkipSequenceNumberAvailabilityCheck()
&& !areStartingSequencesAvailable(taskGroup)) { //
checkOffsetAvailability() over taskGroup.startingSequences
log.warn("Not creating replacement tasks for taskGroup[%d]: its starting
offsets are no longer available.", groupId);
continue;
}
```
2. Send the group through the existing batch reset. Recovers sooner, but
drops the healthy replica's unpublished data, as master does.
3. Keep the current behaviour and document it.
I'd lean towards 1. Separately, the comment above says the supervisor
"recognizes the reset from the exception", but the supervisor only sees a
failed task and finds the stale offset through its own availability check when
it rebuilds the group. Could the comment say that instead?
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskRunner.java:
##########
@@ -87,75 +84,51 @@ protected
List<OrderedPartitionableRecord<KafkaTopicPartition, Long, KafkaRecord
return recordSupplier.poll(task.getIOConfig().getPollTimeout());
}
catch (OffsetOutOfRangeException e) {
+ // OffsetOutOfRangeException means one of two things:
//
- // 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
- //
+ // 1) the seeked-to offset is below the earliest available offset of the
partition. If
+ // resetOffsetAutomatically is set, re-throw so the supervisor can
reset the offsets. Note that
+ // we must not simply fail the task here: only the supervisor can
update the metadata, and it
+ // recognizes the reset from the exception.
+ // 2) the seeked-to offset is beyond the current end of the partition.
This is legitimate - a task
+ // can be asked to read data that has not been written yet - so wait
for it to show up. Kafka
+ // tasks use auto.offset.reset=none, so polling such an offset raises
this exception instead of
+ // silently repositioning the consumer.
log.warn("OffsetOutOfRangeException with message [%s]", e.getMessage());
- possiblyResetOffsetsOrWait(e.offsetOutOfRangePartitions(),
recordSupplier, toolbox);
- return Collections.emptyList();
- }
- }
- @Override
- protected SeekableStreamEndSequenceNumbers<KafkaTopicPartition, Long>
deserializePartitionsFromMetadata(
- ObjectMapper mapper,
- Object object
- )
- {
- return mapper.convertValue(object,
mapper.getTypeFactory().constructParametrizedType(
- SeekableStreamEndSequenceNumbers.class,
- SeekableStreamEndSequenceNumbers.class,
- KafkaTopicPartition.class,
- Long.class
- ));
- }
+ boolean belowEarliestOffset = false;
+ boolean futureOffset = false;
+ final String stream =
task.getIOConfig().getStartSequenceNumbers().getStream();
+ final boolean isMultiTopic = task.getIOConfig().isMultiTopic();
- private void possiblyResetOffsetsOrWait(
- Map<TopicPartition, Long> outOfRangePartitions,
- RecordSupplier<KafkaTopicPartition, Long, KafkaRecordEntity>
recordSupplier,
- TaskToolbox taskToolbox
- ) throws InterruptedException, IOException
- {
- final String stream =
task.getIOConfig().getStartSequenceNumbers().getStream();
- final boolean isMultiTopic = task.getIOConfig().isMultiTopic();
- final Map<TopicPartition, Long> resetPartitions = new HashMap<>();
- boolean doReset = false;
- if (task.getTuningConfig().isResetOffsetAutomatically()) {
- for (Map.Entry<TopicPartition, Long> outOfRangePartition :
outOfRangePartitions.entrySet()) {
- final TopicPartition topicPartition = outOfRangePartition.getKey();
- final long nextOffset = outOfRangePartition.getValue();
- // seek to the beginning to get the least available offset
- StreamPartition<KafkaTopicPartition> streamPartition =
StreamPartition.of(
+ for (Map.Entry<TopicPartition, Long> entry :
e.offsetOutOfRangePartitions().entrySet()) {
+ final TopicPartition topicPartition = entry.getKey();
+ final StreamPartition<KafkaTopicPartition> streamPartition =
StreamPartition.of(
stream,
new KafkaTopicPartition(isMultiTopic, topicPartition.topic(),
topicPartition.partition())
);
- final Long leastAvailableOffset =
recordSupplier.getEarliestSequenceNumber(streamPartition);
- if (leastAvailableOffset == null) {
- throw new ISE(
- "got null sequence number for partition[%s] when fetching from
kafka!",
- topicPartition.partition()
- );
- }
- // reset the seek
- recordSupplier.seek(streamPartition, nextOffset);
- // Reset consumer offset if resetOffsetAutomatically is set to true
- // and the current message offset in the kafka partition is more than
the
- // next message offset that we are trying to fetch
- if (leastAvailableOffset > nextOffset) {
- doReset = true;
- resetPartitions.put(topicPartition, nextOffset);
+ final Long earliestOffset =
recordSupplier.getEarliestSequenceNumber(streamPartition);
+ final Long latestOffset =
recordSupplier.getLatestSequenceNumber(streamPartition);
Review Comment:
Suggestion: I think the future-offset classification can go. Both branches
end in the same wait and retry, so `futureOffset` only decides whether the "not
yet available" line is logged, and it costs a seek round trip to the broker per
partition on every retry. What's left would be: if `resetOffsetAutomatically`
is set and any partition is below the earliest offset, rethrow; otherwise wait
and retry as before.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4367,101 +4377,46 @@ private void createNewTasks() throws
JsonProcessingException
}
log.info("Creating new taskGroup[%d] for partitions[%s].", groupId,
partitionGroups.get(groupId));
- final DateTime minimumMessageTime;
- if (ioConfig.getLateMessageRejectionStartDateTime().isPresent()) {
- minimumMessageTime =
ioConfig.getLateMessageRejectionStartDateTime().get();
- } else {
- minimumMessageTime =
ioConfig.getLateMessageRejectionPeriod().isPresent()
- ?
DateTimes.nowUtc().minus(ioConfig.getLateMessageRejectionPeriod().get())
- : null;
- }
-
- final DateTime maximumMessageTime =
ioConfig.getEarlyMessageRejectionPeriod().isPresent()
- ? DateTimes.nowUtc()
-
.plus(ioConfig.getTaskDuration())
-
.plus(ioConfig.getEarlyMessageRejectionPeriod().get())
- : null;
-
- final Map<PartitionIdType, OrderedSequenceNumber<SequenceOffsetType>>
unfilteredStartingOffsets =
- generateStartingSequencesForPartitionGroup(groupId,
metadataOffsets, partitionsToReset);
-
- final Map<PartitionIdType, OrderedSequenceNumber<SequenceOffsetType>>
startingOffsets;
- if (supportsPartitionExpiration()) {
- startingOffsets =
filterExpiredPartitionsFromStartingOffsets(unfilteredStartingOffsets);
- } else {
- startingOffsets = unfilteredStartingOffsets;
- }
-
- ImmutableMap<PartitionIdType, SequenceOffsetType>
simpleStartingOffsets = startingOffsets
- .entrySet()
- .stream()
- .filter(entry -> entry.getValue().get() != null)
- .collect(
- Collectors.collectingAndThen(
- Collectors.toMap(Entry::getKey, entry ->
entry.getValue().get()),
- ImmutableMap::copyOf
- )
- );
-
- ImmutableMap<PartitionIdType, SequenceOffsetType>
simpleUnfilteredStartingOffsets;
- if (supportsPartitionExpiration()) {
- simpleUnfilteredStartingOffsets = unfilteredStartingOffsets
- .entrySet()
- .stream()
- .filter(entry -> entry.getValue().get() != null)
- .collect(
- Collectors.collectingAndThen(
- Collectors.toMap(Entry::getKey, entry ->
entry.getValue().get()),
- ImmutableMap::copyOf
- )
- );
- } else {
- simpleUnfilteredStartingOffsets = simpleStartingOffsets;
- }
-
- Set<PartitionIdType> exclusiveStartSequenceNumberPartitions;
- if (!useExclusiveStartingSequence) {
- exclusiveStartSequenceNumberPartitions = Collections.emptySet();
- } else {
- exclusiveStartSequenceNumberPartitions = startingOffsets
- .entrySet()
- .stream()
- .filter(x -> x.getValue().get() != null
- && x.getValue().isExclusive())
- .map(Entry::getKey)
- .collect(Collectors.toSet());
- }
-
- // NEW: Extract end offsets for bounded mode
- ImmutableMap<PartitionIdType, SequenceOffsetType> endOffsets = null;
- if (ioConfig.isBounded()) {
- endOffsets = ImmutableMap.copyOf(getEndOffsetsForGroup(groupId));
- }
-
- log.info(
- "Initializing taskGroup[%d] with startingOffsets[%s] and
endOffsets[%s]",
- groupId,
- simpleStartingOffsets,
- endOffsets
- );
- newTaskGroups.put(
- groupId,
- new TaskGroup(
- groupId,
- simpleStartingOffsets,
- simpleUnfilteredStartingOffsets,
- endOffsets,
- minimumMessageTime,
- maximumMessageTime,
- exclusiveStartSequenceNumberPartitions
- )
- );
+ newTaskGroups.put(groupId, buildNewTaskGroup(groupId, metadataOffsets,
partitionsToReset));
}
}
// If any partitions need a reset, issue a single batch reset.
if (!partitionsToReset.isEmpty()) {
resetInternal(createDataSourceMetaDataForReset(ioConfig.getStream(),
partitionsToReset));
+
+ // The new task groups were built before the reset, so the ones that
cover a partition which has to
+ // be reset hold no offset for it: 'getOffsetFromStorageForPartition'
left the partition out rather
+ // than taking it from metadata storage. Installing them as they are
would leave a mixed group
+ // reading the stale partition from its old position until the next
rollover, and would make a group
+ // that consists only of reset partitions run no task at all. Rebuild
the affected groups now that
+ // partitionOffsets has been reset, so that each of them starts from the
reset position.
+ // Collect the affected groupIds first: rebuilding a group mutates
'partitionsToReset', so iterating
+ // its keys directly would risk a ConcurrentModificationException, and
several reset partitions can
+ // map to the same group.
+ final Set<Integer> groupsToRebuild = new HashSet<>();
+ for (PartitionIdType partition : partitionsToReset.keySet()) {
+ groupsToRebuild.add(getTaskGroupIdForPartition(partition));
+ }
+
+ for (Integer groupId : groupsToRebuild) {
Review Comment:
I think this rebuild can go, along with the rest of the changes in this
file. `resetInternal` calls `killTaskGroupForPartitions`, which removes the
group from `partitionGroups`, and then puts it back as an empty set through
`computeIfAbsent`. By the time this loop runs, every affected group has no
partitions, so `buildNewTaskGroup` returns an empty `TaskGroup`, which the
`StreamException` below discards. The earlier concern about installing
incomplete groups was already resolved when 1e73792e restored the throw.
The rebuild also changes what
`KafkaSupervisorTest#testGetOffsetFromStorageForPartitionWithResetOffsetAutomatically`
exercises. The test's `generateSequenceName` override calls
`startPartitions.keySet().iterator().next()`, so on this branch the run ends
with a `NoSuchElementException` from the rebuild rather than the
`StreamException`. The test still passes because it only checks the alert
prefix; asserting the exception class through `getExceptionEvents()`, like the
test around line 1358, would catch that.
Master already stages the new groups, issues one batch reset and throws, so
I'd suggest reverting `SeekableStreamSupervisor.java` to master: the rebuild,
the `dataSourceMetadataFetchedInCreateNewTasks` field, the extra overloads and
the `buildNewTaskGroup` extraction.
--
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]
