capistrant commented on code in PR #19651:
URL: https://github.com/apache/druid/pull/19651#discussion_r3559734359
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1415,26 +1417,102 @@ public void resetOffsets(@Nonnull DataSourceMetadata
resetDataSourceMetadata)
addNotice(new ResetOffsetsNotice(resetDataSourceMetadata));
}
- public void registerNewVersionOfPendingSegment(
+ /**
+ * Notifies every running task in the matching task group(s) of an upgraded
pending segment.
+ *
+ * @return the number of tasks notified, which the caller aggregates into a
per-batch summary
+ */
+ public int registerNewVersionOfPendingSegment(
PendingSegmentRecord pendingSegmentRecord
)
{
+ final String taskAllocatorId = pendingSegmentRecord.getTaskAllocatorId();
+ int notifiedTasks = 0;
+
for (TaskGroup taskGroup : activelyReadingTaskGroups.values()) {
- if
(taskGroup.baseSequenceName.equals(pendingSegmentRecord.getTaskAllocatorId())) {
+ if (taskGroup.baseSequenceName.equals(taskAllocatorId)) {
for (String taskId : taskGroup.taskIds()) {
- taskClient.registerNewVersionOfPendingSegmentAsync(taskId,
pendingSegmentRecord);
+ notifyTaskOfUpgradedPendingSegment(taskId, pendingSegmentRecord);
+ notifiedTasks++;
}
}
}
for (List<TaskGroup> taskGroupList : pendingCompletionTaskGroups.values())
{
for (TaskGroup taskGroup : taskGroupList) {
- if
(taskGroup.baseSequenceName.equals(pendingSegmentRecord.getTaskAllocatorId())) {
+ if (taskGroup.baseSequenceName.equals(taskAllocatorId)) {
for (String taskId : taskGroup.taskIds()) {
- taskClient.registerNewVersionOfPendingSegmentAsync(taskId,
pendingSegmentRecord);
+ notifyTaskOfUpgradedPendingSegment(taskId, pendingSegmentRecord);
+ notifiedTasks++;
}
}
}
}
+
+ // Only the exceptional no-match case is logged per segment; the notified
count is returned to the task action,
+ // which summarizes the whole batch in one log line, and each notification
is captured by the per-task metric.
+ if (notifiedTasks == 0) {
+ // No running task matched: the segment will not be re-announced until
handoff. This is a potential silent-loss
+ // window where data will not be queryable until handoff.
+ log.warn(
+ "Could not find any task matching taskAllocatorId[%s] in
supervisor[%s] for upgraded pending segment[%s]"
+ + " (upgradedFrom[%s]); it will not be re-announced until handoff.
Currently tracking [%d] activelyReading"
+ + " and [%d] pendingCompletion task group(s).",
Review Comment:
hmm I guess probably not useful. was kinda just packing in bits of
information wherever I could find them in this PR. but you're right I'm not
sure what I do with this number in this log so I will remove
--
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]
