This is an automated email from the ASF dual-hosted git repository.
capistrant 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 4c0f5240c1a fix: Fix undesireable load rule behavior for clustered
segments with conforming shape but no matched groups (#19728)
4c0f5240c1a is described below
commit 4c0f5240c1a488ad7bbe446049544936c4cec7ab
Author: Lucas Capistrant <[email protected]>
AuthorDate: Thu Jul 23 07:07:03 2026 -0500
fix: Fix undesireable load rule behavior for clustered segments with
conforming shape but no matched groups (#19728)
---
.../druid/server/coordinator/duty/RunRules.java | 109 +-------
.../rules/ClusterGroupPartialLoadMatcher.java | 31 +--
.../coordinator/rules/PartialLoadMatcher.java | 19 +-
.../duty/RunRulesEmptyDeferringHandlerTest.java | 255 ------------------
.../duty/RunRulesPartialLoadPlacementTest.java | 285 +++++++++++++++++++++
...WildcardClusterGroupPartialLoadMatcherTest.java | 22 +-
6 files changed, 317 insertions(+), 404 deletions(-)
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/duty/RunRules.java
b/server/src/main/java/org/apache/druid/server/coordinator/duty/RunRules.java
index 88ea485f679..a03d7fe64bc 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/duty/RunRules.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/duty/RunRules.java
@@ -26,23 +26,17 @@ import org.apache.druid.java.util.common.Stopwatch;
import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.server.coordinator.DruidCluster;
import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
-import org.apache.druid.server.coordinator.loading.PartialLoadProfile;
import org.apache.druid.server.coordinator.loading.StrategicSegmentAssigner;
import org.apache.druid.server.coordinator.rules.BroadcastDistributionRule;
-import org.apache.druid.server.coordinator.rules.PartialLoadMatcher;
import org.apache.druid.server.coordinator.rules.Rule;
-import org.apache.druid.server.coordinator.rules.SegmentActionHandler;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.Dimension;
import org.apache.druid.server.coordinator.stats.RowKey;
import org.apache.druid.server.coordinator.stats.Stats;
import org.apache.druid.timeline.DataSegment;
import org.joda.time.DateTime;
-import org.joda.time.Interval;
-import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
@@ -89,22 +83,10 @@ public class RunRules implements CoordinatorDuty
final Set<DataSegment> overshadowed =
params.getDataSourcesSnapshot().getOvershadowedSegments();
final StrategicSegmentAssigner segmentAssigner =
params.getSegmentAssigner();
- // Wrap the assigner with a per-shard-group buffering handler.
Partial-load matchers that resolve asymmetrically
- // across siblings dispatch their "no positive content for this segment"
decision as an empty-fingerprint
- // partial-load. The buffer holds those decisions until we finish
iterating the shard group; if any sibling
- // produced a positive match we flush them (preserving broker-side
shard-group completeness), otherwise we discard
- // them (avoiding wasted empty loads on groups with no positive content
for the matcher anywhere).
- final EmptyDeferringHandler segmentHandler = new
EmptyDeferringHandler(segmentAssigner);
final DateTime now = DateTimes.nowUtc();
final Object2IntOpenHashMap<String> datasourceToSegmentsWithNoRule = new
Object2IntOpenHashMap<>();
- // Streaming shard-group boundary state.
SegmentHolder.NEWEST_SEGMENT_FIRST groups segments contiguously by
- // (dataSource, interval, version), so on any change in that triple we
flush the buffer for the previous group.
- String currentDs = null;
- Interval currentInterval = null;
- String currentVersion = null;
-
for (DataSegment segment : usedSegments) {
// Do not apply rules on overshadowed segments as they will be
// marked unused and eventually unloaded from all historicals
@@ -112,22 +94,12 @@ public class RunRules implements CoordinatorDuty
continue;
}
- // Detect a shard-group boundary and flush deferred empty loads for the
previous group.
- if (!segment.getDataSource().equals(currentDs)
- || !segment.getInterval().equals(currentInterval)
- || !segment.getVersion().equals(currentVersion)) {
- segmentHandler.flushAndReset();
- currentDs = segment.getDataSource();
- currentInterval = segment.getInterval();
- currentVersion = segment.getVersion();
- }
-
// Find and apply matching rule
List<Rule> rules =
ruleHandler.getRulesWithDefault(segment.getDataSource());
boolean foundMatchingRule = false;
for (Rule rule : rules) {
if (rule.appliesTo(segment, now)) {
- rule.run(segment, segmentHandler);
+ rule.run(segment, segmentAssigner);
foundMatchingRule = true;
break;
}
@@ -138,9 +110,6 @@ public class RunRules implements CoordinatorDuty
}
}
- // Tail flush for the last shard group.
- segmentHandler.flushAndReset();
-
processSegmentDeletes(segmentAssigner, params.getCoordinatorStats());
alertForSegmentsWithNoRules(datasourceToSegmentsWithNoRule);
alertForInvalidRules(segmentAssigner);
@@ -212,80 +181,4 @@ public class RunRules implements CoordinatorDuty
.stream()
.anyMatch(rule -> rule instanceof
BroadcastDistributionRule);
}
-
- /**
- * Per-shard-group buffering decorator for {@link SegmentActionHandler}.
Intercepts partial-load dispatches with
- * the {@link PartialLoadMatcher#EMPTY_LOAD_FINGERPRINT} sentinel and holds
them until {@link #flushAndReset} is
- * called by the surrounding shard-group iteration. Positive partial-loads
(non-empty fingerprint) pass through
- * immediately and mark the current group as having a positive match.
- *
- * <p>At flush time, buffered empties are dispatched only if a positive
match was seen in the same group. This
- * preserves broker-side shard-group completeness for asymmetric matchers
while avoiding empty loads when no segment
- * in the group has positive content for the matcher.
- *
- * <p>All other handler methods (full load, broadcast, delete) pass through
to the delegate unchanged.
- */
- static final class EmptyDeferringHandler implements SegmentActionHandler
- {
- private final SegmentActionHandler delegate;
- private final List<DeferredEmpty> pendingEmpties = new ArrayList<>();
- private boolean anyPositiveInGroup = false;
-
- EmptyDeferringHandler(SegmentActionHandler delegate)
- {
- this.delegate = delegate;
- }
-
- @Override
- public void replicateSegment(DataSegment segment, Map<String, Integer>
tierToReplicaCount)
- {
- delegate.replicateSegment(segment, tierToReplicaCount);
- }
-
- @Override
- public void replicateSegmentPartially(
- DataSegment segment,
- PartialLoadProfile profile,
- Map<String, Integer> tierToReplicaCount
- )
- {
- if
(PartialLoadMatcher.EMPTY_LOAD_FINGERPRINT.equals(profile.fingerprint())) {
- pendingEmpties.add(new DeferredEmpty(segment, profile,
tierToReplicaCount));
- } else {
- anyPositiveInGroup = true;
- delegate.replicateSegmentPartially(segment, profile,
tierToReplicaCount);
- }
- }
-
- @Override
- public void broadcastSegment(DataSegment segment)
- {
- delegate.broadcastSegment(segment);
- }
-
- @Override
- public void deleteSegment(DataSegment segment)
- {
- delegate.deleteSegment(segment);
- }
-
- void flushAndReset()
- {
- if (anyPositiveInGroup) {
- for (DeferredEmpty d : pendingEmpties) {
- delegate.replicateSegmentPartially(d.segment, d.profile,
d.tierToReplicaCount);
- }
- }
- pendingEmpties.clear();
- anyPositiveInGroup = false;
- }
-
- private record DeferredEmpty(
- DataSegment segment,
- PartialLoadProfile profile,
- Map<String, Integer> tierToReplicaCount
- )
- {
- }
- }
}
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/rules/ClusterGroupPartialLoadMatcher.java
b/server/src/main/java/org/apache/druid/server/coordinator/rules/ClusterGroupPartialLoadMatcher.java
index 18a27c50440..410cfb22213 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/rules/ClusterGroupPartialLoadMatcher.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/rules/ClusterGroupPartialLoadMatcher.java
@@ -24,7 +24,6 @@ import com.google.common.hash.Hashing;
import com.google.common.io.BaseEncoding;
import org.apache.druid.segment.loading.PartialClusterGroupLoadSpec;
import org.apache.druid.timeline.DataSegment;
-import org.apache.druid.timeline.partition.ShardSpec;
import javax.annotation.Nullable;
import java.util.List;
@@ -60,22 +59,22 @@ public abstract class ClusterGroupPartialLoadMatcher
implements PartialLoadMatch
* Returns the load spec for the resolved cluster-group indices, or null
when this matcher has nothing to
* contribute for the given segment.
*
- * <p>Null is returned when:
+ * <p>Null (opaque) is returned only when the matcher cannot reason about
the segment at all:
* <ul>
- * <li>the segment is not clustered,</li>
+ * <li>the segment is not clustered, or</li>
* <li>the matcher's patterns are incompatible with the segment's
clustering scheme (see
- * {@link #resolveClusterGroupIndices}), or</li>
- * <li>no configured pattern matches the segment's tuples <em>and</em> the
segment is not a core partition
- * (i.e. {@code partitionNum >= numCorePartitions}). The empty load is
only useful to keep the broker's
- * shard-group completeness check happy, and that check applies only
to the core partition group; appended
- * segments are queried individually and don't need an empty stub when
no positive content matches.</li>
+ * {@link #resolveClusterGroupIndices}).</li>
* </ul>
+ * A null result hands the decision to the rule's {@link
CannotMatchBehavior}.
*
- * <p>When the segment is a core partition of a clustered shard group, the
matcher is compatible, and no pattern
- * matches any tuple, returns the "empty" load (same {@code
partialClusterGroup} type with an empty index list
- * and {@link #EMPTY_LOAD_FINGERPRINT}). The historical-side loader honors
it by performing no load, leaving the
- * segment uniformly placed in the timeline alongside its positively-matched
siblings so the broker treats the
- * group as complete.
+ * <p>When the matcher <em>is</em> compatible with the segment's clustering
scheme, it always returns a non-null
+ * result: a positive load for the matched cluster-group indices, or the
"empty" load (same
+ * {@code partialClusterGroup} type with an empty index list and {@link
#EMPTY_LOAD_FINGERPRINT}) when no configured
+ * pattern matches any tuple. This holds regardless of whether the segment
is a core or an appended
+ * ({@code partitionNum >= numCorePartitions}) partition: an empty result
means "the matcher analyzed this segment
+ * and nothing here should load," which keeps the segment announceable
rather than silently dropping it. The empty
+ * load is dispatched like any other partial load onto the rule's tiered
replicants, so the segment stays in the
+ * broker's timeline; the historical-side loader honors the empty index list
by downloading no cluster-group data.
*/
@Override
@Nullable
@@ -90,12 +89,6 @@ public abstract class ClusterGroupPartialLoadMatcher
implements PartialLoadMatch
// handling takes over rather than dispatching a stub empty load.
return null;
}
- final ShardSpec shardSpec = segment.getShardSpec();
- if (resolved.isEmpty() && shardSpec.getPartitionNum() >=
shardSpec.getNumCorePartitions()) {
- // No patterns match and this segment isn't part of a core partition
group, so no need for an empty load. Fall
- // through to the cannot-match handling.
- return null;
- }
final String fingerprint = resolved.isEmpty() ? EMPTY_LOAD_FINGERPRINT :
computeFingerprint(resolved);
return new MatchResult(PartialClusterGroupLoadSpec.wireForm(baseLoadSpec,
resolved, fingerprint), fingerprint);
}
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/rules/PartialLoadMatcher.java
b/server/src/main/java/org/apache/druid/server/coordinator/rules/PartialLoadMatcher.java
index b296631591f..5bc1690a20e 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/rules/PartialLoadMatcher.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/rules/PartialLoadMatcher.java
@@ -44,11 +44,11 @@ public interface PartialLoadMatcher
{
/**
* Universal fingerprint sentinel for an "empty match" — a matcher decision
to partial-load a segment with no
- * scheme-specific content. Used by matchers that handle asymmetric
resolution across siblings of a shard group:
- * when the matcher applies to a segment but resolves to no positive content
(e.g., a cluster-group matcher on a
- * clustered segment whose tuples don't intersect any configured pattern),
it returns a {@link MatchResult} with
- * this fingerprint so the coordinator's {@code RunRules} duty can defer the
empty-load dispatch and only flush it
- * when at least one sibling in the same shard group produced a positive
match.
+ * scheme-specific content. A matcher that applies to a segment but resolves
to no positive content (e.g., a
+ * cluster-group matcher on a clustered segment whose tuples don't intersect
any configured pattern) returns a
+ * {@link MatchResult} with this fingerprint. The empty load is dispatched
like any other partial load onto the
+ * rule's tiered replicants, so the segment stays announced (and thus in the
broker's timeline); the historical
+ * downloads no scheme-specific content for it.
*
* <p>All matchers share this fingerprint for empty loads — different
matchers' empty wire forms are equivalent
* from a "what's on the historical" perspective (no scheme-specific extras
downloaded), and at most one rule
@@ -72,14 +72,5 @@ public interface PartialLoadMatcher
*/
record MatchResult(Map<String, Object> wrappedLoadSpec, String fingerprint)
{
- /**
- * Whether this is an "empty match" — the matcher applies to the segment
but resolves to no positive content.
- * Recognized via {@link #EMPTY_LOAD_FINGERPRINT}. Empty loads are
dispatched only when at least one sibling in
- * the same shard group produced a positive match; otherwise they're
discarded by {@code RunRules}.
- */
- public boolean isEmpty()
- {
- return EMPTY_LOAD_FINGERPRINT.equals(fingerprint);
- }
}
}
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesEmptyDeferringHandlerTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesEmptyDeferringHandlerTest.java
deleted file mode 100644
index 5a828f42dc1..00000000000
---
a/server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesEmptyDeferringHandlerTest.java
+++ /dev/null
@@ -1,255 +0,0 @@
-/*
- * 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.duty;
-
-import org.apache.druid.java.util.common.Intervals;
-import org.apache.druid.server.coordinator.loading.PartialLoadProfile;
-import org.apache.druid.server.coordinator.rules.PartialLoadMatcher;
-import org.apache.druid.server.coordinator.rules.SegmentActionHandler;
-import org.apache.druid.timeline.DataSegment;
-import org.apache.druid.timeline.SegmentId;
-import org.apache.druid.timeline.partition.NumberedShardSpec;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-class RunRulesEmptyDeferringHandlerTest
-{
- private static final Map<String, Integer> TIER1_REPLICANTS = Map.of("tier1",
1);
- private static final String POSITIVE_FINGERPRINT = "v1:positive";
-
- @Test
- void positiveLoadPassesThroughImmediately()
- {
- final RecordingHandler delegate = new RecordingHandler();
- final RunRules.EmptyDeferringHandler handler = new
RunRules.EmptyDeferringHandler(delegate);
- final DataSegment seg = segment(0);
- final PartialLoadProfile positive = profile(POSITIVE_FINGERPRINT);
-
- handler.replicateSegmentPartially(seg, positive, TIER1_REPLICANTS);
-
- // Positive load is dispatched immediately, not buffered.
- Assertions.assertEquals(1, delegate.partialLoads.size());
- Assertions.assertEquals(seg, delegate.partialLoads.get(0).segment);
- Assertions.assertSame(positive, delegate.partialLoads.get(0).profile);
- }
-
- @Test
- void emptyLoadIsBufferedUntilFlush()
- {
- final RecordingHandler delegate = new RecordingHandler();
- final RunRules.EmptyDeferringHandler handler = new
RunRules.EmptyDeferringHandler(delegate);
- final DataSegment seg = segment(0);
- final PartialLoadProfile empty =
profile(PartialLoadMatcher.EMPTY_LOAD_FINGERPRINT);
-
- handler.replicateSegmentPartially(seg, empty, TIER1_REPLICANTS);
-
- // Empty load is buffered, not dispatched.
- Assertions.assertTrue(delegate.partialLoads.isEmpty());
- }
-
- @Test
- void flushDispatchesBufferedEmptiesWhenPositiveSeen()
- {
- final RecordingHandler delegate = new RecordingHandler();
- final RunRules.EmptyDeferringHandler handler = new
RunRules.EmptyDeferringHandler(delegate);
- final DataSegment p0 = segment(0);
- final DataSegment p1 = segment(1);
- final PartialLoadProfile positive = profile(POSITIVE_FINGERPRINT);
- final PartialLoadProfile empty =
profile(PartialLoadMatcher.EMPTY_LOAD_FINGERPRINT);
-
- handler.replicateSegmentPartially(p0, positive, TIER1_REPLICANTS);
- handler.replicateSegmentPartially(p1, empty, TIER1_REPLICANTS);
- // After the group: p0 dispatched (positive), p1 buffered.
- Assertions.assertEquals(1, delegate.partialLoads.size());
-
- handler.flushAndReset();
- // Now p1 should be dispatched too — the group had a positive match.
- Assertions.assertEquals(2, delegate.partialLoads.size());
- Assertions.assertEquals(p1, delegate.partialLoads.get(1).segment);
- }
-
- @Test
- void flushDiscardsBufferedEmptiesWhenNoPositive()
- {
- final RecordingHandler delegate = new RecordingHandler();
- final RunRules.EmptyDeferringHandler handler = new
RunRules.EmptyDeferringHandler(delegate);
- final PartialLoadProfile empty =
profile(PartialLoadMatcher.EMPTY_LOAD_FINGERPRINT);
-
- handler.replicateSegmentPartially(segment(0), empty, TIER1_REPLICANTS);
- handler.replicateSegmentPartially(segment(1), empty, TIER1_REPLICANTS);
-
- handler.flushAndReset();
-
- // No positive in the group — buffered empties are discarded, never
dispatched.
- Assertions.assertTrue(delegate.partialLoads.isEmpty());
- }
-
- @Test
- void flushResetsStateBetweenGroups()
- {
- final RecordingHandler delegate = new RecordingHandler();
- final RunRules.EmptyDeferringHandler handler = new
RunRules.EmptyDeferringHandler(delegate);
- final PartialLoadProfile positive = profile(POSITIVE_FINGERPRINT);
- final PartialLoadProfile empty =
profile(PartialLoadMatcher.EMPTY_LOAD_FINGERPRINT);
-
- // Group 1: has a positive match → buffered empty gets dispatched on flush.
- handler.replicateSegmentPartially(segment(0), positive, TIER1_REPLICANTS);
- handler.replicateSegmentPartially(segment(1), empty, TIER1_REPLICANTS);
- handler.flushAndReset();
- Assertions.assertEquals(2, delegate.partialLoads.size());
-
- // Group 2: only empties → discarded on flush. The previous group's
positive must not leak through.
- handler.replicateSegmentPartially(segment(10), empty, TIER1_REPLICANTS);
- handler.replicateSegmentPartially(segment(11), empty, TIER1_REPLICANTS);
- handler.flushAndReset();
- Assertions.assertEquals(2, delegate.partialLoads.size()); // unchanged
- }
-
- @Test
- void flushOnEmptyBufferIsNoOp()
- {
- final RecordingHandler delegate = new RecordingHandler();
- final RunRules.EmptyDeferringHandler handler = new
RunRules.EmptyDeferringHandler(delegate);
-
- handler.flushAndReset();
-
- Assertions.assertTrue(delegate.partialLoads.isEmpty());
- }
-
- @Test
- void fullLoadPassesThroughUnchanged()
- {
- final RecordingHandler delegate = new RecordingHandler();
- final RunRules.EmptyDeferringHandler handler = new
RunRules.EmptyDeferringHandler(delegate);
- final DataSegment seg = segment(0);
-
- handler.replicateSegment(seg, TIER1_REPLICANTS);
-
- Assertions.assertEquals(1, delegate.fullLoads.size());
- Assertions.assertEquals(seg, delegate.fullLoads.get(0).segment);
- }
-
- @Test
- void broadcastPassesThroughUnchanged()
- {
- final RecordingHandler delegate = new RecordingHandler();
- final RunRules.EmptyDeferringHandler handler = new
RunRules.EmptyDeferringHandler(delegate);
- final DataSegment seg = segment(0);
-
- handler.broadcastSegment(seg);
-
- Assertions.assertEquals(List.of(seg), delegate.broadcasts);
- }
-
- @Test
- void deletePassesThroughUnchanged()
- {
- final RecordingHandler delegate = new RecordingHandler();
- final RunRules.EmptyDeferringHandler handler = new
RunRules.EmptyDeferringHandler(delegate);
- final DataSegment seg = segment(0);
-
- handler.deleteSegment(seg);
-
- Assertions.assertEquals(List.of(seg), delegate.deletes);
- }
-
- @Test
- void positiveLoadDispatchedAfterBufferedEmptyStillFlushesEmpty()
- {
- // Order shouldn't matter: empties buffered before the positive arrives
should still get flushed because the
- // positive marks the group as having a positive match.
- final RecordingHandler delegate = new RecordingHandler();
- final RunRules.EmptyDeferringHandler handler = new
RunRules.EmptyDeferringHandler(delegate);
- final PartialLoadProfile positive = profile(POSITIVE_FINGERPRINT);
- final PartialLoadProfile empty =
profile(PartialLoadMatcher.EMPTY_LOAD_FINGERPRINT);
-
- handler.replicateSegmentPartially(segment(0), empty, TIER1_REPLICANTS);
- handler.replicateSegmentPartially(segment(1), positive, TIER1_REPLICANTS);
- Assertions.assertEquals(1, delegate.partialLoads.size()); // positive only
-
- handler.flushAndReset();
- Assertions.assertEquals(2, delegate.partialLoads.size()); // empty now
flushed too
- }
-
- private static DataSegment segment(int partitionNum)
- {
- final NumberedShardSpec shardSpec = new NumberedShardSpec(partitionNum, 2);
- return DataSegment
- .builder(SegmentId.of("ds", Intervals.of("2026-01-01/2026-01-02"),
"v", shardSpec))
- .shardSpec(shardSpec)
- .loadSpec(Map.of("type", "local", "path", "/seg"))
- .size(0)
- .build();
- }
-
- private static PartialLoadProfile profile(String fingerprint)
- {
- return PartialLoadProfile.forRequest(Map.of("type", "test", "fingerprint",
fingerprint), fingerprint);
- }
-
- private record RecordedPartialLoad(DataSegment segment, PartialLoadProfile
profile, Map<String, Integer> replicants)
- {
- }
-
- private record RecordedFullLoad(DataSegment segment, Map<String, Integer>
replicants)
- {
- }
-
- private static final class RecordingHandler implements SegmentActionHandler
- {
- final List<RecordedPartialLoad> partialLoads = new ArrayList<>();
- final List<RecordedFullLoad> fullLoads = new ArrayList<>();
- final List<DataSegment> broadcasts = new ArrayList<>();
- final List<DataSegment> deletes = new ArrayList<>();
-
- @Override
- public void replicateSegment(DataSegment segment, Map<String, Integer>
tierToReplicaCount)
- {
- fullLoads.add(new RecordedFullLoad(segment, new
HashMap<>(tierToReplicaCount)));
- }
-
- @Override
- public void replicateSegmentPartially(
- DataSegment segment,
- PartialLoadProfile profile,
- Map<String, Integer> tierToReplicaCount
- )
- {
- partialLoads.add(new RecordedPartialLoad(segment, profile, new
HashMap<>(tierToReplicaCount)));
- }
-
- @Override
- public void broadcastSegment(DataSegment segment)
- {
- broadcasts.add(segment);
- }
-
- @Override
- public void deleteSegment(DataSegment segment)
- {
- deletes.add(segment);
- }
- }
-}
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesPartialLoadPlacementTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesPartialLoadPlacementTest.java
new file mode 100644
index 00000000000..47cbb007068
--- /dev/null
+++
b/server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesPartialLoadPlacementTest.java
@@ -0,0 +1,285 @@
+/*
+ * 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.duty;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.client.DruidServer;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.concurrent.Execs;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
+import org.apache.druid.server.coordinator.DruidCluster;
+import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
+import org.apache.druid.server.coordinator.ServerHolder;
+import org.apache.druid.server.coordinator.balancer.BalancerStrategy;
+import org.apache.druid.server.coordinator.balancer.CostBalancerStrategy;
+import org.apache.druid.server.coordinator.loading.SegmentLoadQueueManager;
+import org.apache.druid.server.coordinator.loading.TestLoadQueuePeon;
+import org.apache.druid.server.coordinator.rules.CannotMatchBehavior;
+import org.apache.druid.server.coordinator.rules.ForeverPartialLoadRule;
+import org.apache.druid.server.coordinator.rules.PeriodPartialLoadRule;
+import org.apache.druid.server.coordinator.rules.Rule;
+import
org.apache.druid.server.coordinator.rules.WildcardClusterGroupPartialLoadMatcher;
+import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
+import org.apache.druid.server.coordinator.stats.Stats;
+import org.apache.druid.timeline.ClusterGroupTuples;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
+import org.apache.druid.timeline.partition.NumberedShardSpec;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.joda.time.Period;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Placement-level behavior of cluster-group partial-load rules driven through
the full {@link RunRules} duty. These
+ * assert what actually lands on a tier: a compatible-but-unmatched
cluster-group segment is announced (empty loaded)
+ * on the claiming rule's tier rather than dropped from every tier or fully
downloaded.
+ */
+public class RunRulesPartialLoadPlacementTest
+{
+ private static final String DATASOURCE = "ds";
+ private static final String TIER = "tier1";
+ private static final Interval CHUNK = Intervals.of("2026-01-01/2026-01-02");
+
+ private ListeningExecutorService exec;
+ private BalancerStrategy balancerStrategy;
+ private SegmentLoadQueueManager loadQueueManager;
+
+ @Before
+ public void setUp()
+ {
+ exec = MoreExecutors.listeningDecorator(Execs.multiThreaded(1,
"RunRulesPartialLoadPlacementTest-%d"));
+ balancerStrategy = new CostBalancerStrategy(exec);
+ loadQueueManager = new SegmentLoadQueueManager(null, null);
+ }
+
+ @After
+ public void tearDown()
+ {
+ exec.shutdown();
+ }
+
+ /**
+ * A shard group in which every core partition is compatible-but-unmatched
must be empty loaded so it stays
+ * announced and queryable, rather than being dropped from every tier and
invisible to the Broker(s)
+ */
+ @Test
+ public void fullyColdCoreGroup_isWeakLoadedNotDiscarded()
+ {
+ // Two core partitions of one shard group; neither tuple matches the
rule's include pattern.
+ final DataSegment core0 = clusteredSegment(new NumberedShardSpec(0, 2),
"acme");
+ final DataSegment core1 = clusteredSegment(new NumberedShardSpec(1, 2),
"globex");
+
+ final CoordinatorRunStats stats = runRules(matchNobodyForeverRule(),
core0, core1);
+
+ Assert.assertEquals(
+ "both core partitions of a fully-unmatched group are empty loaded, not
discarded",
+ 2L,
+ stats.getSegmentStat(Stats.Segments.PARTIAL_ASSIGNED, TIER, DATASOURCE)
+ );
+ }
+
+ /**
+ * An appended (non-core) partition that is compatible-but-unmatched must be
empty loaded, not
+ * fully downloaded via the cannot-match fallback.
+ */
+ @Test
+ public void appendedColdSegment_isWeakLoadedNotFullLoaded()
+ {
+ final DataSegment appended = clusteredSegment(new NumberedShardSpec(2, 2),
"acme");
+
+ final CoordinatorRunStats stats = runRules(matchNobodyForeverRule(),
appended);
+
+ Assert.assertEquals(
+ "appended unmatched segment is empty partial loaded",
+ 1L,
+ stats.getSegmentStat(Stats.Segments.PARTIAL_ASSIGNED, TIER, DATASOURCE)
+ );
+ Assert.assertFalse(
+ "appended unmatched segment must not be fully downloaded",
+ stats.hasStat(Stats.Segments.ASSIGNED)
+ );
+ }
+
+ /**
+ * When a group has at least one positively-matched sibling, its unmatched
core siblings are empty loaded on the
+ * rule's own tier to ensure the core partition set is complete.
+ */
+ @Test
+ public void partiallyHotGroup_coldSiblingWeakLoadedOnOwnTier()
+ {
+ final DataSegment matched = clusteredSegment(new NumberedShardSpec(0, 2),
"acme");
+ final DataSegment unmatched = clusteredSegment(new NumberedShardSpec(1,
2), "globex");
+
+ final ForeverPartialLoadRule rule = new ForeverPartialLoadRule(
+ ImmutableMap.of(TIER, 1),
+ null,
+ new WildcardClusterGroupPartialLoadMatcher(List.of(Map.of("tenant",
"acme")), null),
+ CannotMatchBehavior.FULL_LOAD
+ );
+
+ final CoordinatorRunStats stats = runRules(rule, matched, unmatched);
+
+ // Both the positive match and the empty match are loaded
+ Assert.assertEquals(2L,
stats.getSegmentStat(Stats.Segments.PARTIAL_ASSIGNED, TIER, DATASOURCE));
+ }
+
+ /**
+ * A realistic core partition set of 3 (tenant, region) tuples, governed by
an in-window P7D period rule whose
+ * include pattern matches a tenant none of them carry. Every partition
resolves to the empty load, and even with no
+ * positive sibling, the whole group must still be empty loaded.
+ */
+ @Test
+ public void
fullyColdThreePartitionCoreSet_underInWindowPeriodRule_isWeakLoaded()
+ {
+ final Interval recent = recentChunk();
+ final DataSegment p0 = tenantRegionSegment(recent, new
NumberedShardSpec(0, 3), "acme", "us-east-1");
+ final DataSegment p1 = tenantRegionSegment(recent, new
NumberedShardSpec(1, 3), "acme", "us-west-2");
+ final DataSegment p2 = tenantRegionSegment(recent, new
NumberedShardSpec(2, 3), "globex", "us-east-1");
+
+ final PeriodPartialLoadRule rule = new PeriodPartialLoadRule(
+ Period.days(7),
+ null,
+ ImmutableMap.of(TIER, 1),
+ null,
+ new WildcardClusterGroupPartialLoadMatcher(List.of(Map.of("tenant",
"biz", "region", "*")), null),
+ CannotMatchBehavior.FALL_THROUGH
+ );
+
+ final CoordinatorRunStats stats = runRules(rule, p0, p1, p2);
+
+ Assert.assertEquals(
+ "all 3 unmatched core partitions are partial loaded so the group stays
queryable on demand",
+ 3L,
+ stats.getSegmentStat(Stats.Segments.PARTIAL_ASSIGNED, TIER, DATASOURCE)
+ );
+ Assert.assertFalse("no partition is fully downloaded",
stats.hasStat(Stats.Segments.ASSIGNED));
+ }
+
+ private ForeverPartialLoadRule matchNobodyForeverRule()
+ {
+ // Include pattern resolves against the "tenant" clustering column
(compatible) but matches none of the segments'
+ // tuples, so every segment resolves to the empty load. onCannotMatch is
irrelevant for a compatible matcher.
+ return new ForeverPartialLoadRule(
+ ImmutableMap.of(TIER, 1),
+ null,
+ new WildcardClusterGroupPartialLoadMatcher(List.of(Map.of("tenant",
"nobody")), null),
+ CannotMatchBehavior.FULL_LOAD
+ );
+ }
+
+ private CoordinatorRunStats runRules(Rule rule, DataSegment... segments)
+ {
+ return runRules(singleTierCluster(), rule, segments);
+ }
+
+ private CoordinatorRunStats runRules(DruidCluster cluster, Rule rule,
DataSegment... segments)
+ {
+ final List<Rule> rules = Collections.singletonList(rule);
+ final RunRules ruleRunner = new RunRules((ds, set) -> set.size(),
datasource -> rules);
+
+ DruidCoordinatorRuntimeParams params = DruidCoordinatorRuntimeParams
+ .builder()
+ .withDruidCluster(cluster)
+ .withUsedSegments(segments)
+ .withBalancerStrategy(balancerStrategy)
+ .withDynamicConfigs(
+ CoordinatorDynamicConfig.builder()
+ .withSmartSegmentLoading(false)
+ .withUseRoundRobinSegmentAssignment(false)
+ .build()
+ )
+ .withSegmentAssignerUsing(loadQueueManager)
+ .build();
+
+ params = ruleRunner.run(params);
+ return params.getCoordinatorStats();
+ }
+
+ private static DruidCluster singleTierCluster()
+ {
+ return DruidCluster.builder().addTier(TIER, historical("hist1",
TIER)).build();
+ }
+
+ private static ServerHolder historical(String name, String tier)
+ {
+ final DruidServer server =
+ new DruidServer(name, name, null, 10L << 30, null,
ServerType.HISTORICAL, tier, 0);
+ return new ServerHolder(server.toImmutableDruidServer(), new
TestLoadQueuePeon());
+ }
+
+ private static DataSegment clusteredSegment(NumberedShardSpec shardSpec,
String tenant)
+ {
+ final ClusterGroupTuples groups = new ClusterGroupTuples(
+ RowSignature.builder().add("tenant", ColumnType.STRING).build(),
+ List.of(Collections.singletonList(tenant))
+ );
+ return segment(CHUNK, shardSpec, groups);
+ }
+
+ private static DataSegment tenantRegionSegment(
+ Interval interval,
+ NumberedShardSpec shardSpec,
+ String tenant,
+ String region
+ )
+ {
+ final ClusterGroupTuples groups = new ClusterGroupTuples(
+ RowSignature.builder().add("tenant", ColumnType.STRING).add("region",
ColumnType.STRING).build(),
+ List.of(List.of(tenant, region))
+ );
+ return segment(interval, shardSpec, groups);
+ }
+
+ private static DataSegment segment(
+ Interval interval,
+ NumberedShardSpec shardSpec,
+ ClusterGroupTuples groups
+ )
+ {
+ return DataSegment
+ .builder(SegmentId.of(DATASOURCE, interval, "v", shardSpec))
+ .shardSpec(shardSpec)
+ .loadSpec(Map.of("type", "local", "path", "/seg"))
+ .size(0)
+ .clusterGroups(groups)
+ .build();
+ }
+
+ /** A one-day chunk ending at "now" so an in-window P7D period rule applies
to it. */
+ private static Interval recentChunk()
+ {
+ final DateTime end = DateTimes.nowUtc();
+ return new Interval(end.minusDays(1), end);
+ }
+}
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/rules/WildcardClusterGroupPartialLoadMatcherTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/rules/WildcardClusterGroupPartialLoadMatcherTest.java
index 03f8ddc3221..4c69e98c3ab 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/rules/WildcardClusterGroupPartialLoadMatcherTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/rules/WildcardClusterGroupPartialLoadMatcherTest.java
@@ -484,10 +484,11 @@ class WildcardClusterGroupPartialLoadMatcherTest
}
@Test
- void testNoMatchOnAppendedSegmentReturnsNull()
+ void testNoMatchOnAppendedSegmentReturnsEmptyLoad()
{
- // Appended segment (partitionNum=2, numCorePartitions=2): not part of the
core partition group, so no
- // empty load. The matcher returns null and the rule's cannot-match
handling takes over.
+ // A clustered, compatible segment whose tuples match no pattern resolves
to the empty load regardless of whether
+ // it is a core or an appended (partitionNum >= numCorePartitions)
partition. The empty result keeps the segment
+ // announceable, so a fully-unmatched shard group can still be placed
rather than dropped from every tier.
final ClusterGroupTuples groups = new ClusterGroupTuples(
tenantRegion(),
List.of(List.of("acme", "us-east-1"))
@@ -497,14 +498,17 @@ class WildcardClusterGroupPartialLoadMatcherTest
List.of(Map.of("tenant", "nobody")),
null
);
- Assertions.assertNull(matcher.match(appended, BASE_LOAD_SPEC));
+ final PartialLoadMatcher.MatchResult result = matcher.match(appended,
BASE_LOAD_SPEC);
+ Assertions.assertNotNull(result);
+ Assertions.assertEquals("partialClusterGroup",
result.wrappedLoadSpec().get("type"));
+ Assertions.assertEquals(List.of(),
result.wrappedLoadSpec().get("clusterGroupIndices"));
}
@Test
- void testNoMatchOnSegmentWithoutCorePartitionsReturnsNull()
+ void testNoMatchOnSegmentWithoutCorePartitionsReturnsEmptyLoad()
{
- // numCorePartitions == 0: append-only ingestion has no core partition
group, so no completeness requirement.
- // The matcher returns null and the rule's cannot-match handling takes
over.
+ // numCorePartitions == 0 is treated like any other clustered, compatible
segment: an unmatched resolve returns
+ // the empty load so the segment can still be announced, not dropped.
final ClusterGroupTuples groups = new ClusterGroupTuples(
tenantRegion(),
List.of(List.of("acme", "us-east-1"))
@@ -514,7 +518,9 @@ class WildcardClusterGroupPartialLoadMatcherTest
List.of(Map.of("tenant", "nobody")),
null
);
- Assertions.assertNull(matcher.match(noCore, BASE_LOAD_SPEC));
+ final PartialLoadMatcher.MatchResult result = matcher.match(noCore,
BASE_LOAD_SPEC);
+ Assertions.assertNotNull(result);
+ Assertions.assertEquals(List.of(),
result.wrappedLoadSpec().get("clusterGroupIndices"));
}
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
