FrankChen021 commented on code in PR #20089:
URL: https://github.com/apache/druid/pull/20089#discussion_r3830375991
##########
server/src/main/java/org/apache/druid/client/BrokerServerView.java:
##########
@@ -351,26 +394,81 @@ private void serverRemovedSegment(DruidServerMetadata
server, DataSegment segmen
}
if (selector.isEmpty()) {
- VersionedIntervalTimeline<String, ServerSelector> timeline =
timelines.get(segment.getDataSource());
- selectors.remove(segmentId);
-
- final PartitionChunk<ServerSelector> removedPartition =
timeline.remove(
- segment.getInterval(), segment.getVersion(),
segment.getShardSpec().createChunk(selector)
- );
-
- if (removedPartition == null) {
- log.warn(
- "Asked to remove timeline entry[interval: %s, version: %s] that
doesn't exist",
- segment.getInterval(),
- segment.getVersion()
+ final long delayMillis =
segmentWatcherConfig.getSegmentDropDelayMillis();
Review Comment:
[P1] Delayed drops expose empty selectors to queries
During the delay, the empty ServerSelector remains in the timeline.
CachingClusteredClient can include that holder, pick() returns null, and
groupSegmentsByServer omits the segment while computeUncoveredIntervals still
sees the timeline holder, allowing a partial result without an uncovered
interval. Remove or mark the holder unavailable during the delay, or make
query-side uncovered-interval handling treat an empty selector as uncovered.
##########
server/src/main/java/org/apache/druid/client/BrokerServerView.java:
##########
@@ -351,26 +394,81 @@ private void serverRemovedSegment(DruidServerMetadata
server, DataSegment segmen
}
if (selector.isEmpty()) {
- VersionedIntervalTimeline<String, ServerSelector> timeline =
timelines.get(segment.getDataSource());
- selectors.remove(segmentId);
-
- final PartitionChunk<ServerSelector> removedPartition =
timeline.remove(
- segment.getInterval(), segment.getVersion(),
segment.getShardSpec().createChunk(selector)
- );
-
- if (removedPartition == null) {
- log.warn(
- "Asked to remove timeline entry[interval: %s, version: %s] that
doesn't exist",
- segment.getInterval(),
- segment.getVersion()
+ final long delayMillis =
segmentWatcherConfig.getSegmentDropDelayMillis();
+ if (delayMillis > 0) {
+ // Schedule a delayed removal to prevent the segment load/drop race
condition.
+ // When a segment is moved from one historical to another, the
broker may receive
+ // the drop callback from the old server before the load callback
from the new one.
+ // By delaying the timeline removal, we give the new server time to
announce the
+ // segment, preventing the segment from temporarily disappearing
from the timeline.
+ // See https://github.com/apache/druid/issues/18738
+ final ScheduledFuture<?> pendingRemoval =
delayedRemovalExecutor.schedule(
+ () -> {
+ if (pendingSegmentRemovals.remove(segmentId, pendingRemoval)) {
Review Comment:
[P1] Stale timer is not rechecked under lock
The callback removes its future from pendingSegmentRemovals before taking
the timeline lock. A segment can be re-added in that gap and install a newer
delayed removal; the old callback can then remove the segment while the newer
callback sees a pending entry and returns, leaving the freshly re-added segment
absent from the timeline. Recheck that the callback still owns the current
pending future under the same lock, or make replacement and removal atomic.
##########
server/src/main/java/org/apache/druid/client/BrokerServerView.java:
##########
@@ -351,26 +394,81 @@ private void serverRemovedSegment(DruidServerMetadata
server, DataSegment segmen
}
if (selector.isEmpty()) {
- VersionedIntervalTimeline<String, ServerSelector> timeline =
timelines.get(segment.getDataSource());
- selectors.remove(segmentId);
-
- final PartitionChunk<ServerSelector> removedPartition =
timeline.remove(
- segment.getInterval(), segment.getVersion(),
segment.getShardSpec().createChunk(selector)
- );
-
- if (removedPartition == null) {
- log.warn(
- "Asked to remove timeline entry[interval: %s, version: %s] that
doesn't exist",
- segment.getInterval(),
- segment.getVersion()
+ final long delayMillis =
segmentWatcherConfig.getSegmentDropDelayMillis();
+ if (delayMillis > 0) {
+ // Schedule a delayed removal to prevent the segment load/drop race
condition.
+ // When a segment is moved from one historical to another, the
broker may receive
+ // the drop callback from the old server before the load callback
from the new one.
+ // By delaying the timeline removal, we give the new server time to
announce the
+ // segment, preventing the segment from temporarily disappearing
from the timeline.
+ // See https://github.com/apache/druid/issues/18738
+ final ScheduledFuture<?> pendingRemoval =
delayedRemovalExecutor.schedule(
+ () -> {
+ if (pendingSegmentRemovals.remove(segmentId, pendingRemoval)) {
+ removeSegmentFromTimeline(segment, selector);
+ }
+ },
+ delayMillis,
+ TimeUnit.MILLISECONDS
+ );
+ final ScheduledFuture<?> previous =
pendingSegmentRemovals.put(segmentId, pendingRemoval);
Review Comment:
[P2] Timer can fire before pending state is published
The scheduled callback can run before pendingSegmentRemovals.put completes.
It then finds no pending entry and returns, after which the future is published
with no callback left to remove the segment from the timeline. Publish the
pending state before scheduling, or synchronize publication with callback
execution.
--
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]
