aho135 commented on code in PR #19372:
URL: https://github.com/apache/druid/pull/19372#discussion_r3164941305
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -4347,6 +4478,179 @@ && checkSourceMetadataMatch(dataSourceMetadata)) {
return Collections.emptyMap();
}
+ /**
+ * Check if all partitions in a task group have reached their bounded end
offsets.
+ * Used to determine if the task group completed successfully vs failed
midway.
+ *
+ * @param groupId The task group ID to check
+ * @return true if all partitions in the group have reached their end
offsets, false otherwise
+ */
+ private boolean hasTaskGroupReachedBoundedEnd(int groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+ Map<PartitionIdType, SequenceOffsetType> currentOffsets =
getOffsetsFromMetadataStorage();
+
+ log.info(
+ "Bounded mode: checking completion for taskGroup[%d]. Current offsets
from metadata: %s, End offsets: %s",
+ groupId,
+ currentOffsets,
+ endOffsets
+ );
+
+ if (currentOffsets == null || currentOffsets.isEmpty()) {
+ log.debug("No checkpointed offsets found, taskGroup[%d] has not
completed", groupId);
+ return false; // No progress yet, task hasn't completed
+ }
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null || partitionsInGroup.isEmpty()) {
+ return false;
+ }
+
+ // Check if ALL partitions in this group have reached their end offsets
+ for (PartitionIdType partition : partitionsInGroup) {
+ SequenceOffsetType endOffset = endOffsets.get(partition);
+ SequenceOffsetType currentOffset = currentOffsets.get(partition);
+
+ if (currentOffset == null) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] has no checkpointed offset, not
complete",
+ partition,
+ groupId
+ );
+ return false; // Partition hasn't started processing
+ }
+
+ if (!isOffsetAtOrBeyond(currentOffset, endOffset)) {
+ log.debug(
+ "Partition[%s] in taskGroup[%d] at offset[%s], has not reached
end[%s]",
+ partition,
+ groupId,
+ currentOffset,
+ endOffset
+ );
+ return false; // This partition hasn't reached its end
+ }
+ }
+
+ log.info(
+ "All partitions in taskGroup[%d] have reached their end offsets",
+ groupId
+ );
+ return true; // All partitions have reached their end offsets
+ }
+
+ /**
+ * Get current offsets for all partitions in a task group from metadata
storage.
+ */
+ private Map<PartitionIdType, SequenceOffsetType>
getCurrentOffsetsForGroup(int groupId)
+ {
+ Map<PartitionIdType, SequenceOffsetType> allOffsets =
getOffsetsFromMetadataStorage();
+ if (allOffsets == null || allOffsets.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+ if (partitionsInGroup == null) {
+ return Collections.emptyMap();
+ }
+
+ return partitionsInGroup.stream()
+ .filter(allOffsets::containsKey)
+ .collect(Collectors.toMap(
+ p -> p,
+ allOffsets::get
+ ));
+ }
+
+ /**
+ * Get end offsets for all partitions in a task group from bounded config.
+ */
+ private Map<PartitionIdType, SequenceOffsetType> getEndOffsetsForGroup(int
groupId)
+ {
+ BoundedStreamConfig boundedConfig = ioConfig.getBoundedStreamConfig();
+ Map<PartitionIdType, SequenceOffsetType> endOffsets =
+ convertBoundedConfigMap(boundedConfig.getEndSequenceNumbers());
+ Set<PartitionIdType> partitionsInGroup = partitionGroups.get(groupId);
+
+ if (partitionsInGroup == null) {
+ return Collections.emptyMap();
+ }
+
+ return partitionsInGroup.stream()
+ .filter(endOffsets::containsKey)
+ .collect(Collectors.toMap(
+ p -> p,
+ endOffsets::get
+ ));
+ }
+
+ /**
+ * Check if all bounded tasks have completed.
+ * Called after createNewTasks() in runInternal to ensure tasks have been
created first.
+ *
+ * For bounded supervisors, we determine completion by checking if new tasks
would be created.
+ * In createNewTasks(), bounded mode checks hasTaskGroupReachedBoundedEnd()
before creating tasks.
+ * If that returns true (offsets reached), no new tasks are created.
+ *
+ * So completion is: no active tasks, no pending tasks, and createNewTasks()
chose not to create any.
+ * This is indicated by empty task groups after createNewTasks() has run.
+ *
+ * We do NOT separately check metadata storage here because:
+ * 1. Metadata may contain stale offsets from previous supervisor runs
+ * 2. createNewTasks() already does the offset checking logic
+ * 3. If tasks were killed/failed and work is incomplete, createNewTasks()
will recreate them
+ *
+ * @return true if all bounded work is complete, false otherwise
Review Comment:
Ahh, let me update the javadoc to be more clear. The implementation does
take that scenario into account. Checking active/pending task group is just a
cheap check to avoid hitting the metadata store every run. But if both are
empty it will check the offsets in the metadata store as the final validation
that it is complete
--
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]
