capistrant commented on code in PR #19461:
URL: https://github.com/apache/druid/pull/19461#discussion_r3237563942
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentHolder.java:
##########
@@ -90,18 +92,54 @@ public SegmentHolder(
Duration requestTimeout,
@Nullable LoadPeonCallback callback
)
+ {
+ this(segment, action, null, requestTimeout, callback);
+ }
+
+ /**
+ * Creates a holder for a load (or move) request that may carry a
partial-load profile. When {@code profile} is
+ * non-null and the action is a load, the outbound {@link
SegmentChangeRequestLoad}'s segment has its load spec
+ * replaced with {@link PartialLoadProfile#wrappedLoadSpec()} so the
historical sees the partial-load wrapper. The
+ * holder's own {@link #segment} field stays the original so identity /
equals / hashCode / callback resolution
+ * remain unchanged from the regular full-load path.
+ */
+ public SegmentHolder(
+ DataSegment segment,
+ SegmentAction action,
+ @Nullable PartialLoadProfile profile,
+ Duration requestTimeout,
+ @Nullable LoadPeonCallback callback
+ )
{
this.segment = segment;
this.action = action;
- this.changeRequest = (action == SegmentAction.DROP)
- ? new SegmentChangeRequestDrop(segment)
- : new SegmentChangeRequestLoad(segment);
+ this.profile = profile;
+ if (action == SegmentAction.DROP) {
+ this.changeRequest = new SegmentChangeRequestDrop(segment);
+ } else if (profile != null && profile.wrappedLoadSpec() != null) {
Review Comment:
non-null profile with a null wrapped load spec silently falls back to a
normal request. is that ok or would we want to either do something exceptional
or log a warn of some kind to make sure it is addressed
##########
server/src/main/java/org/apache/druid/server/coordinator/ServerHolder.java:
##########
@@ -387,6 +400,16 @@ public int getNumQueuedSegments()
}
public boolean startOperation(SegmentAction action, DataSegment segment)
+ {
+ return startOperation(action, segment, null);
+ }
+
+ /**
+ * Like {@link #startOperation(SegmentAction, DataSegment)}, but
additionally records the {@link PartialLoadProfile}
Review Comment:
nit: wording feels weird to say it is like `startOperation(SegmentAction,
DataSegment)` when all that method does is call this method. I think this would
be better off just describing what it does with a note about profile handling
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/PartialSegmentStatusInTier.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.coordinator.loading;
+
+import org.apache.druid.server.coordinator.ServerHolder;
+import org.apache.druid.timeline.DataSegment;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.NavigableSet;
+import java.util.Objects;
+
+/**
+ * Classifies the servers in a tier by their relationship to a {@link
PartialLoadProfile} request for a specific
+ * segment. Used by the partial-load reconciler in {@link
StrategicSegmentAssigner} to decide what to load, drop, or
+ * cancel under the load-then-drop swap pattern.
+ * <p>
+ * Per the additive-historical model: matching means the announced fingerprint
equals the requested fingerprint. A
+ * full-fallback profile (where the historical was asked to partial-load but
couldn't, announcing
+ * {@code wrappedLoadSpec=null} with the requested fingerprint) is also
treated as matching, the rule was
+ * satisfied even though the historical fell back to full.
+ * <p>
+ * Replicas without any profile (regular full-load) are always classified as
stale relative to a partial-load rule.
+ */
+public class PartialSegmentStatusInTier
+{
+ private final List<ServerHolder> matchingLoaded = new ArrayList<>();
+ private final List<ServerHolder> staleLoaded = new ArrayList<>();
+ private final List<ServerHolder> matchingInFlight = new ArrayList<>();
+ private final List<ServerHolder> staleInFlight = new ArrayList<>();
+ private final List<ServerHolder> eligibleForFreshLoad = new ArrayList<>();
+ private final List<ServerHolder> eligibleForAdditiveReload = new
ArrayList<>();
+
+ public PartialSegmentStatusInTier(
+ DataSegment segment,
+ String requestedFingerprint,
+ NavigableSet<ServerHolder> historicals
+ )
+ {
+ for (ServerHolder server : historicals) {
+ classify(server, segment, requestedFingerprint);
+ }
+ }
+
+ /**
+ * Servers that have the segment loaded with a profile whose fingerprint
matches the request (including full-fallback
+ * announcements with the matching fingerprint). Count toward the tier's
required matching replica count.
+ */
+ public List<ServerHolder> getMatchingLoaded()
+ {
+ return matchingLoaded;
+ }
+
+ /**
+ * Servers that have the segment loaded but with a non-matching profile
(different fingerprint, or no profile at all,
+ * i.e. a regular full-load replica seen against a partial rule). Eligible
for additive reload (the historical
+ * fills in the missing parts in place) and for being dropped once enough
matching replicas exist.
+ */
+ public List<ServerHolder> getStaleLoaded()
+ {
+ return staleLoaded;
+ }
+
+ /**
+ * Servers with an in-flight load whose profile fingerprint matches the
request. Counts toward projected matching
+ * replicas, the load is on its way to satisfying the rule.
+ */
+ public List<ServerHolder> getMatchingInFlight()
+ {
+ return matchingInFlight;
+ }
+
+ /**
+ * Servers with an in-flight load whose profile fingerprint differs from the
request (e.g., the rule changed
+ * mid-flight, or an in-flight regular full-load against a partial rule).
Cancel-and-replace targets when there is a
+ * matching deficit.
+ */
+ public List<ServerHolder> getStaleInFlight()
+ {
+ return staleInFlight;
+ }
+
+ /**
+ * Servers that don't have the segment and can take a fresh load (preferred
destinations for new replicas; empty
+ * slots, no in-place mutation needed).
+ */
+ public List<ServerHolder> getEligibleForFreshLoad()
+ {
+ return eligibleForFreshLoad;
+ }
+
+ /**
+ * Stale-loaded servers that can take an additive reload request (same as
{@link #getStaleLoaded()} once filtered for
+ * decommissioning / queue-full / pending-action, kept separate so the
reconciler can prefer fresh-load destinations
+ * before falling back to in-place additive reload). The historical's
additive load semantics make this the
+ * mitigation for the "no spare server" stuck state.
+ */
+ public List<ServerHolder> getEligibleForAdditiveReload()
+ {
+ return eligibleForAdditiveReload;
+ }
+
+ private void classify(ServerHolder server, DataSegment segment, String
requestedFingerprint)
+ {
+ final SegmentAction action = server.getActionOnSegment(segment);
+ final boolean isLoaded = server.isServingSegment(segment);
+
+ if (isLoaded) {
+ final PartialLoadProfile loaded =
server.getServer().getPartialLoadProfile(segment.getId());
+ if (loaded != null && Objects.equals(loaded.fingerprint(),
requestedFingerprint)) {
+ matchingLoaded.add(server);
+ } else {
+ staleLoaded.add(server);
+ if (canReloadAdditively(server)) {
+ eligibleForAdditiveReload.add(server);
+ }
+ }
+ } else if (action == SegmentAction.LOAD || action ==
SegmentAction.REPLICATE) {
+ final PartialLoadProfile inFlight = server.getInFlightProfile(segment);
+ if (inFlight != null && Objects.equals(inFlight.fingerprint(),
requestedFingerprint)) {
+ matchingInFlight.add(server);
+ } else {
+ staleInFlight.add(server);
+ }
+ } else if (action == null && server.canLoadSegment(segment)) {
+ eligibleForFreshLoad.add(server);
+ }
+ // Other actions (DROP, MOVE_TO, MOVE_FROM) are intentionally not
classified here, they're handled by the
+ // existing replica-counting paths.
Review Comment:
Potentially confusing comment. I assume you mean in extension to the
!isLoaded case we intentionally do not classify, since in the case where teh
segment is loaded we don't check action at all. Can this be dropped and
replaced by a javadoc that kind of sets the full expectation of what is
classified and what is purposely ignored?
##########
server/src/main/java/org/apache/druid/server/coordination/SegmentChangeRequestLoad.java:
##########
@@ -38,6 +40,29 @@
*/
public class SegmentChangeRequestLoad implements DataSegmentChangeRequest
{
+ /**
+ * Builds a load announcement for a segment loaded on a historical. When the
segment was loaded via a
+ * {@link PartialProjectionLoadSpec} wrapper (the coordinator stamped a
partial-load request onto the outbound
+ * segment), the announcement carries the wrapper's fingerprint and {@link
DataSegment#getSize()} as
+ * {@code loadedBytes}; a "full-fallback" advertisement that satisfies the
coordinator's partial-load rule even
+ * though the historical did a regular full load via the inner delegate.
Without this, the coordinator's reconciler
+ * would treat the replica as stale and re-queue the load indefinitely.
+ * <p>
+ * For segments loaded without a partial-load wrapper (the common case),
this returns a bare load request with no
+ * fingerprint or loadedBytes, equivalent to {@link
#SegmentChangeRequestLoad(DataSegment)}.
+ */
+ public static SegmentChangeRequestLoad forAnnouncement(DataSegment segment)
+ {
+ final Map<String, Object> loadSpec = segment.getLoadSpec();
+ if (loadSpec != null &&
PartialProjectionLoadSpec.TYPE.equals(loadSpec.get("type"))) {
+ final Object fingerprint = loadSpec.get("fingerprint");
+ if (fingerprint instanceof String) {
Review Comment:
warn if fingerprint is not string type? maybe/hopefully impossible but
silently falling back to a full load request if this did fail seems hard to
troubleshoot. Unless non-string fingerprints would blow some other part of the
code up
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java:
##########
@@ -273,6 +274,273 @@ public void replicateSegment(DataSegment segment,
Map<String, Integer> tierToRep
}
}
+ /**
+ * Fingerprint-aware partial-load reconciler. Mirrors {@link
#replicateSegment} for replica-count and surplus
+ * accounting, but per-tier the load/drop decisions are made by classifying
replicas as matching or stale relative
+ * to the requested {@link PartialLoadProfile#fingerprint()} and applying
the load-then-drop swap pattern: stale
+ * replicas keep serving until matching replicas have actually loaded.
+ */
+ @Override
+ public void replicateSegmentPartially(
+ DataSegment segment,
+ PartialLoadProfile profile,
+ Map<String, Integer> tierToReplicaCount
+ )
+ {
+ final Map<String, Integer> effectiveTierToReplicaCount =
expandWithAliases(tierToReplicaCount);
+ final Set<String> allTiersInCluster =
Sets.newHashSet(cluster.getTierNames());
+
+ if (effectiveTierToReplicaCount.isEmpty()) {
+ replicaCountMap.computeIfAbsent(segment.getId(),
DruidServer.DEFAULT_TIER);
+ } else {
+ effectiveTierToReplicaCount.forEach((tier, requiredReplicas) -> {
+ reportTierCapacityStats(segment, requiredReplicas, tier);
+
+ SegmentReplicaCount replicaCount =
replicaCountMap.computeIfAbsent(segment.getId(), tier);
+ replicaCount.setRequired(requiredReplicas,
tierToHistoricalCount.getOrDefault(tier, 0));
+
+ if (!allTiersInCluster.contains(tier)) {
+
datasourceToInvalidLoadTiers.computeIfAbsent(segment.getDataSource(), ds -> new
HashSet<>())
+ .add(tier);
+ }
+ });
+ }
+
+ final SegmentReplicaCount replicaCountInCluster =
replicaCountMap.getTotal(segment.getId());
+ if (replicaCountInCluster.required() <= 0) {
+ segmentsWithZeroRequiredReplicas
+ .computeIfAbsent(segment.getDataSource(), ds -> new HashSet<>())
+ .add(segment);
+ }
+
+ final int replicaSurplus = replicaCountInCluster.loadedNotDropping()
+ - replicaCountInCluster.requiredAndLoadable();
+
+ int dropsQueued = 0;
+ for (String tier : allTiersInCluster) {
+ dropsQueued += updateReplicasInTierPartial(
+ segment,
+ profile,
+ tier,
+ effectiveTierToReplicaCount.getOrDefault(tier, 0),
+ replicaSurplus - dropsQueued
+ );
+ }
+ }
+
+ /**
+ * Per-tier reconciliation under the partial-load model. The flow is:
+ * <ol>
+ * <li>Compute matching count: matching-loaded + matching-in-flight −
pending-move-drop.</li>
+ * <li>If matching count is short of {@code requiredReplicas}, cancel
stale-fingerprint in-flight loads to free
+ * slots, then queue fresh partial-load requests on eligible servers
(preferring empty servers, falling back
+ * to additive reload on stale-loaded servers; the historical fills in
missing parts in place).</li>
+ * <li>If matching count exceeds requirement, drop the excess like the
full-load surplus path.</li>
+ * <li>Drop stale-loaded replicas only when the count of actually-loaded
matching replicas already meets the
+ * requirement; this preserves availability across the swap (stale
replicas keep serving until matching
+ * replicas have completed loading and announced).</li>
+ * </ol>
+ * Returns the total number of drop operations queued on this tier (matching
surplus + stale), used to budget
+ * cross-tier drop pressure.
+ */
+ private int updateReplicasInTierPartial(
+ DataSegment segment,
+ PartialLoadProfile profile,
+ String tier,
+ int requiredReplicas,
+ int maxReplicasToDrop
+ )
+ {
+ final SegmentReplicaCount replicaCountOnTier =
replicaCountMap.get(segment.getId(), tier);
+ final int movingReplicas = replicaCountOnTier.moving();
+ final int moveCompletedPendingDrop = Math.max(0,
replicaCountOnTier.moveCompletedPendingDrop());
+
+ final PartialSegmentStatusInTier status = new PartialSegmentStatusInTier(
+ segment,
+ profile.fingerprint(),
+ cluster.getManagedHistoricalsByTier(tier)
+ );
+
+ final int matchingProjected = status.getMatchingLoaded().size()
+ + status.getMatchingInFlight().size()
+ - moveCompletedPendingDrop;
+ final boolean shouldCancelMoves = requiredReplicas == 0 && movingReplicas
> 0;
+
+ // If everything's already in shape and no stale work, fast-exit.
+ if (matchingProjected == requiredReplicas
+ && !shouldCancelMoves
+ && status.getStaleInFlight().isEmpty()
+ && (status.getStaleLoaded().isEmpty() ||
status.getMatchingLoaded().size() < requiredReplicas)) {
+ return 0;
+ }
+
+ if (shouldCancelMoves) {
+ // Convert to SegmentStatusInTier for the existing cancelOperations
move-cancel helper.
+ final SegmentStatusInTier vanillaStatus =
+ new SegmentStatusInTier(segment,
cluster.getManagedHistoricalsByTier(tier));
+ cancelOperations(SegmentAction.MOVE_TO, movingReplicas, segment,
vanillaStatus);
+ cancelOperations(SegmentAction.MOVE_FROM, movingReplicas, segment,
vanillaStatus);
+ }
+
+ // Cancel stale in-flight: when there's a matching deficit we want their
slots back; when requirement is 0 we
+ // want them gone unconditionally so we don't realize a stale fingerprint
nobody asked for. Canceled servers
+ // become eligible for a fresh matching load later in this same run.
+ final int matchingDeficit = requiredReplicas - matchingProjected;
+ final List<ServerHolder> cancelledStaleServers = new ArrayList<>();
+ if (matchingDeficit > 0 || requiredReplicas == 0) {
+ final int toCancel = requiredReplicas == 0 ?
status.getStaleInFlight().size() : matchingDeficit;
+ cancelLoadsOnServers(segment, status.getStaleInFlight(), toCancel,
cancelledStaleServers);
+ if (!cancelledStaleServers.isEmpty()) {
+ incrementStat(Stats.Segments.PARTIAL_STALE_CANCELLED, segment, tier,
cancelledStaleServers.size());
+ }
+ }
+
+ // Queue fresh matching loads to fill the deficit.
+ if (matchingDeficit > 0) {
+ final int numLoadedReplicas = status.getMatchingLoaded().size() +
status.getStaleLoaded().size();
+ final int queued = loadPartialReplicas(
+ matchingDeficit, numLoadedReplicas, segment, tier, status,
cancelledStaleServers, profile
+ );
+ if (queued > 0) {
+ incrementStat(Stats.Segments.PARTIAL_ASSIGNED, segment, tier, queued);
+ }
+ }
+
+ int dropsQueuedOnTier = 0;
+ int dropBudget = maxReplicasToDrop;
+
+ // Surplus matching: drop excess matching replicas. Same shape as the
full-load surplus path.
+ if (matchingProjected > requiredReplicas) {
+ final int surplus = matchingProjected - requiredReplicas;
+ final List<ServerHolder> cancelledMatching = new ArrayList<>();
+ cancelLoadsOnServers(segment, status.getMatchingInFlight(), surplus,
cancelledMatching);
Review Comment:
appear to be missing incrementStat if the cancel takes some action
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/PartialSegmentStatusInTier.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.coordinator.loading;
+
+import org.apache.druid.server.coordinator.ServerHolder;
+import org.apache.druid.timeline.DataSegment;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.NavigableSet;
+import java.util.Objects;
+
+/**
+ * Classifies the servers in a tier by their relationship to a {@link
PartialLoadProfile} request for a specific
+ * segment. Used by the partial-load reconciler in {@link
StrategicSegmentAssigner} to decide what to load, drop, or
+ * cancel under the load-then-drop swap pattern.
+ * <p>
+ * Per the additive-historical model: matching means the announced fingerprint
equals the requested fingerprint. A
+ * full-fallback profile (where the historical was asked to partial-load but
couldn't, announcing
+ * {@code wrappedLoadSpec=null} with the requested fingerprint) is also
treated as matching, the rule was
+ * satisfied even though the historical fell back to full.
+ * <p>
+ * Replicas without any profile (regular full-load) are always classified as
stale relative to a partial-load rule.
+ */
+public class PartialSegmentStatusInTier
+{
+ private final List<ServerHolder> matchingLoaded = new ArrayList<>();
+ private final List<ServerHolder> staleLoaded = new ArrayList<>();
+ private final List<ServerHolder> matchingInFlight = new ArrayList<>();
+ private final List<ServerHolder> staleInFlight = new ArrayList<>();
+ private final List<ServerHolder> eligibleForFreshLoad = new ArrayList<>();
+ private final List<ServerHolder> eligibleForAdditiveReload = new
ArrayList<>();
+
+ public PartialSegmentStatusInTier(
+ DataSegment segment,
+ String requestedFingerprint,
+ NavigableSet<ServerHolder> historicals
+ )
+ {
+ for (ServerHolder server : historicals) {
+ classify(server, segment, requestedFingerprint);
+ }
+ }
+
+ /**
+ * Servers that have the segment loaded with a profile whose fingerprint
matches the request (including full-fallback
+ * announcements with the matching fingerprint). Count toward the tier's
required matching replica count.
+ */
+ public List<ServerHolder> getMatchingLoaded()
+ {
+ return matchingLoaded;
+ }
+
+ /**
+ * Servers that have the segment loaded but with a non-matching profile
(different fingerprint, or no profile at all,
+ * i.e. a regular full-load replica seen against a partial rule). Eligible
for additive reload (the historical
+ * fills in the missing parts in place) and for being dropped once enough
matching replicas exist.
+ */
+ public List<ServerHolder> getStaleLoaded()
+ {
+ return staleLoaded;
+ }
+
+ /**
+ * Servers with an in-flight load whose profile fingerprint matches the
request. Counts toward projected matching
+ * replicas, the load is on its way to satisfying the rule.
+ */
+ public List<ServerHolder> getMatchingInFlight()
+ {
+ return matchingInFlight;
+ }
+
+ /**
+ * Servers with an in-flight load whose profile fingerprint differs from the
request (e.g., the rule changed
+ * mid-flight, or an in-flight regular full-load against a partial rule).
Cancel-and-replace targets when there is a
+ * matching deficit.
+ */
+ public List<ServerHolder> getStaleInFlight()
+ {
+ return staleInFlight;
+ }
+
+ /**
+ * Servers that don't have the segment and can take a fresh load (preferred
destinations for new replicas; empty
+ * slots, no in-place mutation needed).
+ */
+ public List<ServerHolder> getEligibleForFreshLoad()
+ {
+ return eligibleForFreshLoad;
+ }
+
+ /**
+ * Stale-loaded servers that can take an additive reload request (same as
{@link #getStaleLoaded()} once filtered for
+ * decommissioning / queue-full / pending-action, kept separate so the
reconciler can prefer fresh-load destinations
+ * before falling back to in-place additive reload). The historical's
additive load semantics make this the
+ * mitigation for the "no spare server" stuck state.
+ */
+ public List<ServerHolder> getEligibleForAdditiveReload()
+ {
+ return eligibleForAdditiveReload;
+ }
+
+ private void classify(ServerHolder server, DataSegment segment, String
requestedFingerprint)
+ {
+ final SegmentAction action = server.getActionOnSegment(segment);
+ final boolean isLoaded = server.isServingSegment(segment);
+
+ if (isLoaded) {
+ final PartialLoadProfile loaded =
server.getServer().getPartialLoadProfile(segment.getId());
+ if (loaded != null && Objects.equals(loaded.fingerprint(),
requestedFingerprint)) {
+ matchingLoaded.add(server);
+ } else {
+ staleLoaded.add(server);
+ if (canReloadAdditively(server)) {
+ eligibleForAdditiveReload.add(server);
+ }
+ }
+ } else if (action == SegmentAction.LOAD || action ==
SegmentAction.REPLICATE) {
+ final PartialLoadProfile inFlight = server.getInFlightProfile(segment);
+ if (inFlight != null && Objects.equals(inFlight.fingerprint(),
requestedFingerprint)) {
+ matchingInFlight.add(server);
+ } else {
+ staleInFlight.add(server);
+ }
+ } else if (action == null && server.canLoadSegment(segment)) {
+ eligibleForFreshLoad.add(server);
+ }
+ // Other actions (DROP, MOVE_TO, MOVE_FROM) are intentionally not
classified here, they're handled by the
+ // existing replica-counting paths.
+ }
+
+ /**
+ * A stale-loaded server is reload-eligible iff it's not decommissioning,
has no other action queued for the segment,
Review Comment:
do we need something more to satisfy `has no other action queued for the
segment`? In conjunction with `classify` it seems like we are not checking for
the stale loaded but already queued additive load scenario.
##########
server/src/main/java/org/apache/druid/server/http/DataSourcesResource.java:
##########
@@ -755,7 +755,7 @@ private ImmutableDruidDataSource
getQueryableDataSource(final String dataSourceN
}
}
- return new ImmutableDruidDataSource(dataSourceName,
Collections.emptyMap(), segmentMap);
+ return new ImmutableDruidDataSource(dataSourceName,
Collections.emptyMap(), segmentMap.values());
Review Comment:
If constructor choice here is material to the functioning of the platform, I
think that the choice should be explained here in a short comment to avoid
regression
##########
server/src/main/java/org/apache/druid/client/HttpServerInventoryView.java:
##########
@@ -659,6 +664,24 @@ public CallbackAction apply(SegmentCallback input)
}
}
+ /**
+ * Builds a {@link PartialLoadProfile} from a load announcement when the
historical populated the partial-load
+ * fields ({@code fingerprint} + {@code loadedBytes}). Returns null for
full-load announcements.
+ */
+ @Nullable
+ private static PartialLoadProfile
partialLoadProfileFor(SegmentChangeRequestLoad loadRequest)
+ {
+ final String fingerprint = loadRequest.getFingerprint();
+ final Long loadedBytes = loadRequest.getLoadedBytes();
+ if (fingerprint == null || loadedBytes == null) {
+ return null;
+ }
+ // The wrapped LoadSpec is on the segment itself when a partial load was
requested by the coordinator; we don't
+ // duplicate it on the inventory profile; the announcement uses the
full-fallback factory which carries the
+ // fingerprint and the realized loadedBytes only.
+ return PartialLoadProfile.forFullFallback(fingerprint, loadedBytes);
Review Comment:
Is it weird to be doing this? You justify it in the comment, but it still
feels like we should either generalize the factory name/javadoc or create a new
one that does the same, but is not named and documented as a full load fallback
mechanism for the historical
--
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]
