capistrant commented on code in PR #19651:
URL: https://github.com/apache/druid/pull/19651#discussion_r3556190630
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1419,22 +1421,89 @@
PendingSegmentRecord pendingSegmentRecord
)
{
+ final String taskAllocatorId = pendingSegmentRecord.getTaskAllocatorId();
+ int matchedGroups = 0;
+ int notifiedTasks = 0;
+
for (TaskGroup taskGroup : activelyReadingTaskGroups.values()) {
- if
(taskGroup.baseSequenceName.equals(pendingSegmentRecord.getTaskAllocatorId())) {
+ if (taskGroup.baseSequenceName.equals(taskAllocatorId)) {
+ matchedGroups++;
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)) {
+ matchedGroups++;
for (String taskId : taskGroup.taskIds()) {
- taskClient.registerNewVersionOfPendingSegmentAsync(taskId,
pendingSegmentRecord);
+ notifyTaskOfUpgradedPendingSegment(taskId, pendingSegmentRecord);
+ notifiedTasks++;
}
}
}
}
+
+ 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]) "
+ + " supervisor[%s]; it will not be re-announced until handoff.
Currently tracking [%d] activelyReading"
+ + " and [%d] pendingCompletion task group(s).",
+ pendingSegmentRecord.getId(),
+ pendingSegmentRecord.getUpgradedFromSegmentId(),
+ taskAllocatorId,
+ supervisorId,
+ activelyReadingTaskGroups.size(),
+ pendingCompletionTaskGroups.size()
+ );
Review Comment:
whoops
--
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]
