FrankChen021 commented on code in PR #20089:
URL: https://github.com/apache/druid/pull/20089#discussion_r3836152675
##########
server/src/main/java/org/apache/druid/client/BrokerServerView.java:
##########
@@ -351,26 +410,123 @@ 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 to prevent
queries from seeing
+ // an empty ServerSelector during the delay window, which would
cause partial results.
+ // 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.
+ // We only remove from the timeline, not from the selectors map, and
we do not fire
+ // the segmentRemoved callback since the segment may come back
before the delay expires.
+ // See https://github.com/apache/druid/issues/18738
+ final VersionedIntervalTimeline<String, ServerSelector> timeline =
timelines.get(segment.getDataSource());
+ if (timeline != null) {
+ timeline.remove(
+ segment.getInterval(), segment.getVersion(),
segment.getShardSpec().createChunk(selector)
+ );
+ }
- final PartitionChunk<ServerSelector> removedPartition =
timeline.remove(
- segment.getInterval(), segment.getVersion(),
segment.getShardSpec().createChunk(selector)
- );
+ // Schedule a delayed cleanup of the selector from the selectors
map. If a new
+ // server announces the segment before the delay expires, the
removal is cancelled
+ // and the segment is re-added to the timeline, preventing the
segment load/drop
+ // race condition.
+ final ScheduledFuture<?> pendingRemoval =
delayedRemovalExecutor.schedule(
+ () -> {
+ synchronized (lock) {
+ // Re-check under lock that this future is still the current
pending
+ // removal. The remove(key, value) call atomically verifies
the future
+ // is current, and the lock ensures no new server can add
the segment
+ // between the check and the selector cleanup.
+ if (pendingSegmentRemovals.remove(segmentId,
pendingRemoval)) {
Review Comment:
[P1] Self-referential future capture does not compile
The scheduled lambda references `pendingRemoval` while that final local is
still being initialized. Java definite-assignment rules reject this with
“variable pendingRemoval might not have been initialized,” blocking compilation.
##########
server/src/main/java/org/apache/druid/client/BrokerServerView.java:
##########
@@ -351,26 +410,123 @@ 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 to prevent
queries from seeing
+ // an empty ServerSelector during the delay window, which would
cause partial results.
+ // 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.
+ // We only remove from the timeline, not from the selectors map, and
we do not fire
+ // the segmentRemoved callback since the segment may come back
before the delay expires.
+ // See https://github.com/apache/druid/issues/18738
+ final VersionedIntervalTimeline<String, ServerSelector> timeline =
timelines.get(segment.getDataSource());
+ if (timeline != null) {
+ timeline.remove(
+ segment.getInterval(), segment.getVersion(),
segment.getShardSpec().createChunk(selector)
+ );
+ }
- final PartitionChunk<ServerSelector> removedPartition =
timeline.remove(
- segment.getInterval(), segment.getVersion(),
segment.getShardSpec().createChunk(selector)
- );
+ // Schedule a delayed cleanup of the selector from the selectors
map. If a new
+ // server announces the segment before the delay expires, the
removal is cancelled
+ // and the segment is re-added to the timeline, preventing the
segment load/drop
+ // race condition.
+ final ScheduledFuture<?> pendingRemoval =
delayedRemovalExecutor.schedule(
+ () -> {
+ synchronized (lock) {
+ // Re-check under lock that this future is still the current
pending
+ // removal. The remove(key, value) call atomically verifies
the future
+ // is current, and the lock ensures no new server can add
the segment
+ // between the check and the selector cleanup.
+ if (pendingSegmentRemovals.remove(segmentId,
pendingRemoval)) {
+ // Double-check the selector is still empty and unchanged
+ final ServerSelector currentSelector =
selectors.get(segmentId);
+ if (currentSelector == selector &&
currentSelector.isEmpty()) {
+ selectors.remove(segmentId);
Review Comment:
[P1] Delayed cleanup omits segmentRemoved callback
For positive delays, the timer removes only the selector from `selectors`
and never calls `runTimelineCallbacks(...segmentRemoved...)`.
`TimelineServerView` defines that callback as the authority for removal, and
metadata caches depend on it, so pure drops can leave stale segment metadata.
--
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]
