kfaraz commented on code in PR #19651:
URL: https://github.com/apache/druid/pull/19651#discussion_r3550229791
##########
server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java:
##########
@@ -1181,12 +1181,46 @@ private void unannounceSegment(DataSegment segment)
}
}
- public void registerUpgradedPendingSegment(PendingSegmentRecord
pendingSegmentRecord) throws IOException
+ /**
+ * Outcome of a {@link #registerUpgradedPendingSegment} call, returned so
the caller (which owns the emitter) can
+ * emit the corresponding metric.
+ */
+ public enum UpgradeAnnouncementOutcome
+ {
+ ANNOUNCED,
+ SKIPPED_UNKNOWN_BASE,
+ SKIPPED_NO_SINK,
+ SKIPPED_DROPPING
+ }
+
+ public UpgradeAnnouncementOutcome
registerUpgradedPendingSegment(PendingSegmentRecord pendingSegmentRecord)
throws IOException
{
SegmentIdWithShardSpec basePendingSegment =
idToPendingSegment.get(pendingSegmentRecord.getUpgradedFromSegmentId());
SegmentIdWithShardSpec upgradedPendingSegment =
pendingSegmentRecord.getId();
- if (!sinks.containsKey(basePendingSegment) ||
droppingSinks.contains(basePendingSegment)) {
- return;
+ if (basePendingSegment == null ||
droppingSinks.contains(basePendingSegment) ||
!sinks.containsKey(basePendingSegment)) {
+ if (basePendingSegment == null) {
+ // This task never allocated a segment matching upgradedFromSegmentId,
i.e. the request targeted the wrong task.
+ log.info(
+ "Not announcing upgraded pending segment[%s] because this task has
no base sink matching"
+ + " upgradedFromSegmentId[%s]; the upgrade request likely targeted
the wrong task[%s].",
+ upgradedPendingSegment,
pendingSegmentRecord.getUpgradedFromSegmentId(), myId
+ );
+ return UpgradeAnnouncementOutcome.SKIPPED_UNKNOWN_BASE;
+ } else if (droppingSinks.contains(basePendingSegment)) {
+ // Expected during handoff: the base sink is being dropped
+ log.debug(
+ "Not announcing upgraded pending segment[%s] for base segment[%s]
on task[%s] because the base sink is being dropped.",
+ upgradedPendingSegment, basePendingSegment, myId
+ );
+ return UpgradeAnnouncementOutcome.SKIPPED_DROPPING;
+ } else {
+ // Unexpected: the base sink is gone even though this task once held
it.
+ log.info(
+ "Not announcing upgraded pending segment[%s] for base segment[%s]
on task[%s] because the base sink is no longer present.",
+ upgradedPendingSegment, basePendingSegment, myId
Review Comment:
I feel like these logs can be merged into one and can be logged by the
caller itself using the result of this method.
--
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]
