This is an automated email from the ASF dual-hosted git repository.
clintropolis pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new bbb161c37dd feat: partial load rules assignment prefer in-place reload
over fresh load (#20267)
bbb161c37dd is described below
commit bbb161c37dd61e710c5a3cf89fed642701bd79b0
Author: Clint Wylie <[email protected]>
AuthorDate: Fri Sep 11 13:03:13 2026 -0700
feat: partial load rules assignment prefer in-place reload over fresh load
(#20267)
---
.../segment/loading/SegmentLocalCacheManager.java | 182 ++++++++++++-----
.../org/apache/druid/server/SegmentManager.java | 34 +++-
.../coordination/SegmentLoadDropHandler.java | 13 +-
.../druid/server/coordinator/ServerHolder.java | 35 +++-
.../loading/PartialSegmentStatusInTier.java | 64 +++---
.../loading/StrategicSegmentAssigner.java | 115 ++++++++---
.../segment/loading/NoopSegmentCacheManager.java | 2 +-
...egmentLocalCacheManagerPartialRuleLoadTest.java | 161 ++++++++++++++-
.../apache/druid/server/SegmentManagerTest.java | 38 ++++
.../coordination/SegmentLoadDropHandlerTest.java | 90 +++++++++
.../druid/server/coordinator/ServerHolderTest.java | 103 ++++++++++
.../StrategicSegmentAssignerPartialTest.java | 225 ++++++++++++++++++++-
.../druid/test/utils/TestSegmentCacheManager.java | 22 +-
13 files changed, 951 insertions(+), 133 deletions(-)
diff --git
a/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
b/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
index 91fd583d4db..3a881615881 100644
---
a/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
+++
b/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
@@ -36,6 +36,7 @@ import org.apache.druid.java.util.common.FileUtils;
import org.apache.druid.java.util.common.IAE;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.Stopwatch;
+import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.io.Closer;
import org.apache.druid.java.util.emitter.EmittingLogger;
@@ -58,6 +59,7 @@ import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
@@ -1025,19 +1027,13 @@ public class SegmentLocalCacheManager implements
SegmentCacheManager
final ReferenceCountingLock lock = lock(dataSegment);
synchronized (lock) {
try {
- // If a stale non-partial cache entry sits at this segment id (a
CompleteSegmentCacheEntry created by a prior
- // acquireSegment while virtualStoragePartialDownloadsEnabled=false,
for example), evict it before any
- // partial-entry lookup or reservation; otherwise
findExistingPartialWithHold's defensive type-check would
- // throw, and reservePartial's addWeakReservationHold would land on
the incompatible entry. If the stale
- // entry is currently held (in-flight query), this throws a retryable
SegmentLoadingException; the
- // coordinator's load queue retries on next sync, and by then the
query should have released.
- evictStaleNonPartialWeakEntry(dataSegment.getId());
-
if (rangeReader == null) {
// Backend doesn't support range reads (e.g. zipped deep storage).
The rule can't be honored as a partial
// load; clear any prior rule so the segment falls through to the
ordinary weak-full-load path at query
- // time.
- final ReservedPartial existing =
findExistingPartialWithHold(dataSegment.getId());
+ // time. A non-partial entry carries no rule, so there is nothing to
release for one.
+ final ReservedPartial existing =
hasNonPartialEntry(dataSegment.getId())
+ ? null
+ :
findExistingPartialWithHold(dataSegment.getId());
if (existing != null) {
try {
// Snapshot the prior realized footprint before clearRule zeroes
out ruleBundleHolds so the log can
@@ -1076,22 +1072,14 @@ public class SegmentLocalCacheManager implements
SegmentCacheManager
return dataSegment;
}
+ // Committed to attempting the rule now. If a stale non-partial cache
entry sits at this segment id (a
+ // complete created by a prior acquireSegment while
virtualStoragePartialDownloadsEnabled=false, for example),
+ // evict it before any partial-entry lookup or reservation.
+ evictStaleNonPartialWeakEntry(dataSegment.getId());
+
final ReservedPartial reserved = findOrReservePartial(dataSegment,
rangeReader);
try {
final PartialSegmentMetadataCacheEntry metadata =
reserved.metadata();
- // findOrReservePartial only invokes reservePartial (which writes
the info file) on the fresh-reserve
- // branch. On the find-existing branch the info file on disk still
carries the PRIOR rule's wrapped
- // load spec, so a rule swap here would apply in memory only.
Rewrite unconditionally before mount.
- try {
- rewriteInfoFile(dataSegment);
- }
- catch (IOException e) {
- throw new SegmentLoadingException(
- e,
- "Failed to write partial info file for segment[%s]",
- dataSegment.getId()
- );
- }
try {
metadata.mount(reserved.location());
}
@@ -1116,8 +1104,13 @@ public class SegmentLocalCacheManager implements
SegmentCacheManager
final Set<String> selected = Set.copyOf(
wrapper.getSelectedBundleNames(dataSegment,
mapper.getSegmentFileMetadata())
);
+ // Snapshot the rule this call is about to replace, then pin the
UNION of it and the new selection for the
+ // duration of the attempt.
final String priorFingerprint = metadata.getRuleFingerprint();
- metadata.applyRule(wrapper.getFingerprint(), selected);
+ final Set<String> priorSelection =
metadata.getRuleSelectedBundleNames();
+ final Set<String> attemptSelection = new HashSet<>(priorSelection);
+ attemptSelection.addAll(selected);
+ metadata.applyRule(wrapper.getFingerprint(), attemptSelection);
if (priorFingerprint != null &&
!priorFingerprint.equals(wrapper.getFingerprint())) {
log.info(
"Reconciled partial-load rule for segment[%s]: fingerprint
transitioned [%s] → [%s]",
@@ -1126,11 +1119,17 @@ public class SegmentLocalCacheManager implements
SegmentCacheManager
wrapper.getFingerprint()
);
}
- // Block until every eager download completes so the announcement
fingerprint reflects reality: any failure
- // clears the rule state (releasing self-hold + all bundle
rule-holds) and propagates as a load failure so
- // the coordinator's load queue can retry on its next sync. The
announced fingerprint == "rule fully
- // realized" contract stays intact.
- awaitEagerDownloadsOrClearRule(dataSegment, metadata, selected);
+ // Block until every eager download completes, then narrow the union
pinned above down to the new rule. Any
+ // failure rolls the rule state back to the prior rule and
propagates as a load failure so the coordinator's
+ // load queue can retry on its next sync.
+ realizeRuleOrRestorePrior(
+ dataSegment,
+ metadata,
+ wrapper.getFingerprint(),
+ selected,
+ priorFingerprint,
+ priorSelection
+ );
// Wrap the announcement with a DataSegmentAndLoadProfile carrying
the historical's realized footprint AFTER
// eager downloads finished. forAnnouncement reads the profile back
via profileOf() and stamps its
// loadedBytes + fingerprint.
@@ -1154,14 +1153,19 @@ public class SegmentLocalCacheManager implements
SegmentCacheManager
}
/**
- * Submit eager-download tasks to the loading pool for every rule-selected
bundle not yet registered with
- * {@code metadata}, then block until every task completes. On any failure
the rule state is cleared
- * and a {@link SegmentLoadingException} is thrown so the caller treats the
load as failed and retries.
+ * Completes the second half of a rule swap. The caller has pinned the union
of the prior and new selections; this
+ * submits eager-download tasks to the loading pool for every bundle of the
new selection that is not resident yet,
+ * blocks until they all finish, then commits by persisting the new wrapper
to the segment's info file and narrowing
+ * the pin down to {@code selected}. On any failure it puts {@code
priorFingerprint} / {@code priorSelection} back
+ * and throws {@link SegmentLoadingException} so the caller treats the load
as failed and retries.
*/
- private void awaitEagerDownloadsOrClearRule(
+ private void realizeRuleOrRestorePrior(
DataSegment dataSegment,
PartialSegmentMetadataCacheEntry metadata,
- Set<String> selected
+ String fingerprint,
+ Set<String> selected,
+ @Nullable String priorFingerprint,
+ Set<String> priorSelection
) throws SegmentLoadingException
{
final List<Future<?>> pending = new ArrayList<>();
@@ -1236,13 +1240,61 @@ public class SegmentLocalCacheManager implements
SegmentCacheManager
}
}
- if (firstFailure != null) {
- // Any late-completing pool task that still succeeds after we've cleared
the rule will call registerBundle →
- // observing ruleSelectedBundleNames == {} and skipping the rule-hold
acquire.
- metadata.clearRule();
- throw new SegmentLoadingException(
- firstFailure,
- "Failed eager download of rule-selected bundles for segment[%s];
cleared partial-load rule",
+ if (firstFailure == null) {
+ try {
+ rewriteInfoFile(dataSegment);
+ // every bundle the new rule wants is resident and pinned, so the
prior rule's extras can go. Pure
+ // release, since the target selection is a subset of what is held.
+ metadata.applyRule(fingerprint, selected);
+ return;
+ }
+ catch (Throwable t) {
+ firstFailure = t;
+ }
+ }
+
+ restorePriorRule(dataSegment, metadata, priorFingerprint, priorSelection);
+ throw new SegmentLoadingException(
+ firstFailure,
+ "Failed to realize partial-load rule[fingerprint=%s] for segment[%s];
%s",
+ fingerprint,
+ dataSegment.getId(),
+ priorFingerprint == null
+ ? "cleared partial-load rule"
+ : StringUtils.format("restored prior partial-load
rule[fingerprint=%s]", priorFingerprint)
+ );
+ }
+
+ /**
+ * Puts {@code metadata} back on the rule it held before a failed {@link
#loadPartial} attempt, by releasing the
+ * holds that attempt acquired.
+ * <p>
+ * This restores the prior rule <em>exactly</em>, because the attempt never
released a prior hold: the caller pinned
+ * the union of the two selections up front, so every bundle the prior rule
wants is still held and
+ * {@link PartialSegmentMetadataCacheEntry#applyRule} has nothing to
re-acquire here..
+ * <p>
+ * Failure to restore must not mask the download failure that got us here,
so it is logged and swallowed. It is not
+ * expected: releasing a hold does not fail.
+ */
+ private void restorePriorRule(
+ DataSegment dataSegment,
+ PartialSegmentMetadataCacheEntry metadata,
+ @Nullable String priorFingerprint,
+ Set<String> priorSelection
+ )
+ {
+ try {
+ if (priorFingerprint == null) {
+ metadata.clearRule();
+ } else {
+ metadata.applyRule(priorFingerprint, priorSelection);
+ }
+ }
+ catch (Throwable t) {
+ log.warn(
+ t,
+ "Failed to restore prior partial-load rule[fingerprint=%s] on
segment[%s] after a failed reload",
+ priorFingerprint,
dataSegment.getId()
);
}
@@ -1608,10 +1660,9 @@ public class SegmentLocalCacheManager implements
SegmentCacheManager
* reclaim of the partial state on disk is left to eviction, as it is for
{@link #drop}.
* <p>
* The info file is rewritten before the rule is cleared, and a failed
rewrite fails the load. Nothing is left half
- * converted: releasing the holds cannot fail, and a load failure sends the
historical down its drop path, which
- * clears the rule and removes the info file, so there is no stale rule for
a restart to reinstate. Leaving the rule
- * applied and carrying on is not an option, because an unwrapped request
announces as a full load either way, so the
- * coordinator would record a replica with no profile and never ask again.
+ * converted, because a failed rewrite converts nothing at all: {@link
#writeInfoFile} is atomic, so the info file
+ * still has the partial wrapper, and {@link
PartialSegmentMetadataCacheEntry#clearRule} has not run, so the
+ * rule and every hold it owns are still in place.
* <p>
* Callers must hold this segment's {@link #lock(DataSegment)}, which is the
external lock that
* {@link PartialSegmentMetadataCacheEntry#clearRule} requires to be
serialized against
@@ -1643,6 +1694,25 @@ public class SegmentLocalCacheManager implements
SegmentCacheManager
);
}
+ /**
+ * Whether any location holds a cache entry for {@code segmentId} that is
<em>not</em> a
+ * {@link PartialSegmentMetadataCacheEntry}. Such an entry never carries a
partial-load rule, and it is what
+ * {@link #evictStaleNonPartialWeakEntry} has to clear out of the way before
a partial entry can be reserved at the
+ * same id. Callers that are not going to reserve one use this to skip the
partial-entry lookup, whose defensive
+ * type-check would otherwise throw on it.
+ */
+ private boolean hasNonPartialEntry(SegmentId segmentId)
+ {
+ final SegmentCacheEntryIdentifier id = new
SegmentCacheEntryIdentifier(segmentId);
+ for (StorageLocation location : locations) {
+ final CacheEntry entry = location.getCacheEntry(id);
+ if (entry != null && !(entry instanceof
PartialSegmentMetadataCacheEntry)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
/**
* Whether any location already has a cache entry for {@code id}
*/
@@ -1775,8 +1845,13 @@ public class SegmentLocalCacheManager implements
SegmentCacheManager
* Reapply the persisted partial-load rule to a bootstrap-restored metadata
entry. Reads the wrapper from the
* segment's info-file {@code loadSpec}, resolves the selected bundle names
against the just-parsed on-disk
* metadata header, calls {@link
PartialSegmentMetadataCacheEntry#applyRule}, then drives eager downloads for any
- * selected bundle that wasn't restored from disk (via {@link
#awaitEagerDownloadsOrClearRule}). On failure the
+ * selected bundle that wasn't restored from disk (via {@link
#realizeRuleOrRestorePrior}). On failure the
* exception marks the segment as failed → doesn't announce it → the
coordinator's next sync re-issues load.
+ * <p>
+ * A bootstrap-restored entry starts with no rule applied, so the union
pinned here is just the new selection and
+ * the rollback degenerates to clearing the rule, which is the right
starting state for a first application that
+ * failed. The prior state is read back rather than assumed so this stays
correct if bootstrap ever restores a rule
+ * along with the entry.
*/
private void reapplyRuleFromInfoFile(DataSegment dataSegment,
PartialSegmentMetadataCacheEntry partial)
throws SegmentLoadingException
@@ -1796,8 +1871,19 @@ public class SegmentLocalCacheManager implements
SegmentCacheManager
final Set<String> selected = Set.copyOf(
wrapper.getSelectedBundleNames(dataSegment,
mapper.getSegmentFileMetadata())
);
- partial.applyRule(wrapper.getFingerprint(), selected);
- awaitEagerDownloadsOrClearRule(dataSegment, partial, selected);
+ final String priorFingerprint = partial.getRuleFingerprint();
+ final Set<String> priorSelection = partial.getRuleSelectedBundleNames();
+ final Set<String> attemptSelection = new HashSet<>(priorSelection);
+ attemptSelection.addAll(selected);
+ partial.applyRule(wrapper.getFingerprint(), attemptSelection);
+ realizeRuleOrRestorePrior(
+ dataSegment,
+ partial,
+ wrapper.getFingerprint(),
+ selected,
+ priorFingerprint,
+ priorSelection
+ );
}
@Override
diff --git a/server/src/main/java/org/apache/druid/server/SegmentManager.java
b/server/src/main/java/org/apache/druid/server/SegmentManager.java
index 73ab935a084..f8c43ec5646 100644
--- a/server/src/main/java/org/apache/druid/server/SegmentManager.java
+++ b/server/src/main/java/org/apache/druid/server/SegmentManager.java
@@ -141,6 +141,27 @@ public class SegmentManager
return
Optional.ofNullable(dataSources.get(dataSource.getName())).map(DataSourceState::getTimeline);
}
+ /**
+ * Whether this server is already serving {@code dataSegment}, i.e. it is in
its datasource's timeline and so is
+ * queryable right now. A load request for such a segment is a reload rather
than a new load, which is how a
+ * partial-load rule is applied, swapped, or released, see {@code
StrategicSegmentAssigner}.
+ * <p>
+ * This is a lock-free read of the timeline, unlike the {@code compute} that
{@link #loadSegment} mutates it under,
+ * so it answers for the instant it is called and nothing more.
+ */
+ public boolean isSegmentLoaded(final DataSegment dataSegment)
+ {
+ final DataSourceState dataSourceState =
dataSources.get(dataSegment.getDataSource());
+ if (dataSourceState == null) {
+ return false;
+ }
+ return dataSourceState.getTimeline().findChunk(
+ dataSegment.getInterval(),
+ dataSegment.getVersion(),
+ dataSegment.getShardSpec().getPartitionNum()
+ ) != null;
+ }
+
/**
* Given a list of {@link DataSegmentAndDescriptor} produce a {@link
LeafSegmentsBundle} which partitions segments
* into cached, loadable, or missing segments. This gives callers the
flexibilty to decide to perform operations
@@ -320,19 +341,16 @@ public class SegmentManager
* {@link org.apache.druid.client.DataSegmentAndLoadProfile}
wrapping it when the historical actually
* materialized a partial-load footprint. Callers pass the returned
value to the announcement layer so
* partial-load announcements carry accurate {@code loadedBytes}.
+ * <b>Failure cleanup belongs to the caller.</b> This method deliberately
discards nothing when the load fails,
+ * because it cannot tell on its own whether the state it would discard is
half-materialized leftovers from this
+ * attempt or a live replica.
+ *
* @throws SegmentLoadingException if the segment cannot be loaded
* @throws IOException if the segment info cannot be cached on disk
*/
public DataSegment loadSegment(final DataSegment dataSegment) throws
SegmentLoadingException, IOException
{
- final DataSegment loaded;
- try {
- loaded = cacheManager.load(dataSegment);
- }
- catch (SegmentLoadingException e) {
- cacheManager.drop(dataSegment);
- throw e;
- }
+ final DataSegment loaded = cacheManager.load(dataSegment);
// Pass the plain dataSegment (not the potentially-wrapped `loaded`) to
loadSegmentInternal: the wrapper is a
// load-time announcement-path artifact only
loadSegmentInternal(dataSegment);
diff --git
a/server/src/main/java/org/apache/druid/server/coordination/SegmentLoadDropHandler.java
b/server/src/main/java/org/apache/druid/server/coordination/SegmentLoadDropHandler.java
index 1e024fcc5bb..50590bd275b 100644
---
a/server/src/main/java/org/apache/druid/server/coordination/SegmentLoadDropHandler.java
+++
b/server/src/main/java/org/apache/druid/server/coordination/SegmentLoadDropHandler.java
@@ -166,7 +166,18 @@ public class SegmentLoadDropHandler
loaded = segmentManager.loadSegment(segment);
}
catch (Exception e) {
- removeSegment(segment, DataSegmentChangeCallback.NOOP, false);
+ // decides whether to clean up a failed load; a load request for a
segment this server already serves is a
+ // reload, not a new load, and we only want to discard the
half-materialized state a failed *new* load leaves
+ // behind.
+ if (segmentManager.isSegmentLoaded(segment)) {
+ log.warn(
+ e,
+ "Failed to load segment[%s], but it is serving already; leaving
it in place to be retried.",
+ segment.getId()
+ );
+ } else {
+ removeSegment(segment, DataSegmentChangeCallback.NOOP, false);
+ }
throw new SegmentLoadingException(e, "Exception loading segment[%s]",
segment.getId());
}
try {
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/ServerHolder.java
b/server/src/main/java/org/apache/druid/server/coordinator/ServerHolder.java
index 60593175d5f..f6cfd263bb4 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/ServerHolder.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/ServerHolder.java
@@ -223,7 +223,9 @@ public class ServerHolder implements
Comparable<ServerHolder>
* The total size:
* <ol>
* <li>INCLUDES segments loaded on this server</li>
- * <li>INCLUDES segments loading on this server (actions:
LOAD/REPLICATE)</li>
+ * <li>INCLUDES segments loading on this server (actions: LOAD/REPLICATE). A
load of a segment this server already
+ * serves is an in-place reload, and is counted at its {@link
#inPlaceReloadSizeDelta} rather than its full size, so
+ * that the bytes already on disk are not counted twice</li>
* <li>INCLUDES segments moving to this server (action: MOVE_TO)</li>
* <li>INCLUDES segments moving from this server (action: MOVE_FROM). This is
* because these segments have only been <i>marked</i> for drop. We include
@@ -515,8 +517,12 @@ public class ServerHolder implements
Comparable<ServerHolder>
// Add to projected if load is started, remove from projected if drop has
started
if (action.isLoad()) {
- projectedSegmentCounts.addSegment(segment);
- sizeOfLoadingSegments += segment.getSize();
+ if (hasSegmentLoaded(segment.getId())) {
+ sizeOfLoadingSegments += inPlaceReloadSizeDelta(segment);
+ } else {
+ projectedSegmentCounts.addSegment(segment);
+ sizeOfLoadingSegments += segment.getSize();
+ }
} else {
projectedSegmentCounts.removeSegment(segment);
if (action == SegmentAction.DROP) {
@@ -532,8 +538,12 @@ public class ServerHolder implements
Comparable<ServerHolder>
queuedSegments.remove(segment);
if (action.isLoad()) {
- projectedSegmentCounts.removeSegment(segment);
- sizeOfLoadingSegments -= segment.getSize();
+ if (hasSegmentLoaded(segment.getId())) {
+ sizeOfLoadingSegments -= inPlaceReloadSizeDelta(segment);
+ } else {
+ projectedSegmentCounts.removeSegment(segment);
+ sizeOfLoadingSegments -= segment.getSize();
+ }
} else {
projectedSegmentCounts.addSegment(segment);
if (action == SegmentAction.DROP) {
@@ -542,6 +552,21 @@ public class ServerHolder implements
Comparable<ServerHolder>
}
}
+ /**
+ * Upper bound of bytes an in-place reload of {@code segment} can still add
to this server, i.e. a load queued on a
+ * server that is already serving it. This is an upper bound, not the exact
delta, which the coordinator cannot know
+ * until the historical announces the footprint it realized.
+ * <p>
+ * Both the loaded set and the announced profile come from this run's
immutable server snapshot, so the value is
+ * stable across the {@link #addToQueuedSegments} / {@link
#removeFromQueuedSegments} pair for a canceled operation.
+ */
+ private long inPlaceReloadSizeDelta(DataSegment segment)
+ {
+ final PartialLoadProfile loaded =
server.getPartialLoadProfile(segment.getId());
+ final Long loadedBytes = loaded == null ? null : loaded.loadedBytes();
+ return loadedBytes == null ? 0L : Math.max(0L, segment.getSize() -
loadedBytes);
+ }
+
@Override
public int compareTo(ServerHolder serverHolder)
{
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/loading/PartialSegmentStatusInTier.java
b/server/src/main/java/org/apache/druid/server/coordinator/loading/PartialSegmentStatusInTier.java
index 84dd10f3dd4..22aa4251c97 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/loading/PartialSegmentStatusInTier.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/loading/PartialSegmentStatusInTier.java
@@ -41,13 +41,16 @@ import java.util.Objects;
* fingerprint equals the requested fingerprint; rule is satisfied for that
replica) vs "stale" (any other state,
* including a non-profile regular full-load replica).
* <p>
- * As a last resort, when a stale replica has nowhere better to be replaced,
this classifies the same server as a
- * target for an "additive-historical" in-place replace: a partial-load
request arriving at a server that's already
- * (stale-)loaded fills in the missing parts in place rather than
re-downloading from scratch. That's what makes
- * {@link #getEligibleForAdditiveReload()} a safe fallback destination when
the tier has no spare capacity. This
- * option is the least preferred because the contract of an additive reload is
to load only what is now needed and
- * missing; it does not drop anything that is no longer needed, so the server
can end up holding a larger amount of
- * data than the current rule strictly requires.
+ * A stale replica is also classified as a target for an in-place reload: a
partial-load request arriving at a server
+ * that is already (stale-)loaded is honored by swapping the rule on the cache
entry it already has, so only the delta
+ * the new fingerprint adds has to come off deep storage. That makes {@link
#getEligibleForInPlaceReload()} the
+ * <em>preferred</em> destination for a matching deficit as it downloads
strictly less than a fresh load elsewhere,
+ * the replica keeps serving throughout, and no follow-up drop is needed to
retire the replica it replaces.
+ * <p>
+ * What makes this safe is that the historical pins a rule with cache holds
rather than accumulating. Applying a rule
+ * releases the holds on every bundle the new fingerprint does not select, so
a reloaded server ends up pinned by
+ * exactly the new rule; whatever the previous rule left on disk is ordinary
evictable cache, reclaimed under pressure
+ * like any other unheld data, and it is not part of the footprint the server
announces back.
*/
public class PartialSegmentStatusInTier
{
@@ -56,7 +59,7 @@ public class PartialSegmentStatusInTier
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<>();
+ private final List<ServerHolder> eligibleForInPlaceReload = new
ArrayList<>();
public PartialSegmentStatusInTier(
DataSegment segment,
@@ -80,8 +83,8 @@ public class PartialSegmentStatusInTier
/**
* 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.
+ * i.e. a regular full-load replica seen against a partial rule). Eligible
for an in-place reload onto the new
+ * fingerprint, and for being dropped once enough matching replicas exist.
*/
public List<ServerHolder> getStaleLoaded()
{
@@ -119,19 +122,19 @@ public class PartialSegmentStatusInTier
}
/**
- * Stale-loaded servers that can take an additive reload request; kept as a
subset of {@link #getStaleLoaded()}
- * filtered for decommissioning / load-queue-full, so the algorithm can
target them as a fallback destination when
- * no fresh-load slots are available. See {@code
StrategicSegmentAssigner.updateReplicasInTierPartial}.
+ * Stale-loaded servers that can take an in-place reload request: the subset
of {@link #getStaleLoaded()} that passes
+ * {@link #canReloadInPlace}. The preferred destination for a matching
deficit, ahead of
+ * {@link #getEligibleForFreshLoad()}. See {@link
StrategicSegmentAssigner#updateReplicasInTierPartial}.
*/
- public List<ServerHolder> getEligibleForAdditiveReload()
+ public List<ServerHolder> getEligibleForInPlaceReload()
{
- return eligibleForAdditiveReload;
+ return eligibleForInPlaceReload;
}
/**
* Mechanical classification of one server against the request fingerprint.
Branches are mutually exclusive in
* order: <b>loaded</b> ({@link ServerHolder#isServingSegment}: matching /
stale, with stale optionally also added
- * to {@link #eligibleForAdditiveReload}), <b>in-flight
LOAD/REPLICATE/MOVE_TO</b> (matching / stale based on the
+ * to {@link #eligibleForInPlaceReload}), <b>in-flight
LOAD/REPLICATE/MOVE_TO</b> (matching / stale based on the
* peon's queued profile), <b>empty-and-loadable</b> ({@link
#eligibleForFreshLoad}).
* <p>
* A balancer move is counted at its destination: the {@link
SegmentAction#MOVE_TO} carries the profile cloned from
@@ -146,7 +149,7 @@ public class PartialSegmentStatusInTier
* Servers with a queued {@link SegmentAction#DROP} fall through all
branches as well, they're accounted for in
* {@link SegmentReplicaCount} totals and {@link StrategicSegmentAssigner}'s
cross-tier drop budget.
* The {@code isLoaded} branch is gated by {@link
ServerHolder#isServingSegment}, which requires <em>no</em> action
- * queued, so stale-loaded servers added to {@link
#eligibleForAdditiveReload} are guaranteed to be action-free at
+ * queued, so stale-loaded servers added to {@link
#eligibleForInPlaceReload} are guaranteed to be action-free at
* snapshot time.
*/
private void classify(ServerHolder server, DataSegment segment, String
requestedFingerprint)
@@ -160,8 +163,8 @@ public class PartialSegmentStatusInTier
matchingLoaded.add(server);
} else {
staleLoaded.add(server);
- if (canReloadAdditively(server)) {
- eligibleForAdditiveReload.add(server);
+ if (canReloadInPlace(server)) {
+ eligibleForInPlaceReload.add(server);
}
}
} else if (action == SegmentAction.LOAD
@@ -179,18 +182,21 @@ public class PartialSegmentStatusInTier
}
/**
- * Filters a stale-loaded server for additive-reload eligibility: not
decommissioning, and not over its per-run
- * load-queue budget. The "no other action queued" requirement that you'd
otherwise expect to find here is
- * already satisfied implicitly, this is only called from the {@code
isLoaded} branch of {@link #classify}, which
- * requires {@link ServerHolder#isServingSegment} = true (loaded AND no
queued action). Same-run dedup against
- * subsequent re-queueing on the same server is enforced at {@link
ServerHolder#startOperation}, not here.
+ * Whether a server that already serves the segment can take an in-place
reload of it: not decommissioning, and not
+ * over its per-run load-queue budget. Callers are responsible for
establishing that the server actually serves the
+ * segment with no other action queued ({@link
ServerHolder#isServingSegment}); {@link #classify} gets that from the
+ * {@code isLoaded} branch it calls this from. Same-run dedup against
subsequent re-queueing on the same server is
+ * enforced at {@link ServerHolder#startOperation}, not here.
+ * <p>
+ * {@link ServerHolder#canLoadSegment} is not usable in its place because it
requires the server to <em>not</em>
+ * already have the segment, which is precisely the case being handled here.
* <p>
- * Disk space is not checked: the additive reload's marginal cost is at most
- * {@code segment.size − alreadyLoadedSize}, and a strict full-size disk
check would over-conservatively block
- * reloads on near-full servers that already host the stale replica. If the
historical is too full to add the
- * missing parts, the load fails at the historical and reports as failed;
the reconciler retries next run.
+ * Disk space is not checked: the reload's marginal cost is at most {@code
segment.size − alreadyLoadedSize}, and a
+ * strict full-size disk check would over-conservatively block reloads on
near-full servers that already host the
+ * stale replica. If the historical is too full to add the missing parts,
the load fails at the historical and
+ * reports as failed; the reconciler retries next run.
*/
- private static boolean canReloadAdditively(ServerHolder server)
+ public static boolean canReloadInPlace(ServerHolder server)
{
return !server.isDecommissioning() && !server.isLoadQueueFull();
}
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java
b/server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java
index 89e51f99660..8691b34fe5e 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java
@@ -19,7 +19,6 @@
package org.apache.druid.server.coordinator.loading;
-import com.google.common.collect.Iterators;
import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
import org.apache.druid.client.DruidServer;
import org.apache.druid.server.coordinator.DruidCluster;
@@ -346,7 +345,7 @@ public class StrategicSegmentAssigner implements
SegmentActionHandler
* <h3>Algorithm</h3>
* <ol>
* <li><b>Classify.</b> Build {@link PartialSegmentStatusInTier} for this
tier; every server falls into at most
- * one of: matching-loaded, stale-loaded (optionally also
eligible-for-additive-reload),
+ * one of: matching-loaded, stale-loaded (optionally also
eligible-for-in-place-reload),
* matching-in-flight, stale-in-flight, eligible-for-fresh-load, or
unclassified (drop or move source
* pending; see {@link PartialSegmentStatusInTier#classify} for why).
Matching means the announced
* fingerprint equals this request's fingerprint; stale is anything
else, including a non-profile regular
@@ -356,17 +355,18 @@ public class StrategicSegmentAssigner implements
SegmentActionHandler
* needed, because the classification never counts both endpoints of a
move.</li>
* <li><b>If matching count is short of {@code requiredReplicas}</b>
(deficit):
* <ol type="a">
- * <li>Cancel stale-in-flight loads to free their slots. Canceled
servers become same-run fresh-load
- * destinations.</li>
- * <li>Queue fresh partial-load requests up to the deficit.
Destination preference order, applied in
+ * <li>Cancel stale-in-flight loads to free their slots. Canceled
servers become same-run load destinations.
+ * </li>
+ * <li>Queue matching partial-load requests up to the deficit.
Destination preference order, applied in
* {@link #loadPartialReplicas}:
* <ol>
- * <li>Empty servers (clean slate, no in-place mutation
needed).</li>
- * <li>Servers whose stale-in-flight load we just canceled in
(a), their slot is now free.</li>
- * <li>Stale-loaded servers eligible for additive reload (the
historical fills in the missing parts in
- * place). This is the fallback path that mitigates the "no
spare server" stuck state.
+ * <li>Stale-loaded servers eligible for an in-place reload: the
historical swaps the rule on the cache
+ * entry it already has, so only the delta the new
fingerprint adds is downloaded, the replica
+ * keeps serving throughout, and the stale replica retires
itself instead of needing a drop.
* Same-run dedup is enforced by {@link
ServerHolder#startOperation}, which rejects a second
* queue attempt on a server whose segment is already
queued.</li>
+ * <li>Empty servers, and servers whose stale-in-flight load we
just canceled in (a), ordered by the
+ * balancer strategy. These download the whole request from
deep storage.</li>
* </ol>
* </li>
* </ol>
@@ -377,7 +377,9 @@ public class StrategicSegmentAssigner implements
SegmentActionHandler
* already meets the requirement. This preserves availability across
the swap: stale replicas keep serving
* until matching replicas have completed loading and announced, then
get dropped. The
* {@code maxReplicasToDrop} budget caps how many drops we queue per
coordinator run to avoid drop
- * storms.</li>
+ * storms. A stale replica this run just queued an in-place reload on
is never dropped here even though the
+ * classification snapshot still lists it as stale-loaded: reloads are
only queued under a deficit, and a
+ * deficit means matching-loaded is below the requirement, which is
exactly what this gate tests.</li>
* </ol>
*
* <h3>Returns</h3>
@@ -490,16 +492,30 @@ public class StrategicSegmentAssigner implements
SegmentActionHandler
}
/**
- * Queues fresh partial-load requests on up to {@code numToLoad} eligible
servers. Preference order: servers that can
- * take a fresh load, then stale-loaded servers (additive reload; the
historical fills missing parts in place).
+ * Queues matching partial-load requests on up to {@code numToLoad} eligible
servers, preferring in-place reloads
+ * over fresh loads.
* <p>
- * The fresh-load candidates are the classifier's empty servers together
with {@code canceledStaleServers}.
- * {@link ServerHolder#cancelOperation} clears the queued action and
restores the projected size, so a server whose
- * stale in-flight load {@link #cancelLoadsOnServers} canceled can take a
fresh load. {@link #serversToLoadSegment}
- * returns an iterator over fresh load candidates.
+ * A stale-loaded server already holds the segment's metadata plus whatever
the previous rule pinned, and the
+ * historical honors a partial-load request there by swapping the rule on
that cache entry: only the delta the new
+ * fingerprint adds comes off deep storage, the replica keeps serving
throughout, and the replica it replaces is
+ * itself, so no follow-up drop is needed.
* <p>
- * An iterator over candidates that can additive reload the segment is there
for backup in case we can't fully
- * replicate on our priority one fresh load path.
+ * The fresh-load candidates are the classifier's empty servers together
with the {@code canceledStaleServers} that
+ * are left empty by the cancellation. {@link ServerHolder#cancelOperation}
clears the queued action and restores the
+ * projected size, so those can take a fresh load; {@link
#serversToLoadSegment} orders them by the balancer strategy
+ * (or round robin).
+ * <p>
+ * A reload the historical <em>fails</em> is retried in place on later runs
rather than handed to a fresh server: the
+ * failure is asynchronous, so this run only ever learns that the request
was queued, and the replica stays serving
+ * under its previous profile either way (see
+ * {@link
org.apache.druid.server.coordination.SegmentLoadDropHandler#addSegment}), so
the tier is never short a
+ * copy while it retries. Repeated failure does not need a fallback to break
out of, because the way for one to
+ * persist is a historical too full to pin the new bundles, and that
resolves itself: disk-usage balancing moves
+ * segments off the fullest server in the tier
+ * ({@code SegmentToMoveCalculator.computeSegmentsToMoveToBalanceDiskUsage})
until the reload fits, and a tier with
+ * nowhere left to move to is a cluster-wide capacity problem stalling far
more than this one segment. Every failed
+ * attempt is alerted by the historical and logged by {@code
HttpLoadQueuePeon.onRequestFailed}, so a retry loop
+ * that is not making progress is visible rather than silent.
*/
private int loadPartialReplicas(
int numToLoad,
@@ -517,25 +533,68 @@ public class StrategicSegmentAssigner implements
SegmentActionHandler
return 0;
}
- // The classifier's list is already the complete candidate set when
nothing was canceled.
+ // The classifier's lists are already the complete candidate sets when
nothing was canceled.
+ final List<ServerHolder> inPlaceDestinations;
final List<ServerHolder> freshCandidates;
if (canceledStaleServers.isEmpty()) {
+ inPlaceDestinations = status.getEligibleForInPlaceReload();
freshCandidates = status.getEligibleForFreshLoad();
} else {
+ inPlaceDestinations = new
ArrayList<>(status.getEligibleForInPlaceReload());
freshCandidates = new ArrayList<>(status.getEligibleForFreshLoad());
- freshCandidates.addAll(canceledStaleServers);
+ for (ServerHolder server : canceledStaleServers) {
+ if (server.isServingSegment(segment) &&
PartialSegmentStatusInTier.canReloadInPlace(server)) {
+ inPlaceDestinations.add(server);
+ } else {
+ freshCandidates.add(server);
+ }
+ }
}
- final Iterator<ServerHolder> destinations = Iterators.concat(
- serversToLoadSegment(segment, tier, freshCandidates),
- status.getEligibleForAdditiveReload().iterator()
+ int numLoadsQueued = queuePartialLoads(
+ numToLoad,
+ segment,
+ inPlaceDestinations.iterator(),
+ isAlreadyLoadedOnTier,
+ profile
);
+ if (numLoadsQueued >= numToLoad) {
+ return numLoadsQueued;
+ }
- if (!destinations.hasNext()) {
+ // Built only once the in-place reloads have fallen short:
RoundRobinServerSelector advances its per-tier cursor
+ // past ineligible servers as soon as the iterator is constructed, so an
unused one still perturbs placement.
+ final Iterator<ServerHolder> freshDestinations =
serversToLoadSegment(segment, tier, freshCandidates);
+ if (inPlaceDestinations.isEmpty() && !freshDestinations.hasNext()) {
incrementSkipStat(Stats.Segments.ASSIGN_SKIPPED, "No eligible server",
segment, tier);
return 0;
}
+ return numLoadsQueued + queuePartialLoads(
+ numToLoad - numLoadsQueued,
+ segment,
+ freshDestinations,
+ isAlreadyLoadedOnTier,
+ profile
+ );
+ }
+
+ /**
+ * Queues a partial-load request carrying {@code profile} on up to {@code
numToLoad} of the given destinations,
+ * stopping as soon as the deficit is covered. Servers that refuse the queue
attempt (throttled, or a peon error)
+ * don't count towards {@code numToLoad}, so the next destination is tried
in their place.
+ * <p>
+ * {@code destinations} is consumed lazily, {@link #serversToLoadSegment}
returns iterators whose traversal has
+ * placement side effects.
+ */
+ private int queuePartialLoads(
+ int numToLoad,
+ DataSegment segment,
+ Iterator<ServerHolder> destinations,
+ boolean isAlreadyLoadedOnTier,
+ PartialLoadProfile profile
+ )
+ {
int numLoadsQueued = 0;
while (numLoadsQueued < numToLoad && destinations.hasNext()) {
final ServerHolder server = destinations.next();
@@ -748,9 +807,8 @@ public class StrategicSegmentAssigner implements
SegmentActionHandler
* <li>their load queue is already at the configured {@code
maxSegmentsInNodeLoadingQueue} budget for this run.</li>
* <li>they are decommissioning, since their replicas are on the way out
and reloading them is wasted work.</li>
* </ul>
- * These are the same two eligibility conditions {@code
PartialSegmentStatusInTier.canReloadAdditively} applies to
- * the partial-load reconciler's in-place reload. {@link
ServerHolder#canLoadSegment} is not usable here because it
- * requires the server to <em>not</em> already have the segment, which is
precisely the case being handled.
+ * The last two are {@link PartialSegmentStatusInTier#canReloadInPlace},
shared with the partial-load reconciler's
+ * in-place reload.
*/
private int revertPartialProfileReplicas(DataSegment segment, String tier)
{
@@ -771,8 +829,7 @@ public class StrategicSegmentAssigner implements
SegmentActionHandler
private boolean revertPartialProfileReplica(DataSegment segment,
ServerHolder server)
{
return server.isServingSegment(segment)
- && !server.isDecommissioning()
- && !server.isLoadQueueFull()
+ && PartialSegmentStatusInTier.canReloadInPlace(server)
&& server.getServer().getPartialLoadProfile(segment.getId()) != null
&& loadQueueManager.loadSegment(segment, server,
SegmentAction.LOAD, null);
}
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/NoopSegmentCacheManager.java
b/server/src/test/java/org/apache/druid/segment/loading/NoopSegmentCacheManager.java
index 2d9cfc39de3..b45ded107f4 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/NoopSegmentCacheManager.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/NoopSegmentCacheManager.java
@@ -71,7 +71,7 @@ public class NoopSegmentCacheManager implements
SegmentCacheManager
}
@Override
- public DataSegment load(DataSegment segment)
+ public DataSegment load(DataSegment segment) throws SegmentLoadingException
{
throw new UnsupportedOperationException();
}
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialRuleLoadTest.java
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialRuleLoadTest.java
index b3be2adaa37..587f5280a38 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialRuleLoadTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialRuleLoadTest.java
@@ -67,6 +67,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import javax.annotation.Nullable;
import java.io.File;
import java.io.IOException;
import java.util.Arrays;
@@ -754,6 +755,39 @@ class SegmentLocalCacheManagerPartialRuleLoadTest
Assertions.assertNull(manager.getRuleFingerprintForSegment(SEGMENT_ID));
}
+ @Test
+ void testRangeReaderNullDoesNotDisturbANonPartialEntry() throws Exception
+ {
+ // A non-partial cache entry sits at this segment id and the coordinator
asks for a rule this historical cannot
+ // honor. Giving up on the rule must leave that entry alone: it is not
going to be replaced by a partial one, so
+ // there is nothing to clear out of the way, and discarding its cached
data would buy nothing. Evicting it used
+ // to be attempted before the range-reader check, which both destroyed
cache for nothing when the entry was
+ // unheld and failed the load outright when it was held, as it is here.
+ manager = makeManager(true, true);
+ final StorageLocation location = manager.getLocations().get(0);
+ final SegmentCacheEntryIdentifier id = new
SegmentCacheEntryIdentifier(SEGMENT_ID);
+
+ final DataSegment noRangeReader =
+ partialWrapperSegmentWithNullRangeReader(List.of(AGG_BUNDLE),
"v1:cannot-honor");
+ // An on-demand acquire of a segment whose load spec cannot range-read
registers a non-partial entry, held for
+ // as long as the action is open.
+ final AcquireSegmentAction inFlightQuery =
manager.acquireSegment(noRangeReader, AcquireMode.PARTIAL);
+ try {
+ Assertions.assertNotNull(
+ location.getCacheEntry(id),
+ "precondition: an in-flight on-demand acquire holds a non-partial
entry"
+ );
+
+ manager.load(noRangeReader);
+
+ Assertions.assertNotNull(location.getCacheEntry(id), "the non-partial
entry is untouched");
+ Assertions.assertNull(manager.getRuleFingerprintForSegment(SEGMENT_ID),
"and no rule is applied");
+ }
+ finally {
+ inFlightQuery.close();
+ }
+ }
+
@Test
void testLoadFailureLeavesNoRuleApplied() throws Exception
{
@@ -781,6 +815,117 @@ class SegmentLocalCacheManagerPartialRuleLoadTest
);
}
+ @Test
+ void testFailedEagerDownloadRestoresThePriorRule() throws Exception
+ {
+ // A reload whose eager downloads fail must put the PRIOR rule back rather
than clear the rule outright. The
+ // historical keeps serving the replica and keeps announcing the prior
profile (the failed load never announces a
+ // new one), so the coordinator has to still find that profile's bundles
pinned; clearing would leave the replica
+ // advertising a footprint it no longer holds, with every bundle of it
evictable.
+ final StorageLoadingThreadPool loadingPool =
StorageLoadingThreadPool.createFromConfig(
+ SegmentLoaderConfig.builder()
+ .locations(List.of(new
StorageLocationConfig(cacheRoot, 1024L * 1024L * 1024L, null)))
+ .virtualStorage(true)
+ .virtualStoragePartialDownloadsEnabled(true)
+ .build()
+ );
+ manager = makeManagerAtLocations(true, true, List.of(cacheRoot),
loadingPool);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(partialWrapperSegment(List.of(AGG_BUNDLE),
"v1:rule-original"));
+ final PartialSegmentMetadataCacheEntry meta =
weakReservedMetadata(location, SEGMENT_ID);
+ Assertions.assertEquals("v1:rule-original", meta.getRuleFingerprint());
+ Assertions.assertTrue(meta.isBundleRuleHeld(AGG_BUNDLE));
+
+ // Stopping the loading pool makes every eager download for the new rule's
bundle fail to even be submitted.
+ loadingPool.stop();
+
+ Assertions.assertThrows(
+ SegmentLoadingException.class,
+ () -> manager.load(partialWrapperSegment(List.of(OTHER_AGG_BUNDLE),
"v2:rule-updated"))
+ );
+
+ Assertions.assertEquals(
+ "v1:rule-original",
+ manager.getRuleFingerprintForSegment(SEGMENT_ID),
+ "a failed reload must restore the prior rule's fingerprint, not clear
it"
+ );
+ Assertions.assertTrue(
+ meta.isBundleRuleHeld(AGG_BUNDLE),
+ "the prior rule's bundle must still be pinned after the failed reload"
+ );
+ Assertions.assertTrue(
+ location.isWeakReserved(new
PartialSegmentBundleCacheEntryIdentifier(SEGMENT_ID, AGG_BUNDLE)),
+ "the prior rule's bundle must never have been unreserved during the
attempt, so it was never evictable"
+ );
+ Assertions.assertFalse(
+ meta.isBundleRuleHeld(OTHER_AGG_BUNDLE),
+ "the attempted rule's bundle must not be left pinned"
+ );
+ }
+
+ @Test
+ void testRestartAfterFailedReloadReappliesThePriorRuleNotTheFailedOne()
throws Exception
+ {
+ // The durable half of the rollback. A failed reload restores the prior
rule in memory and the replica keeps
+ // announcing it, so the info file has to still describe that rule too. If
it described the rule that just
+ // failed, a restart would reapply it from disk, and a second failure
there takes the replica down entirely:
+ // SegmentManager.loadSegmentOnBootstrap drops on failure and the segment
never gets announced.
+ final StorageLoadingThreadPool loadingPool =
StorageLoadingThreadPool.createFromConfig(
+ SegmentLoaderConfig.builder()
+ .locations(List.of(new
StorageLocationConfig(cacheRoot, 1024L * 1024L * 1024L, null)))
+ .virtualStorage(true)
+ .virtualStoragePartialDownloadsEnabled(true)
+ .build()
+ );
+ final SegmentLocalCacheManager beforeRestart =
+ makeManagerAtLocations(true, true, List.of(cacheRoot), loadingPool);
+ try {
+ beforeRestart.load(partialWrapperSegment(List.of(AGG_BUNDLE),
"v1:rule-original"));
+ Assertions.assertEquals("v1:rule-original",
beforeRestart.getRuleFingerprintForSegment(SEGMENT_ID));
+
+ loadingPool.stop();
+ Assertions.assertThrows(
+ SegmentLoadingException.class,
+ () ->
beforeRestart.load(partialWrapperSegment(List.of(OTHER_AGG_BUNDLE),
"v2:rule-updated"))
+ );
+ Assertions.assertEquals(
+ "v1:rule-original",
+ beforeRestart.getRuleFingerprintForSegment(SEGMENT_ID),
+ "precondition: the failed reload rolled the in-memory rule back"
+ );
+ }
+ finally {
+ beforeRestart.shutdown();
+ }
+
+ // Restart over the same cache directory.
+ manager = makeManager(true, true);
+ final List<DataSegment> cached = manager.getCachedSegments();
+ final DataSegment rediscovered = cached.stream()
+ .filter(s ->
s.getId().equals(SEGMENT_ID))
+ .findFirst()
+ .orElseThrow();
+ Assertions.assertEquals(
+ "v1:rule-original",
+ rediscovered.getLoadSpec().get("fingerprint"),
+ "the persisted load spec must describe the rule the replica is serving
under, not the failed one"
+ );
+
+ manager.bootstrap(rediscovered, SegmentLazyLoadFailCallback.NOOP);
+ Assertions.assertEquals(
+ "v1:rule-original",
+ manager.getRuleFingerprintForSegment(SEGMENT_ID),
+ "bootstrap must reapply the prior rule, not the one that failed"
+ );
+ Assertions.assertTrue(
+ manager.getLocations().get(0).isWeakReserved(
+ new PartialSegmentBundleCacheEntryIdentifier(SEGMENT_ID,
AGG_BUNDLE)
+ ),
+ "and pin that rule's bundle"
+ );
+ }
+
@Test
void testBootstrapReinstallsRuleHoldsFromPersistedInfoFile() throws Exception
{
@@ -937,6 +1082,20 @@ class SegmentLocalCacheManagerPartialRuleLoadTest
boolean partialDownloadsEnabled,
List<File> locationRoots
)
+ {
+ return makeManagerAtLocations(virtualStorage, partialDownloadsEnabled,
locationRoots, null);
+ }
+
+ /**
+ * @param loadingPool the loading pool to use, or null to build one from the
config. Pass one in to keep a handle on
+ * it, e.g. to {@link StorageLoadingThreadPool#stop()} it
mid-test and make eager downloads fail.
+ */
+ private SegmentLocalCacheManager makeManagerAtLocations(
+ boolean virtualStorage,
+ boolean partialDownloadsEnabled,
+ List<File> locationRoots,
+ @Nullable StorageLoadingThreadPool loadingPool
+ )
{
final List<StorageLocationConfig> locConfigs = locationRoots.stream()
.map(root -> new StorageLocationConfig(root, 1024L * 1024L * 1024L,
null))
@@ -950,7 +1109,7 @@ class SegmentLocalCacheManagerPartialRuleLoadTest
return new SegmentLocalCacheManager(
storageLocations,
loaderConfig,
- StorageLoadingThreadPool.createFromConfig(loaderConfig),
+ loadingPool == null ?
StorageLoadingThreadPool.createFromConfig(loaderConfig) : loadingPool,
new LeastBytesUsedStorageLocationSelectorStrategy(storageLocations),
TestHelper.getTestIndexIO(jsonMapper, ColumnConfig.DEFAULT),
jsonMapper
diff --git
a/server/src/test/java/org/apache/druid/server/SegmentManagerTest.java
b/server/src/test/java/org/apache/druid/server/SegmentManagerTest.java
index 9a06d4d3a79..0b8a908547d 100644
--- a/server/src/test/java/org/apache/druid/server/SegmentManagerTest.java
+++ b/server/src/test/java/org/apache/druid/server/SegmentManagerTest.java
@@ -55,6 +55,7 @@ import org.apache.druid.segment.loading.StorageLocation;
import org.apache.druid.segment.loading.StorageLocationConfig;
import org.apache.druid.server.SegmentManager.DataSourceState;
import org.apache.druid.server.metrics.NoopServiceEmitter;
+import org.apache.druid.test.utils.TestSegmentCacheManager;
import org.apache.druid.testing.InitializedNullHandlingTest;
import org.apache.druid.testing.TemporaryFolderExtension;
import org.apache.druid.timeline.DataSegment;
@@ -309,6 +310,43 @@ public class SegmentManagerTest extends
InitializedNullHandlingTest
assertResult(SEGMENTS);
}
+ @Test
+ public void testFailedReloadDoesNotDropTheCachedSegment() throws
SegmentLoadingException, IOException
+ {
+ final TestSegmentCacheManager failingCacheManager = new
TestSegmentCacheManager();
+ failingCacheManager.failLoadsAfter(1);
+ final SegmentManager manager = new SegmentManager(failingCacheManager);
+ final DataSegment segment = SEGMENTS.get(0);
+
+ manager.loadSegment(segment);
+ Assertions.assertTrue(manager.isSegmentLoaded(segment));
+
+ Assertions.assertThrows(SegmentLoadingException.class, () ->
manager.loadSegment(segment));
+
+ Assertions.assertTrue(manager.isSegmentLoaded(segment), "the replica stays
in the timeline");
+ Assertions.assertFalse(
+
failingCacheManager.getObservedSegmentsRemovedFromCache().contains(segment.getId()),
+ "a failed reload must not drop the live replica's cached data"
+ );
+ }
+
+ @Test
+ public void testFailedFirstLoadLeavesCleanupToTheCaller()
+ {
+ final TestSegmentCacheManager failingCacheManager = new
TestSegmentCacheManager();
+ failingCacheManager.failLoadsAfter(0);
+ final SegmentManager manager = new SegmentManager(failingCacheManager);
+ final DataSegment segment = SEGMENTS.get(0);
+
+ Assertions.assertThrows(SegmentLoadingException.class, () ->
manager.loadSegment(segment));
+
+ Assertions.assertFalse(manager.isSegmentLoaded(segment), "a failed load
adds nothing to the timeline");
+ Assertions.assertFalse(
+
failingCacheManager.getObservedSegmentsRemovedFromCache().contains(segment.getId()),
+ "and drops nothing either; SegmentLoadDropHandler.addSegment owns that
decision"
+ );
+ }
+
@Test
public void testLoadDuplicatedSegmentsInParallel()
throws ExecutionException, InterruptedException
diff --git
a/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java
b/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java
index cf05c9a75da..13ae7ee19d3 100644
---
a/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java
@@ -187,6 +187,96 @@ public class SegmentLoadDropHandlerTest
);
}
+ @Test
+ public void testFailedReloadKeepsTheSegmentServing()
+ {
+ final TestSegmentCacheManager cacheManager = new TestSegmentCacheManager();
+ cacheManager.failLoadsAfter(1);
+ final SegmentManager segmentManager = new SegmentManager(cacheManager);
+ final SegmentLoadDropHandler handler =
initSegmentLoadDropHandler(segmentManager);
+
+ final DataSegment segment = makeSegment("test", "1",
Intervals.of("P1d/2011-04-01"));
+
+ handler.addSegment(segment, DataSegmentChangeCallback.NOOP, null);
+ Assertions.assertTrue(segmentManager.isSegmentLoaded(segment));
+
Assertions.assertTrue(segmentAnnouncer.getObservedSegments().contains(segment));
+
+ // Second request for the same segment is a reload, and this cache manager
fails it.
+ handler.addSegment(segment, DataSegmentChangeCallback.NOOP, null);
+ for (Runnable runnable : scheduledRunnable) {
+ runnable.run();
+ }
+
+ Assertions.assertTrue(
+ segmentManager.isSegmentLoaded(segment),
+ "a failed reload must leave the replica in the timeline, still
queryable"
+ );
+ Assertions.assertTrue(
+ segmentAnnouncer.getObservedSegments().contains(segment),
+ "a failed reload must not unannounce the replica, the coordinator
still counts it and asks again"
+ );
+ Assertions.assertFalse(
+
cacheManager.getObservedSegmentsRemovedFromCache().contains(segment.getId()),
+ "a failed reload must not drop the replica's cached data"
+ );
+ }
+
+ @Test
+ public void testFailedFirstLoadIsStillCleanedUp()
+ {
+ final TestSegmentCacheManager cacheManager = new TestSegmentCacheManager();
+ cacheManager.failLoadsAfter(0);
+ final SegmentManager segmentManager = new SegmentManager(cacheManager);
+ final SegmentLoadDropHandler handler =
initSegmentLoadDropHandler(segmentManager);
+
+ final DataSegment segment = makeSegment("test", "1",
Intervals.of("P1d/2011-04-01"));
+
+ handler.addSegment(segment, DataSegmentChangeCallback.NOOP, null);
+ for (Runnable runnable : scheduledRunnable) {
+ runnable.run();
+ }
+
+ Assertions.assertFalse(segmentManager.isSegmentLoaded(segment));
+
Assertions.assertFalse(segmentAnnouncer.getObservedSegments().contains(segment));
+ Assertions.assertTrue(
+
cacheManager.getObservedSegmentsRemovedFromCache().contains(segment.getId()),
+ "a failed new load is cleaned up"
+ );
+ }
+
+ @Test
+ public void testFailedLoadPreservesAReplicaThatAppearedWhileItWasLoading()
+ {
+ final TestSegmentCacheManager cacheManager = new TestSegmentCacheManager();
+ final SegmentManager segmentManager = new SegmentManager(cacheManager);
+ final SegmentLoadDropHandler handler =
initSegmentLoadDropHandler(segmentManager);
+
+ final DataSegment segment = makeSegment("test", "1",
Intervals.of("P1d/2011-04-01"));
+
+ // Stand in for the concurrent loader: the segment lands in the timeline,
then this load fails. A snapshot taken
+ // before the load would have said "not serving" and torn the replica down.
+ cacheManager.failLoadsAfter(1);
+ handler.addSegment(segment, DataSegmentChangeCallback.NOOP, null);
+ Assertions.assertTrue(segmentManager.isSegmentLoaded(segment),
"precondition: the replica is serving");
+ segmentAnnouncer.getObservedSegments().clear();
+ segmentAnnouncer.announceSegment(segment);
+
+ handler.addSegment(segment, DataSegmentChangeCallback.NOOP, null);
+ for (Runnable runnable : scheduledRunnable) {
+ runnable.run();
+ }
+
+ Assertions.assertTrue(segmentManager.isSegmentLoaded(segment), "the
serving replica stays in the timeline");
+ Assertions.assertTrue(
+ segmentAnnouncer.getObservedSegments().contains(segment),
+ "and stays announced"
+ );
+ Assertions.assertFalse(
+
cacheManager.getObservedSegmentsRemovedFromCache().contains(segment.getId()),
+ "and keeps its cached data"
+ );
+ }
+
@Test
@Timeout(value = 60_000L, unit = TimeUnit.MILLISECONDS, threadMode =
ThreadMode.SEPARATE_THREAD)
public void testProcessBatch() throws Exception
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/ServerHolderTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/ServerHolderTest.java
index 3375da17882..833f7aa0ad9 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/ServerHolderTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/ServerHolderTest.java
@@ -21,17 +21,22 @@ package org.apache.druid.server.coordinator;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
+import org.apache.druid.client.DruidServer;
import org.apache.druid.client.ImmutableDruidDataSource;
import org.apache.druid.client.ImmutableDruidServer;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.server.coordination.DruidServerMetadata;
import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.coordinator.loading.PartialLoadProfile;
+import org.apache.druid.server.coordinator.loading.SegmentAction;
+import org.apache.druid.server.coordinator.loading.SegmentLoadQueueManager;
import org.apache.druid.server.coordinator.loading.TestLoadQueuePeon;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.partition.NoneShardSpec;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import javax.annotation.Nullable;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -68,6 +73,19 @@ public class ServerHolderTest
"src2", new ImmutableDruidDataSource("src2", Collections.emptyMap(),
Collections.singletonList(SEGMENTS.get(1)))
);
+ private static final long SEGMENT_SIZE = 1000L;
+
+ private static final Map<String, Object> PARTIAL_LOAD_SPEC =
+ ImmutableMap.of("type", "partialProjection", "fingerprint", "v1:abc");
+
+ /**
+ * Non-zero sized counterparts of {@link #SEGMENTS}, for the projection
accounting tests.
+ */
+ private static final List<DataSegment> SIZED_SEGMENTS = ImmutableList.of(
+ DataSegment.builder(SEGMENTS.get(0)).size(SEGMENT_SIZE).build(),
+ DataSegment.builder(SEGMENTS.get(1)).size(SEGMENT_SIZE).build()
+ );
+
@Test
public void testCompareTo()
{
@@ -195,4 +213,89 @@ public class ServerHolderTest
Assertions.assertFalse(h1.isServingSegment(SEGMENTS.get(1)));
Assertions.assertFalse(h1.isLoadQueueFull());
}
+
+ @Test
+ public void testLoadOfAbsentSegmentProjectsItsFullSize()
+ {
+ final ServerHolder holder = holderServing(null);
+ final long sizeUsedBefore = holder.getSizeUsed();
+ final int projectedCountBefore =
holder.getProjectedSegmentCounts().getTotalSegmentCount();
+
+ Assertions.assertTrue(holder.startOperation(SegmentAction.LOAD,
SIZED_SEGMENTS.get(1)));
+
+ Assertions.assertEquals(sizeUsedBefore + SEGMENT_SIZE,
holder.getSizeUsed());
+ Assertions.assertEquals(
+ projectedCountBefore + 1,
+ holder.getProjectedSegmentCounts().getTotalSegmentCount()
+ );
+ }
+
+ @Test
+ public void testInPlaceReloadOfPartialReplicaProjectsOnlyTheDelta()
+ {
+ // A partial replica announces its realized footprint as curr_size, so
reloading it in place can add at most the
+ // rest of the segment. Counting the whole segment again would double
count the bytes already on disk.
+ final ServerHolder holder =
holderServing(PartialLoadProfile.forLoaded(PARTIAL_LOAD_SPEC, "v1:abc", 400L));
+ final long sizeUsedBefore = holder.getSizeUsed();
+ final int projectedCountBefore =
holder.getProjectedSegmentCounts().getTotalSegmentCount();
+
+ Assertions.assertTrue(holder.startOperation(SegmentAction.LOAD,
SIZED_SEGMENTS.get(0)));
+
+ Assertions.assertEquals(sizeUsedBefore + (SEGMENT_SIZE - 400L),
holder.getSizeUsed());
+ Assertions.assertEquals(
+ projectedCountBefore,
+ holder.getProjectedSegmentCounts().getTotalSegmentCount(),
+ "an in-place reload refreshes a replica that is already projected, it
does not add one"
+ );
+ }
+
+ @Test
+ public void testInPlaceReloadOfFullReplicaProjectsNothing()
+ {
+ // A replica with no profile is a regular full load that already holds the
whole segment, so applying a
+ // partial-load rule to it (or reverting it back to a plain load spec)
cannot add any bytes.
+ final ServerHolder holder = holderServing(null);
+ final long sizeUsedBefore = holder.getSizeUsed();
+
+ Assertions.assertTrue(holder.startOperation(SegmentAction.LOAD,
SIZED_SEGMENTS.get(0)));
+
+ Assertions.assertEquals(sizeUsedBefore, holder.getSizeUsed());
+ Assertions.assertEquals(1,
holder.getProjectedSegmentCounts().getTotalSegmentCount());
+ }
+
+ @Test
+ public void testCancellingAnInPlaceReloadRestoresTheProjection()
+ {
+ // add/remove have to agree on the delta, otherwise a cancelled reload
leaves the server's projection skewed for
+ // the rest of the run.
+ final ServerHolder holder =
holderServing(PartialLoadProfile.forLoaded(PARTIAL_LOAD_SPEC, "v1:abc", 400L));
+ final long sizeUsedBefore = holder.getSizeUsed();
+ final int projectedCountBefore =
holder.getProjectedSegmentCounts().getTotalSegmentCount();
+
+ // Queue through the load queue manager rather than startOperation
directly, so that the peon holds the segment
+ // and can accept the cancellation.
+ Assertions.assertTrue(
+ new SegmentLoadQueueManager(null, null)
+ .loadSegment(SIZED_SEGMENTS.get(0), holder, SegmentAction.LOAD,
null)
+ );
+ Assertions.assertTrue(holder.cancelOperation(SegmentAction.LOAD,
SIZED_SEGMENTS.get(0)));
+
+ Assertions.assertEquals(sizeUsedBefore, holder.getSizeUsed());
+ Assertions.assertEquals(
+ projectedCountBefore,
+ holder.getProjectedSegmentCounts().getTotalSegmentCount()
+ );
+ }
+
+ /**
+ * A historical serving {@code SIZED_SEGMENTS.get(0)}, announced with {@code
profile} when it is non-null so that the
+ * server's curr_size reflects a partial footprint rather than the whole
segment.
+ */
+ private static ServerHolder holderServing(@Nullable PartialLoadProfile
profile)
+ {
+ final DruidServer server =
+ new DruidServer("name1", "host1", null, 10_000L, null,
ServerType.HISTORICAL, "tier1", 0);
+ server.addDataSegment(SIZED_SEGMENTS.get(0), profile);
+ return new ServerHolder(server.toImmutableDruidServer(), new
TestLoadQueuePeon());
+ }
}
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssignerPartialTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssignerPartialTest.java
index f6bd341fd86..74b0e1582d1 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssignerPartialTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssignerPartialTest.java
@@ -495,9 +495,10 @@ public class StrategicSegmentAssignerPartialTest
@Test
public void testFullLoadReplicaTreatedAsStaleAgainstPartialRule()
{
- // s1 holds the segment as a regular full-load (no profile). Under a
partial rule it counts as stale: queue a
- // fresh load to satisfy the partial rule, but don't drop s1 yet; stale
stays serving until the matching load
- // completes. Note: s1 is also reload-eligible (additive), but because s2
is empty, s2 is preferred.
+ // s1 holds the segment as a regular full-load (no profile). Under a
partial rule it counts as stale, and it is
+ // still an in-place candidate: under virtual storage a queried full-load
replica is already backed by a partial
+ // cache entry that on-demand reads populated, so applying the rule there
pins bundles that are largely resident
+ // and avoids both a full download onto the empty s2 and the later drop of
s1.
final DataSegment segment = createSegment();
final ServerHolder s1 = createServerWithLoaded(TIER1, segment, null);
final ServerHolder s2 = createServer(TIER1);
@@ -513,11 +514,183 @@ public class StrategicSegmentAssignerPartialTest
stats.hasStat(Stats.Segments.PARTIAL_STALE_DROPPED),
"Stale must not be dropped before matching has actually loaded"
);
- Assertions.assertTrue(s2.getLoadingSegments().contains(segment));
- Assertions.assertEquals(profileForRevenue(), ((TestLoadQueuePeon)
s2.getPeon()).getProfileFor(segment));
+ Assertions.assertEquals(
+ profileForRevenue(),
+ ((TestLoadQueuePeon) s1.getPeon()).getProfileFor(segment),
+ "the profileless stale replica is reloaded in place under the rule's
fingerprint"
+ );
+ Assertions.assertTrue(s2.getLoadingSegments().isEmpty(), "no fresh
download is needed on the empty server");
Assertions.assertTrue(s1.getPeon().getSegmentsToDrop().isEmpty());
}
+ @Test
+ public void testInPlaceReloadPreferredOverFreshLoadOnEmptyServer()
+ {
+ // Same preference, but with a stale *partial* replica rather than a
full-load one, and with more empty servers
+ // than the deficit: the only thing queued is the in-place reload on the
server that already holds a footprint.
+ final DataSegment segment = createSegment();
+ final PartialLoadProfile usersProfile = PartialLoadProfile.forLoaded(
+ Map.of("type", "partialProjection", "projections", List.of("users"),
"fingerprint", FP_USERS),
+ FP_USERS,
+ 512L
+ );
+ final ServerHolder stale = createServerWithLoaded(TIER1, segment,
usersProfile);
+ final ServerHolder empty1 = createServer(TIER1);
+ final ServerHolder empty2 = createServer(TIER1);
+ final DruidCluster cluster = DruidCluster.builder().addTier(TIER1, stale,
empty1, empty2).build();
+
+ final DruidCoordinatorRuntimeParams params = makeRuntimeParams(cluster,
segment);
+ params.getSegmentAssigner()
+ .replicateSegmentPartially(segment, profileForRevenue(),
ImmutableMap.of(TIER1, 1));
+
+ final CoordinatorRunStats stats = params.getCoordinatorStats();
+ Assertions.assertEquals(1L,
stats.getSegmentStat(Stats.Segments.PARTIAL_ASSIGNED, TIER1,
segment.getDataSource()));
+ Assertions.assertEquals(profileForRevenue(), ((TestLoadQueuePeon)
stale.getPeon()).getProfileFor(segment));
+ Assertions.assertTrue(empty1.getLoadingSegments().isEmpty());
+ Assertions.assertTrue(empty2.getLoadingSegments().isEmpty());
+ }
+
+ @Test
+ public void testInPlaceReloadsFirstThenFreshLoadsCoverTheRest()
+ {
+ // Deficit of 2 with only one stale replica to reload in place: the
in-place reload covers one, and a fresh load
+ // on an empty server covers the other. The second empty server stays
untouched.
+ final DataSegment segment = createSegment();
+ final ServerHolder stale = createServerWithLoaded(TIER1, segment,
staleProfileForUsers());
+ final ServerHolder empty1 = createServer(TIER1);
+ final ServerHolder empty2 = createServer(TIER1);
+ final DruidCluster cluster = DruidCluster.builder().addTier(TIER1, stale,
empty1, empty2).build();
+
+ final DruidCoordinatorRuntimeParams params = makeRuntimeParams(cluster,
segment);
+ params.getSegmentAssigner()
+ .replicateSegmentPartially(segment, profileForRevenue(),
ImmutableMap.of(TIER1, 2));
+
+ final CoordinatorRunStats stats = params.getCoordinatorStats();
+ Assertions.assertEquals(2L,
stats.getSegmentStat(Stats.Segments.PARTIAL_ASSIGNED, TIER1,
segment.getDataSource()));
+ Assertions.assertEquals(profileForRevenue(), ((TestLoadQueuePeon)
stale.getPeon()).getProfileFor(segment));
+ Assertions.assertEquals(
+ 1,
+ empty1.getLoadingSegments().size() +
empty2.getLoadingSegments().size(),
+ "exactly one of the empty servers takes the remaining replica"
+ );
+ }
+
+ @Test
+ public void testFreshLoadCoversTheDeficitWhenTheInPlaceReloadCannotBeQueued()
+ {
+ // The in-place reload is preferred, but if queueing it on the stale
server fails the deficit is still real and a
+ // fresh candidate has to cover it. Without the fall-through the tier
would sit under-replicated for the run while
+ // an idle server was available.
+ final DataSegment segment = createSegment();
+ final DruidServer staleServer = createDruidServer(TIER1);
+ staleServer.addDataSegment(segment, staleProfileForUsers());
+ final ServerHolder stale =
+ new ServerHolder(staleServer.toImmutableDruidServer(), new
RefusingLoadQueuePeon());
+ final ServerHolder empty = createServer(TIER1);
+ final DruidCluster cluster = DruidCluster.builder().addTier(TIER1, stale,
empty).build();
+
+ final DruidCoordinatorRuntimeParams params = makeRuntimeParams(cluster,
segment);
+ params.getSegmentAssigner()
+ .replicateSegmentPartially(segment, profileForRevenue(),
ImmutableMap.of(TIER1, 1));
+
+ final CoordinatorRunStats stats = params.getCoordinatorStats();
+ Assertions.assertEquals(1L,
stats.getSegmentStat(Stats.Segments.PARTIAL_ASSIGNED, TIER1,
segment.getDataSource()));
+ Assertions.assertTrue(
+ empty.getLoadingSegments().contains(segment),
+ "the fresh candidate covers the deficit the refused in-place reload
left"
+ );
+ Assertions.assertEquals(profileForRevenue(), ((TestLoadQueuePeon)
empty.getPeon()).getProfileFor(segment));
+ Assertions.assertTrue(
+ stale.getPeon().getSegmentsToDrop().isEmpty(),
+ "the stale replica keeps serving until the replacement lands"
+ );
+ }
+
+ @Test
+ public void testLoadQueueFullStaleServerIsReplacedByFreshLoad()
+ {
+ // A stale server already at its per-run load-queue budget is not an
in-place candidate, so the empty server takes
+ // the replacement instead of the tier stalling on a server that cannot
accept the request.
+ final DataSegment segment = createSegment();
+ final DataSegment other = createSegment(Intervals.of("2020/2021"));
+ final ServerHolder stale = createServerWithLoadedAndQueueLimit(TIER1, 1,
staleProfileForUsers(), segment);
+ // Consume the server's single load-queue slot so canReloadInPlace rejects
it.
+ stale.getPeon().loadSegment(other, SegmentAction.LOAD, null);
+ Assertions.assertTrue(stale.startOperation(SegmentAction.LOAD, other));
+ Assertions.assertTrue(stale.isLoadQueueFull());
+
+ final ServerHolder empty = createServer(TIER1);
+ final DruidCluster cluster = DruidCluster.builder().addTier(TIER1, stale,
empty).build();
+
+ final DruidCoordinatorRuntimeParams params = makeRuntimeParams(cluster,
segment);
+ params.getSegmentAssigner()
+ .replicateSegmentPartially(segment, profileForRevenue(),
ImmutableMap.of(TIER1, 1));
+
+ final CoordinatorRunStats stats = params.getCoordinatorStats();
+ Assertions.assertEquals(1L,
stats.getSegmentStat(Stats.Segments.PARTIAL_ASSIGNED, TIER1,
segment.getDataSource()));
+ Assertions.assertTrue(empty.getLoadingSegments().contains(segment));
+ Assertions.assertNull(((TestLoadQueuePeon)
stale.getPeon()).getProfileFor(segment));
+ }
+
+ @Test
+ public void testDecommissioningStaleServerIsReplacedByFreshLoad()
+ {
+ // A decommissioning stale replica is on its way out, so reloading it in
place would be wasted work. The fresh
+ // load on the empty server is what satisfies the rule.
+ final DataSegment segment = createSegment();
+ final ServerHolder decommStale =
createDecommissioningServerWithLoaded(TIER1, segment, staleProfileForUsers());
+ final ServerHolder empty = createServer(TIER1);
+ final DruidCluster cluster = DruidCluster.builder().addTier(TIER1,
decommStale, empty).build();
+
+ final DruidCoordinatorRuntimeParams params = makeRuntimeParams(cluster,
segment);
+ params.getSegmentAssigner()
+ .replicateSegmentPartially(segment, profileForRevenue(),
ImmutableMap.of(TIER1, 1));
+
+ final CoordinatorRunStats stats = params.getCoordinatorStats();
+ Assertions.assertEquals(1L,
stats.getSegmentStat(Stats.Segments.PARTIAL_ASSIGNED, TIER1,
segment.getDataSource()));
+ Assertions.assertTrue(empty.getLoadingSegments().contains(segment));
+ Assertions.assertNull(((TestLoadQueuePeon)
decommStale.getPeon()).getProfileFor(segment));
+ }
+
+ @Test
+ public void testCancelledStaleInFlightOnLoadedServerIsReloadedInPlace()
+ {
+ // s1 both serves the segment under a stale fingerprint and has another
stale load in flight on top of it, so it
+ // classifies as stale-in-flight, not stale-loaded, and is missing from
the in-place bucket at snapshot time.
+ // Cancelling its load leaves it serving the segment, so it must be picked
up as an in-place destination: the
+ // fresh-load path filters on canLoadSegment, which rejects a server that
already has the segment.
+ final DataSegment segment = createSegment();
+ final PartialLoadProfile usersProfile = PartialLoadProfile.forRequest(
+ Map.of("type", "partialProjection", "projections", List.of("users"),
"fingerprint", FP_USERS),
+ FP_USERS
+ );
+ final DruidServer druidServer = createDruidServer(TIER1);
+ druidServer.addDataSegment(segment, usersProfile);
+ final TestLoadQueuePeon peon = new TestLoadQueuePeon();
+ peon.addInFlightHolder(
+ new SegmentHolder(segment, SegmentAction.LOAD, usersProfile,
Duration.standardSeconds(10), null)
+ );
+ final ServerHolder s1 = new
ServerHolder(druidServer.toImmutableDruidServer(), peon);
+ final DruidCluster cluster = DruidCluster.builder().addTier(TIER1,
s1).build();
+
+ final DruidCoordinatorRuntimeParams params = makeRuntimeParams(cluster,
segment);
+ params.getSegmentAssigner()
+ .replicateSegmentPartially(segment, profileForRevenue(),
ImmutableMap.of(TIER1, 1));
+
+ final CoordinatorRunStats stats = params.getCoordinatorStats();
+ Assertions.assertEquals(
+ 1L,
+ stats.getSegmentStat(Stats.Segments.PARTIAL_STALE_CANCELLED, TIER1,
segment.getDataSource())
+ );
+ Assertions.assertEquals(
+ 1L,
+ stats.getSegmentStat(Stats.Segments.PARTIAL_ASSIGNED, TIER1,
segment.getDataSource()),
+ "the cancelled slot is refilled in place rather than left empty"
+ );
+ Assertions.assertEquals(profileForRevenue(), peon.getProfileFor(segment));
+ Assertions.assertFalse(stats.hasStat(Stats.Segments.ASSIGN_SKIPPED));
+ }
+
@Test
public void testStaleDroppedAfterMatchingLoadedSatisfiesRequirement()
{
@@ -541,12 +714,11 @@ public class StrategicSegmentAssignerPartialTest
}
@Test
- public void testStuckStateAdditiveReloadOnStaleServers()
+ public void testInPlaceReloadOnEveryStaleServer()
{
- // Both servers hold the segment with mismatched fingerprints (and there
are no spare servers). Reconciler
- // queues additive-reload requests on both stale servers; the historical's
additive-load semantics fill in the
- // missing parts in place. matching count is still 0 in the same run, so
stale isn't dropped; but next run, after
- // the loads land, the servers reclassify as matching.
+ // Both servers hold the segment with mismatched fingerprints. Reconciler
queues an in-place reload on each; the
+ // historical swaps the rule on the cache entry it already has. matching
count is still 0 in the same run, so
+ // stale isn't dropped; but next run, after the loads land, the servers
reclassify as matching.
final DataSegment segment = createSegment();
final PartialLoadProfile usersProfile = PartialLoadProfile.forLoaded(
Map.of("type", "partialProjection", "projections", List.of("users"),
"fingerprint", FP_USERS),
@@ -1062,6 +1234,25 @@ public class StrategicSegmentAssignerPartialTest
.build();
}
+ /**
+ * Peon that refuses every load, the way the real peon surfaces an error
while queueing. {@link
+ * SegmentLoadQueueManager#loadSegment} catches it, rolls the operation back
off the {@link ServerHolder} and
+ * reports the destination as unusable.
+ */
+ private static class RefusingLoadQueuePeon extends TestLoadQueuePeon
+ {
+ @Override
+ public void loadSegment(
+ DataSegment segment,
+ SegmentAction action,
+ @Nullable PartialLoadProfile profile,
+ @Nullable LoadPeonCallback callback
+ )
+ {
+ throw new IllegalStateException("Cannot queue segment[" +
segment.getId() + "]");
+ }
+ }
+
/**
* Peon that records the segments marked to drop, so that a move source
shows up as MOVE_FROM in the
* {@link ServerHolder}'s queue the way the real peon reports it.
@@ -1089,6 +1280,20 @@ public class StrategicSegmentAssignerPartialTest
}
}
+ /**
+ * A loaded profile under a fingerprint the tests' rule never asks for, i.e.
a stale <em>partial</em> replica. A
+ * stale replica announcing no profile at all is an in-place candidate too,
this just exercises the partial-to-
+ * partial swap specifically.
+ */
+ private static PartialLoadProfile staleProfileForUsers()
+ {
+ return PartialLoadProfile.forLoaded(
+ Map.of("type", "partialProjection", "projections", List.of("users"),
"fingerprint", FP_USERS),
+ FP_USERS,
+ 512L
+ );
+ }
+
private static PartialLoadProfile profileForRevenue()
{
return PartialLoadProfile.forRequest(
diff --git
a/server/src/test/java/org/apache/druid/test/utils/TestSegmentCacheManager.java
b/server/src/test/java/org/apache/druid/test/utils/TestSegmentCacheManager.java
index f4cbb9dcd6d..34a744f3a8b 100644
---
a/server/src/test/java/org/apache/druid/test/utils/TestSegmentCacheManager.java
+++
b/server/src/test/java/org/apache/druid/test/utils/TestSegmentCacheManager.java
@@ -31,6 +31,7 @@ import org.apache.druid.segment.loading.AcquireMode;
import org.apache.druid.segment.loading.AcquireSegmentAction;
import org.apache.druid.segment.loading.AcquireSegmentResult;
import org.apache.druid.segment.loading.NoopSegmentCacheManager;
+import org.apache.druid.segment.loading.SegmentLoadingException;
import org.apache.druid.segment.loading.TombstoneSegmentizerFactory;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.SegmentId;
@@ -60,6 +61,12 @@ public class TestSegmentCacheManager extends
NoopSegmentCacheManager
private final Set<SegmentId> observedSegmentsRemovedFromCache;
private final AtomicInteger observedShutdownBootstrapCount;
+ /**
+ * Loads still allowed to succeed before {@link #load} starts failing, see
{@link #failLoadsAfter}. Unlimited
+ * unless a test says otherwise.
+ */
+ private final AtomicInteger remainingSuccessfulLoads = new
AtomicInteger(Integer.MAX_VALUE);
+
public TestSegmentCacheManager()
{
this(ImmutableSet.of());
@@ -110,9 +117,22 @@ public class TestSegmentCacheManager extends
NoopSegmentCacheManager
return segment;
}
+ /**
+ * Makes {@link #load} succeed {@code numSuccessfulLoads} more times and
fail every load after that. Lets a test
+ * establish a serving replica and then fail a reload of it, which is what
distinguishes failure cleanup that is
+ * safe from cleanup that would tear down a live replica.
+ */
+ public void failLoadsAfter(int numSuccessfulLoads)
+ {
+ remainingSuccessfulLoads.set(numSuccessfulLoads);
+ }
+
@Override
- public DataSegment load(final DataSegment segment)
+ public DataSegment load(final DataSegment segment) throws
SegmentLoadingException
{
+ if (remainingSuccessfulLoads.getAndUpdate(remaining -> remaining > 0 ?
remaining - 1 : remaining) <= 0) {
+ throw new SegmentLoadingException("Test-induced load failure for
segment[%s]", segment.getId());
+ }
observedSegments.add(segment);
getSegmentInternal(segment);
return segment;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
