FrankChen021 commented on code in PR #19884:
URL: https://github.com/apache/druid/pull/19884#discussion_r3720541485
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java:
##########
@@ -615,8 +615,16 @@ private int updateReplicasInTier(
final int movingReplicas = replicaCountOnTier.moving();
final boolean shouldCancelMoves = requiredReplicas == 0 && movingReplicas
> 0;
+ // A replica serving under a partial-load profile is pinned by a
partial-load rule, but we got here through the
+ // regular full-load path, so that rule no longer applies: the
datasource's partial-load rule was replaced or
+ // shadowed by a higher-priority load rule, or its matcher stopped
resolving and fell through to FULL_LOAD. That
+ // replica needs an in-place reload carrying the plain unwrapped load spec
so the historical releases its rule
+ // holds. When the tier wants no replicas at all we skip it, the drops
below are already on their way and dropping
+ // clears the rule on the historical.
+ final int replicasToRevert = requiredReplicas > 0 ?
replicaCountOnTier.loadedWithPartialProfile() : 0;
Review Comment:
[P2] Revert profiles during partial-to-broadcast transitions
This reconciliation is only reached through `replicateSegment`, while
`BroadcastDistributionRule` calls `broadcastSegment`; its
`loadBroadcastSegment` immediately returns for an already-serving historical.
Replacing a partial rule with a broadcast rule therefore never sends the
unwrapped reload, leaving the old partial profile and rule holds indefinitely.
Apply equivalent reversion in the broadcast path and cover this transition.
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java:
##########
@@ -645,20 +653,54 @@ private int updateReplicasInTier(
}
// Cancel loads and queue drops if the projected count exceeds the
requirement
+ int dropsQueuedOnTier = 0;
if (projectedReplicas > requiredReplicas) {
int replicaSurplus = projectedReplicas - requiredReplicas;
int canceledLoads =
cancelOperations(SegmentAction.LOAD, replicaSurplus, segment,
segmentStatus);
int numReplicasToDrop = Math.min(replicaSurplus - canceledLoads,
maxReplicasToDrop);
if (numReplicasToDrop > 0) {
- int dropsQueuedOnTier = dropReplicas(numReplicasToDrop, segment, tier,
segmentStatus);
+ dropsQueuedOnTier = dropReplicas(numReplicasToDrop, segment, tier,
segmentStatus);
incrementStat(Stats.Segments.DROPPED, segment, tier,
dropsQueuedOnTier);
- return dropsQueuedOnTier;
}
}
- return 0;
+ // Release partial-load rules that no longer apply. Done last so the
load/drop decisions above claim their
+ // servers first: a replica that just picked up an action is no longer
`isServingSegment`, so it is skipped here
+ // and reverted on a later run if it is still around.
+ if (replicasToRevert > 0) {
+ final int reverted = revertPartialProfileReplicas(segment, tier);
+ if (reverted > 0) {
+ incrementStat(Stats.Segments.PARTIAL_RULE_REVERTED, segment, tier,
reverted);
+ }
+ }
+
+ return dropsQueuedOnTier;
+ }
+
+ /**
+ * Queues an in-place reload on every server in {@code tier} that serves
{@code segment} under a
+ * {@link PartialLoadProfile}. The request carries the plain unwrapped
{@code segment}, which is what tells the
+ * historical to release its rule holds rather than apply or swap one.
+ * <p>
+ * The replica count is deliberately left alone: these servers <em>are</em>
serving, so they still satisfy the
+ * rule's replication requirement and must not be double-counted as a
deficit. This only refreshes what they hold.
+ * <p>
+ * Servers with any queued action are skipped via {@link
ServerHolder#isServingSegment}, which covers both the
+ * load/drop decisions made earlier in this run and operations left over
from a previous one.
+ */
+ private int revertPartialProfileReplicas(DataSegment segment, String tier)
+ {
+ int numReverted = 0;
+ for (ServerHolder server : cluster.getManagedHistoricalsByTier(tier)) {
+ if (server.isServingSegment(segment)
+ && server.getServer().getPartialLoadProfile(segment.getId()) != null
+ && loadQueueManager.loadSegment(segment, server, SegmentAction.LOAD,
null)) {
Review Comment:
[P1] Honor the per-server load queue limit
`ServerHolder.startOperation` does not reject operations after
`maxSegmentsInNodeLoadingQueue` is reached; normal load paths first check
`canLoadSegment` or `isLoadQueueFull`. This loop directly queues every profiled
segment, so changing a large datasource to a full-load rule can enqueue all
reverts on each historical in one coordinator run, bypassing the configured cap
and flooding the peon/historical. Check the queue budget before enqueueing and
add a capped-queue test.
--
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]
