FrankChen021 commented on code in PR #20089:
URL: https://github.com/apache/druid/pull/20089#discussion_r4082450242
##########
server/src/main/java/org/apache/druid/client/BrokerServerView.java:
##########
@@ -351,26 +420,249 @@ private void serverRemovedSegment(DruidServerMetadata
server, DataSegment segmen
}
if (selector.isEmpty()) {
- VersionedIntervalTimeline<String, ServerSelector> timeline =
timelines.get(segment.getDataSource());
- selectors.remove(segmentId);
+ final long delayMillis =
segmentWatcherConfig.getSegmentDropDelayMillis();
+ if (delayMillis > 0) {
+ // Remove the segment from the timeline immediately so that queries
never observe an
+ // empty ServerSelector on the timeline (an empty selector makes
groupSegmentsByServer
+ // skip the holder while computeUncoveredIntervals still counts it
as covered, which
+ // yields a silent partial result). The selector is kept in the
selectors map so that
+ // a new server announcing the segment during the delay can re-add
it to the timeline.
+ // segmentRemoved is not fired until the selector is actually
cleaned up, because the
+ // segment may come back before the delay expires.
+ // See https://github.com/apache/druid/issues/18738
+ removeFromTimelineQuietly(segment, selector);
+
+ // Schedule the delayed cleanup of the selector from the selectors
map. If a new server
+ // announces the segment before the delay expires, the entry is
cancelled and the
+ // segment is re-added to the timeline.
+ //
+ // The callback closes over `entry`, which is created before
scheduling and therefore
+ // definitely assigned. Passing a *different* object to the
scheduled task would half-fix
+ // this: the entry identity is what lets the callback tell "I am
still the current entry"
+ // from "a newer entry replaced me", which a self-comparison of the
future cannot do.
+ final long scheduledAtNanos = System.nanoTime();
+ final EntryFuture entry = new EntryFuture();
+ entry.setDelayedRemovalFuture(
+ delayedRemovalExecutor.schedule(
+ () -> runDelayedRemoval(segment, selector, entry,
scheduledAtNanos),
+ delayMillis,
+ TimeUnit.MILLISECONDS
+ )
+ );
- final PartitionChunk<ServerSelector> removedPartition =
timeline.remove(
- segment.getInterval(), segment.getVersion(),
segment.getShardSpec().createChunk(selector)
- );
+ // Register the entry as the current pending removal. The timer may
already have fired
+ // (delayMillis can be shorter than this thread's scheduling
latency); that is safe
+ // because the callback re-validates the entry under `lock` and
bails out if it is not
+ // yet present, in which case the inline cleanup below performs the
removal instead.
+ final EntryFuture previous = pendingSegmentRemovals.put(segmentId,
entry);
+ if (previous != null) {
+ previous.cancel(false);
+ log.warn("Replaced existing pending removal for segment[%s].",
segmentId);
+ }
+
+ // Close the window where the timer fired before the put completed:
the callback did
+ // nothing in that case, so clean up inline under the same lock that
the callback uses.
+ if (entry.isDone()) {
+ cleanupSelectorIfCurrent(segmentId, selector, entry);
+ }
- if (removedPartition == null) {
- log.warn(
- "Asked to remove timeline entry[interval: %s, version: %s] that
doesn't exist",
- segment.getInterval(),
- segment.getVersion()
+ log.debug(
+ "Scheduled delayed cleanup of segment[%s] in [%d]ms. Waiting for
a new server to announce it.",
+ segmentId,
+ delayMillis
);
} else {
- runTimelineCallbacks(callback -> callback.segmentRemoved(segment));
+ removeSegmentFromTimeline(segment, selector);
}
}
}
}
+ /**
+ * Removes a segment's timeline entry without firing {@code segmentRemoved}
and without
+ * dropping the selector from {@link #selectors}. Used by the delayed-drop
path, where the
+ * segment may be re-announced before the delay expires.
+ */
+ private void removeFromTimelineQuietly(final DataSegment segment, final
ServerSelector selector)
+ {
+ final VersionedIntervalTimeline<String, ServerSelector> timeline =
timelines.get(segment.getDataSource());
+ if (timeline != null) {
+ timeline.remove(segment.getInterval(), segment.getVersion(),
segment.getShardSpec().createChunk(selector));
+ }
+ }
+
+ /**
+ * Body of the scheduled delayed-removal task. Runs on {@link
#delayedRemovalExecutor}.
+ *
+ * @param scheduledAtNanos {@link System#nanoTime()} sampled when the task
was scheduled; see
+ * the ordering note in {@link
#cleanupSelectorIfCurrent}.
+ */
+ private void runDelayedRemoval(
+ final DataSegment segment,
+ final ServerSelector selector,
+ final EntryFuture entry,
+ final long scheduledAtNanos
+ )
+ {
+ cleanupSelectorIfCurrent(segment.getId(), selector, entry);
Review Comment:
[P1] Do not emit removal after losing cleanup ownership
**Finding:** When the delayed task is canceled after it has started, or
loses the identity check because a reload or newer drop replaced its
EntryFuture, cleanupSelectorIfCurrent returns false. runDelayedRemoval discards
that result and still enqueues segmentRemoved at line 512. The load/drop race
can therefore evict metadata for a segment that was re-added (this callback is
the authority used by the metadata caches); even when cleanup wins, the enqueue
happens after releasing lock, so a re-add can enqueue segmentAdded first and
the same single-thread callback executor will process the stale removal after
it.
**Suggestion:** Only emit segmentRemoved when this entry actually completes
the cleanup, and serialize that successful cleanup plus callback submission
with the same lock/re-add transition so callback order cannot put a stale
removal after a re-add.
--
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]
