FrankChen021 commented on code in PR #20245:
URL: https://github.com/apache/druid/pull/20245#discussion_r3924863855
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java:
##########
@@ -672,6 +679,13 @@ private int updateReplicasInTier(
cancelOperations(SegmentAction.MOVE_FROM, movingReplicas, segment,
segmentStatus);
}
+ // If segment is unavailable, prioritize load by changing REPLICATE
actions to LOAD
+ if (shouldPrioritizeLoadOfUnavailableSegment) {
+ for (ServerHolder server :
segmentStatus.getServersPerforming(SegmentAction.REPLICATE)) {
Review Comment:
[P1] Rebuild status after reprioritization
segmentStatus is built before this loop converts REPLICATE actions to LOAD.
If a tier has a queued replicate surplus, the later surplus branch still looks
in the snapshot's REPLICATE list while the LOAD list is empty, so it can cancel
neither action and leave excess loads queued. Rebuild the status after
reprioritization or reconcile the affected servers directly before surplus
cancellation.
##########
server/src/main/java/org/apache/druid/server/coordinator/ServerHolder.java:
##########
@@ -167,7 +167,7 @@ private void initializeQueuedSegments(
}
final SegmentAction action = holder.getAction();
- addToQueuedSegments(holder.getSegment(), simplify(action));
+ addToQueuedSegments(holder.getSegment(), action);
Review Comment:
[P2] Include REPLICATE in loading segments
By preserving REPLICATE here, getLoadingSegments() still filters for exact
LOAD at line 353. TierSegmentBalancer uses that accessor to prioritize cheap
moves, so queued replica loads are omitted from the prioritized set. Make the
accessor use action.isLoad() or otherwise include REPLICATE.
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java:
##########
@@ -1003,6 +1021,25 @@ private int loadReplicas(
return numLoadsQueued;
}
+ /**
+ * Tries to increase the load priority of the given unavailable segment (by
+ * changing the action from {@link SegmentAction#REPLICATE} to {@link
SegmentAction#LOAD})
+ * if it is already present in the queue of the server.
+ *
+ * @return true only if the priority was increased successfully.
+ */
+ private boolean prioritizeLoadOfUnavailableSegment(
+ DataSegment segment,
+ ServerHolder server,
+ @Nullable PartialLoadProfile profile
+ )
+ {
+ return server.getActionOnSegment(segment) == SegmentAction.REPLICATE
+ && Objects.equals(profile, server.getProjectedProfile(segment))
+ && server.cancelOperation(SegmentAction.REPLICATE, segment)
+ && loadQueueManager.loadSegment(segment, server,
SegmentAction.LOAD, profile);
Review Comment:
[P2] Preserve queue budget when replacing REPLICATE
prioritizeLoadOfUnavailableSegment first cancels REPLICATE, then queues the
same segment as LOAD through startOperation. startOperation increments
totalAssignmentsInRun for every load, while cancellation does not decrement it;
replacing one queue slot therefore consumes an extra per-run budget and can
make isLoadQueueFull() reject unrelated loads. Preserve the replacement's
existing slot and accounting.
--
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]
