You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by kf...@apache.org on 2023/07/31 12:26:54 UTC
[druid] branch 27.0.0 updated: [Backport 27.0.0] Fix reported replication factor of segment with zero required replicas (#14704)
This is an automated email from the ASF dual-hosted git repository.
kfaraz pushed a commit to branch 27.0.0
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/27.0.0 by this push:
new 9d67a67a7e [Backport 27.0.0] Fix reported replication factor of segment with zero required replicas (#14704)
9d67a67a7e is described below
commit 9d67a67a7e462587332d85064bc2df0cca347969
Author: Kashif Faraz <ka...@gmail.com>
AuthorDate: Mon Jul 31 17:56:47 2023 +0530
[Backport 27.0.0] Fix reported replication factor of segment with zero required replicas (#14704)
* Fix reported replication factor of segment with zero required replicas (#14701)
* Use LoadQueuePeonTester instead of TestLoadQueuePeon
---
.../loading/StrategicSegmentAssigner.java | 25 +++++++++------
.../server/coordinator/duty/RunRulesTest.java | 36 ++++++++++++++++++++++
2 files changed, 52 insertions(+), 9 deletions(-)
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 6546cce69c..038707a19f 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
@@ -21,6 +21,7 @@ package org.apache.druid.server.coordinator.loading;
import com.google.common.collect.Sets;
import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
+import org.apache.druid.client.DruidServer;
import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.server.coordinator.DruidCluster;
import org.apache.druid.server.coordinator.ServerHolder;
@@ -194,18 +195,24 @@ public class StrategicSegmentAssigner implements SegmentActionHandler
@Override
public void replicateSegment(DataSegment segment, Map<String, Integer> tierToReplicaCount)
{
- // Identify empty tiers and determine total required replicas
final Set<String> allTiersInCluster = Sets.newHashSet(cluster.getTierNames());
- tierToReplicaCount.forEach((tier, requiredReplicas) -> {
- reportTierCapacityStats(segment, requiredReplicas, tier);
- SegmentReplicaCount replicaCount = replicaCountMap.computeIfAbsent(segment.getId(), tier);
- replicaCount.setRequired(requiredReplicas, tierToHistoricalCount.getOrDefault(tier, 0));
+ if (tierToReplicaCount == null || tierToReplicaCount.isEmpty()) {
+ // Track the counts for a segment even if it requires 0 replicas on all tiers
+ replicaCountMap.computeIfAbsent(segment.getId(), DruidServer.DEFAULT_TIER);
+ } else {
+ // Identify empty tiers and determine total required replicas
+ tierToReplicaCount.forEach((tier, requiredReplicas) -> {
+ reportTierCapacityStats(segment, requiredReplicas, tier);
- if (!allTiersInCluster.contains(tier)) {
- tiersWithNoServer.add(tier);
- }
- });
+ SegmentReplicaCount replicaCount = replicaCountMap.computeIfAbsent(segment.getId(), tier);
+ replicaCount.setRequired(requiredReplicas, tierToHistoricalCount.getOrDefault(tier, 0));
+
+ if (!allTiersInCluster.contains(tier)) {
+ tiersWithNoServer.add(tier);
+ }
+ });
+ }
SegmentReplicaCount replicaCountInCluster = replicaCountMap.getTotal(segment.getId());
final int replicaSurplus = replicaCountInCluster.loadedNotDropping()
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesTest.java
index 46240a340d..5f63064ff6 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesTest.java
@@ -45,7 +45,10 @@ import org.apache.druid.server.coordinator.ServerHolder;
import org.apache.druid.server.coordinator.balancer.CostBalancerStrategy;
import org.apache.druid.server.coordinator.balancer.RandomBalancerStrategy;
import org.apache.druid.server.coordinator.loading.LoadQueuePeon;
+import org.apache.druid.server.coordinator.loading.LoadQueuePeonTester;
import org.apache.druid.server.coordinator.loading.SegmentLoadQueueManager;
+import org.apache.druid.server.coordinator.loading.SegmentReplicaCount;
+import org.apache.druid.server.coordinator.loading.SegmentReplicationStatus;
import org.apache.druid.server.coordinator.rules.ForeverLoadRule;
import org.apache.druid.server.coordinator.rules.IntervalDropRule;
import org.apache.druid.server.coordinator.rules.IntervalLoadRule;
@@ -1262,6 +1265,39 @@ public class RunRulesTest
EasyMock.verify(mockPeon);
}
+ @Test
+ public void testSegmentWithZeroRequiredReplicasHasZeroReplicationFactor()
+ {
+ EasyMock.expect(databaseRuleManager.getRulesWithDefault(EasyMock.anyObject())).andReturn(
+ Collections.singletonList(
+ new ForeverLoadRule(Collections.emptyMap(), false)
+ )
+ ).anyTimes();
+ EasyMock.replay(databaseRuleManager);
+
+ final DruidCluster cluster = DruidCluster
+ .builder()
+ .add(createServerHolder("server", "normal", new LoadQueuePeonTester()))
+ .build();
+
+ final DataSegment segment = usedSegments.get(0);
+ DruidCoordinatorRuntimeParams params = createCoordinatorRuntimeParams(cluster, segment)
+ .withBalancerStrategy(new RandomBalancerStrategy())
+ .withSegmentAssignerUsing(loadQueueManager)
+ .build();
+ params = ruleRunner.run(params);
+
+ Assert.assertNotNull(params);
+ SegmentReplicationStatus replicationStatus = params.getSegmentReplicationStatus();
+ Assert.assertNotNull(replicationStatus);
+
+ SegmentReplicaCount replicaCounts = replicationStatus.getReplicaCountsInCluster(segment.getId());
+ Assert.assertNotNull(replicaCounts);
+ Assert.assertEquals(0, replicaCounts.required());
+ Assert.assertEquals(0, replicaCounts.totalLoaded());
+ Assert.assertEquals(0, replicaCounts.requiredAndLoadable());
+ }
+
private CoordinatorRunStats runDutyAndGetStats(DruidCoordinatorRuntimeParams params)
{
params = ruleRunner.run(params);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org