FrankChen021 commented on code in PR #20092:
URL: https://github.com/apache/druid/pull/20092#discussion_r4133467259
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4367,101 +4371,36 @@ 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.
+ for (PartitionIdType partition : partitionsToReset.keySet()) {
+ final int groupId = getTaskGroupIdForPartition(partition);
+ final TaskGroup taskGroup = newTaskGroups.get(groupId);
+
+ if (taskGroup != null) {
+ newTaskGroups.put(groupId, buildNewTaskGroup(groupId,
metadataOffsets, partitionsToReset));
Review Comment:
[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.
--
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]
