You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "github-code-scanning[bot] (via GitHub)" <gi...@apache.org> on 2023/04/11 05:07:47 UTC
[GitHub] [druid] github-code-scanning[bot] commented on a diff in pull request #13197: Segment loading: Allow cancellation and prioritization of load queue items
github-code-scanning[bot] commented on code in PR #13197:
URL: https://github.com/apache/druid/pull/13197#discussion_r1162306189
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/BalanceSegments.java:
##########
@@ -20,292 +20,365 @@
package org.apache.druid.server.coordinator.duty;
import com.google.common.collect.Lists;
-import org.apache.druid.client.ImmutableDruidServer;
-import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.client.ImmutableDruidDataSource;
import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.server.coordinator.BalancerSegmentHolder;
import org.apache.druid.server.coordinator.BalancerStrategy;
-import org.apache.druid.server.coordinator.CoordinatorStats;
-import org.apache.druid.server.coordinator.DruidCoordinator;
import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
-import org.apache.druid.server.coordinator.LoadPeonCallback;
-import org.apache.druid.server.coordinator.LoadQueuePeon;
+import org.apache.druid.server.coordinator.SegmentLoader;
+import org.apache.druid.server.coordinator.SegmentStateManager;
import org.apache.druid.server.coordinator.ServerHolder;
+import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
+import org.apache.druid.server.coordinator.stats.Stats;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.SegmentId;
-import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.NavigableSet;
+import java.util.Set;
import java.util.SortedSet;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
/**
+ *
*/
public class BalanceSegments implements CoordinatorDuty
{
protected static final EmittingLogger log = new EmittingLogger(BalanceSegments.class);
+ private final SegmentStateManager stateManager;
- protected final DruidCoordinator coordinator;
-
- protected final Map<String, ConcurrentHashMap<SegmentId, BalancerSegmentHolder>> currentlyMovingSegments =
- new HashMap<>();
-
- public BalanceSegments(DruidCoordinator coordinator)
+ public BalanceSegments(SegmentStateManager stateManager)
{
- this.coordinator = coordinator;
+ this.stateManager = stateManager;
}
- protected void reduceLifetimes(String tier)
+ /**
+ * Reduces the lifetimes of segments currently being moved in all the tiers.
+ * Raises alerts for segments stuck in queue
+ * Returns the set of tiers that are currently moving some segments and won't be
+ * eligible for assigning more balancing moves in this run.
+ */
+ private void reduceLifetimesAndAlert(int maxLifetime)
Review Comment:
## Useless parameter
The parameter 'maxLifetime' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/4746)
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java:
##########
@@ -439,18 +426,21 @@
// This forcing code should be revised
// when/if the autocompaction code policy to decide which segments to compact changes
if (dropExisting == null || !dropExisting) {
- if (segmentsToCompact.stream().allMatch(dataSegment -> dataSegment.isTombstone())) {
+ if (segmentsToCompact.stream().allMatch(DataSegment::isTombstone)) {
dropExisting = true;
- LOG.info("Forcing dropExisting to %s since all segments to compact are tombstones", dropExisting);
+ LOG.info("Forcing dropExisting to true since all segments to compact are tombstones.");
}
}
- // make tuningConfig
final String taskId = indexingServiceClient.compactSegments(
"coordinator-issued",
segmentsToCompact,
config.getTaskPriority(),
- ClientCompactionTaskQueryTuningConfig.from(config.getTuningConfig(), config.getMaxRowsPerSegment(), config.getMetricsSpec() != null),
+ ClientCompactionTaskQueryTuningConfig.from(
+ config.getTuningConfig(),
+ config.getMaxRowsPerSegment(),
Review Comment:
## Deprecated method or constructor invocation
Invoking [DataSourceCompactionConfig.getMaxRowsPerSegment](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/4366)
##########
server/src/test/java/org/apache/druid/server/coordinator/BalanceSegmentsTest.java:
##########
@@ -358,265 +257,172 @@
@Test
public void testMoveToDecommissioningServer()
{
- mockDruidServer(druidServer1, "1", "normal", 30L, 100L, segments);
- mockDruidServer(druidServer2, "2", "normal", 0L, 100L, Collections.emptyList());
-
- EasyMock.replay(druidServer3);
- EasyMock.replay(druidServer4);
-
- mockCoordinator(coordinator);
+ final ServerHolder serverHolder1 = createHolder(server1, false, allSegments);
+ final ServerHolder serverHolder2 = createHolder(server2, true);
BalancerStrategy strategy = EasyMock.createMock(BalancerStrategy.class);
- EasyMock.expect(strategy.pickSegmentsToMove(EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyInt()))
- .andReturn(ImmutableList.of(new BalancerSegmentHolder(druidServer1, segment1)).iterator())
- .anyTimes();
+ expectPickLoadingSegmentsAndReturnEmpty(strategy, 2);
+ expectPickLoadedSegmentsAndReturn(
+ strategy, 2,
+ new BalancerSegmentHolder(serverHolder1, segment1)
+ );
EasyMock.expect(strategy.findNewSegmentHomeBalancer(EasyMock.anyObject(), EasyMock.anyObject())).andAnswer(() -> {
List<ServerHolder> holders = (List<ServerHolder>) EasyMock.getCurrentArguments()[1];
return holders.get(0);
}).anyTimes();
EasyMock.replay(strategy);
- DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(
- ImmutableList.of(druidServer1, druidServer2),
- ImmutableList.of(peon1, peon2),
- ImmutableList.of(false, true)
- )
- .withBalancerStrategy(strategy)
- .withBroadcastDatasources(broadcastDatasources)
- .build();
+ DruidCoordinatorRuntimeParams params =
+ defaultRuntimeParamsBuilder(serverHolder1, serverHolder2)
+ .withBalancerStrategy(strategy)
+ .withBroadcastDatasources(broadcastDatasources)
+ .build();
- params = new BalanceSegmentsTester(coordinator).run(params);
+ params = new BalanceSegments(stateManager).run(params);
EasyMock.verify(strategy);
- Assert.assertEquals(0, params.getCoordinatorStats().getTieredStat("movedCount", "normal"));
+ Assert.assertEquals(0, params.getCoordinatorStats().getTieredStat(Stats.Segments.MOVED, "normal"));
}
@Test
public void testMoveFromDecommissioningServer()
{
- mockDruidServer(druidServer1, "1", "normal", 30L, 100L, segments);
- mockDruidServer(druidServer2, "2", "normal", 0L, 100L, Collections.emptyList());
-
- EasyMock.replay(druidServer3);
- EasyMock.replay(druidServer4);
-
- mockCoordinator(coordinator);
+ final ServerHolder holder1 = createHolder(server1, allSegments);
+ final ServerHolder holder2 = createHolder(server2);
- ServerHolder holder2 = new ServerHolder(druidServer2, peon2, false);
BalancerStrategy strategy = EasyMock.createMock(BalancerStrategy.class);
- EasyMock.expect(strategy.pickSegmentsToMove(EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyInt()))
- .andReturn(ImmutableList.of(new BalancerSegmentHolder(druidServer1, segment1)).iterator())
- .once();
+ expectPickLoadingSegmentsAndReturnEmpty(strategy, 1);
+ expectPickLoadedSegmentsAndReturn(
+ strategy, 1,
+ new BalancerSegmentHolder(holder1, segment1)
+ );
EasyMock.expect(strategy.findNewSegmentHomeBalancer(EasyMock.anyObject(), EasyMock.anyObject()))
.andReturn(holder2)
.once();
EasyMock.replay(strategy);
- DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(
- ImmutableList.of(druidServer1, druidServer2),
- ImmutableList.of(peon1, peon2),
- ImmutableList.of(true, false)
- )
+ DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(holder1, holder2)
.withDynamicConfigs(CoordinatorDynamicConfig.builder().withMaxSegmentsToMove(1).build())
.withBalancerStrategy(strategy)
.withBroadcastDatasources(broadcastDatasources)
.build();
- params = new BalanceSegmentsTester(coordinator).run(params);
+ params = new BalanceSegments(stateManager).run(params);
EasyMock.verify(strategy);
- Assert.assertEquals(1, params.getCoordinatorStats().getTieredStat("movedCount", "normal"));
- Assert.assertEquals(0, peon1.getNumberOfSegmentsInQueue());
- Assert.assertEquals(1, peon2.getNumberOfSegmentsInQueue());
+ Assert.assertEquals(1, params.getCoordinatorStats().getTieredStat(Stats.Segments.MOVED, "normal"));
+ Assert.assertEquals(0, holder1.getPeon().getNumberOfSegmentsToLoad());
+ Assert.assertEquals(1, holder2.getPeon().getNumberOfSegmentsToLoad());
}
@Test
public void testMoveMaxLoadQueueServerBalancer()
{
- mockDruidServer(druidServer1, "1", "normal", 30L, 100L, segments);
- mockDruidServer(druidServer2, "2", "normal", 0L, 100L, Collections.emptyList());
-
- EasyMock.replay(druidServer3);
- EasyMock.replay(druidServer4);
-
- // Mock stuff that the coordinator needs
- mockCoordinator(coordinator);
-
- BalancerStrategy predefinedPickOrderStrategy = new PredefinedPickOrderBalancerStrategy(
- balancerStrategy,
- ImmutableList.of(
- new BalancerSegmentHolder(druidServer1, segment1),
- new BalancerSegmentHolder(druidServer1, segment2),
- new BalancerSegmentHolder(druidServer1, segment3),
- new BalancerSegmentHolder(druidServer1, segment4)
+ final int maxSegmentsInQueue = 1;
+ final ServerHolder holder1 = createHolder(server1, maxSegmentsInQueue, allSegments);
+ final ServerHolder holder2 = createHolder(server2, maxSegmentsInQueue);
+
+ final CoordinatorDynamicConfig dynamicConfig = CoordinatorDynamicConfig
+ .builder()
+ .withMaxSegmentsToMove(MAX_SEGMENTS_TO_MOVE)
+ .withMaxSegmentsInNodeLoadingQueue(maxSegmentsInQueue)
+ .build();
+ DruidCoordinatorRuntimeParams params = CoordinatorRuntimeParamsTestHelpers
+ .newBuilder()
+ .withDruidCluster(
+ DruidClusterBuilder.newBuilder().addTier("normal", holder1, holder2).build()
)
- );
-
- DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(
- ImmutableList.of(druidServer1, druidServer2),
- ImmutableList.of(peon1, peon2)
- )
- .withBalancerStrategy(predefinedPickOrderStrategy)
- .withBroadcastDatasources(broadcastDatasources)
- .withDynamicConfigs(
- CoordinatorDynamicConfig
- .builder()
- .withMaxSegmentsToMove(MAX_SEGMENTS_TO_MOVE)
- .withMaxSegmentsInNodeLoadingQueue(1)
- .build()
+ .withLoadManagementPeons(
+ ImmutableMap.of(server1.getName(), holder1.getPeon(), server2.getName(), holder2.getPeon())
)
+ .withUsedSegmentsInTest(allSegments)
+ .withBalancerStrategy(balancerStrategy)
+ .withBroadcastDatasources(broadcastDatasources)
+ .withDynamicConfigs(dynamicConfig)
+ .withReplicationManager(createReplicationThrottler(dynamicConfig))
.build();
- params = new BalanceSegmentsTester(coordinator).run(params);
+ params = new BalanceSegments(stateManager).run(params);
// max to move is 5, all segments on server 1, but only expect to move 1 to server 2 since max node load queue is 1
- Assert.assertEquals(1, params.getCoordinatorStats().getTieredStat("movedCount", "normal"));
- }
-
- @Test
- public void testMoveSameSegmentTwice()
- {
- mockDruidServer(druidServer1, "1", "normal", 30L, 100L, segments);
- mockDruidServer(druidServer2, "2", "normal", 0L, 100L, Collections.emptyList());
-
- EasyMock.replay(druidServer3);
- EasyMock.replay(druidServer4);
-
- // Mock stuff that the coordinator needs
- mockCoordinator(coordinator);
-
- BalancerStrategy predefinedPickOrderStrategy = new PredefinedPickOrderBalancerStrategy(
- balancerStrategy,
- ImmutableList.of(
- new BalancerSegmentHolder(druidServer1, segment1)
- )
- );
-
- DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(
- ImmutableList.of(druidServer1, druidServer2),
- ImmutableList.of(peon1, peon2)
- )
- .withBalancerStrategy(predefinedPickOrderStrategy)
- .withBroadcastDatasources(broadcastDatasources)
- .withDynamicConfigs(
- CoordinatorDynamicConfig.builder().withMaxSegmentsToMove(
- 2
- ).build()
- )
- .build();
-
- params = new BalanceSegmentsTester(coordinator).run(params);
- Assert.assertEquals(1, params.getCoordinatorStats().getTieredStat("movedCount", "normal"));
+ Assert.assertEquals(1, params.getCoordinatorStats().getTieredStat(Stats.Segments.MOVED, "normal"));
}
@Test
public void testRun1()
{
// Mock some servers of different usages
- mockDruidServer(druidServer1, "1", "normal", 30L, 100L, segments);
- mockDruidServer(druidServer2, "2", "normal", 0L, 100L, Collections.emptyList());
-
- EasyMock.replay(druidServer3);
- EasyMock.replay(druidServer4);
-
- // Mock stuff that the coordinator needs
- mockCoordinator(coordinator);
-
DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(
- ImmutableList.of(druidServer1, druidServer2),
- ImmutableList.of(peon1, peon2)
+ createHolder(server1, allSegments),
+ createHolder(server2)
).build();
- params = new BalanceSegmentsTester(coordinator).run(params);
- Assert.assertTrue(params.getCoordinatorStats().getTieredStat("movedCount", "normal") > 0);
+ params = new BalanceSegments(stateManager).run(params);
+ Assert.assertTrue(params.getCoordinatorStats().getTieredStat(Stats.Segments.MOVED, "normal") > 0);
}
@Test
public void testRun2()
{
// Mock some servers of different usages
- mockDruidServer(druidServer1, "1", "normal", 30L, 100L, segments);
- mockDruidServer(druidServer2, "2", "normal", 0L, 100L, Collections.emptyList());
- mockDruidServer(druidServer3, "3", "normal", 0L, 100L, Collections.emptyList());
- mockDruidServer(druidServer4, "4", "normal", 0L, 100L, Collections.emptyList());
-
- // Mock stuff that the coordinator needs
- mockCoordinator(coordinator);
-
- DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(druidServers, peons).build();
+ DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(
+ createHolder(server1, allSegments),
+ createHolder(server2),
+ createHolder(server3),
+ createHolder(server4)
+ ).build();
- params = new BalanceSegmentsTester(coordinator).run(params);
- Assert.assertTrue(params.getCoordinatorStats().getTieredStat("movedCount", "normal") > 0);
+ params = new BalanceSegments(stateManager).run(params);
+ Assert.assertTrue(params.getCoordinatorStats().getTieredStat(Stats.Segments.MOVED, "normal") > 0);
}
- /**
- * Testing that the dynamic coordinator config value, percentOfSegmentsToConsiderPerMove, is honored when calling
- * out to pickSegmentToMove. This config limits the number of segments that are considered when looking for a segment
- * to move.
- */
@Test
- public void testThatDynamicConfigIsHonoredWhenPickingSegmentToMove()
+ public void testThatMaxSegmentsToMoveIsHonored()
{
- mockDruidServer(druidServer1, "1", "normal", 50L, 100L, Arrays.asList(segment1, segment2));
- mockDruidServer(druidServer2, "2", "normal", 30L, 100L, Arrays.asList(segment3, segment4));
- mockDruidServer(druidServer3, "3", "normal", 0L, 100L, Collections.emptyList());
-
- EasyMock.replay(druidServer4);
-
- mockCoordinator(coordinator);
+ // Move from non-decomissioning servers
+ final ServerHolder holder1 = createHolder(server1, segment1, segment2);
+ final ServerHolder holder2 = createHolder(server2, segment3, segment4);
+ final ServerHolder holder3 = createHolder(server3);
BalancerStrategy strategy = EasyMock.createMock(BalancerStrategy.class);
-
- // Move from non-decomissioning servers
- EasyMock.expect(
- strategy.pickSegmentsToMove(
- ImmutableList.of(
- new ServerHolder(druidServer3, peon3, false),
- new ServerHolder(druidServer2, peon2, false),
- new ServerHolder(druidServer1, peon1, false)
- ),
- broadcastDatasources,
- 40.0
- )
- )
- .andReturn(ImmutableList.of(new BalancerSegmentHolder(druidServer2, segment3)).iterator());
+ expectPickLoadingSegmentsAndReturnEmpty(strategy, 1);
+ expectPickLoadedSegmentsAndReturn(
+ strategy,
+ 1,
+ new BalancerSegmentHolder(holder2, segment3)
+ );
EasyMock.expect(strategy.findNewSegmentHomeBalancer(EasyMock.anyObject(), EasyMock.anyObject()))
- .andReturn(new ServerHolder(druidServer3, peon3))
+ .andReturn(holder3)
.anyTimes();
EasyMock.replay(strategy);
- DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(
- ImmutableList.of(druidServer1, druidServer2, druidServer3),
- ImmutableList.of(peon1, peon2, peon3),
- ImmutableList.of(false, false, false)
- )
- .withDynamicConfigs(
- CoordinatorDynamicConfig.builder()
- .withMaxSegmentsToMove(1)
- .withUseBatchedSegmentSampler(false)
- .withPercentOfSegmentsToConsiderPerMove(40)
- .build()
- )
- .withBalancerStrategy(strategy)
- .withBroadcastDatasources(broadcastDatasources)
- .build();
-
- params = new BalanceSegmentsTester(coordinator).run(params);
+ DruidCoordinatorRuntimeParams params =
+ defaultRuntimeParamsBuilder(holder1, holder2, holder3)
+ .withDynamicConfigs(
+ CoordinatorDynamicConfig.builder()
+ .withMaxSegmentsToMove(1)
+ .withUseBatchedSegmentSampler(true)
+ .withPercentOfSegmentsToConsiderPerMove(40)
Review Comment:
## Deprecated method or constructor invocation
Invoking [Builder.withPercentOfSegmentsToConsiderPerMove](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/4744)
##########
server/src/test/java/org/apache/druid/server/coordinator/BalanceSegmentsTest.java:
##########
@@ -358,265 +257,172 @@
@Test
public void testMoveToDecommissioningServer()
{
- mockDruidServer(druidServer1, "1", "normal", 30L, 100L, segments);
- mockDruidServer(druidServer2, "2", "normal", 0L, 100L, Collections.emptyList());
-
- EasyMock.replay(druidServer3);
- EasyMock.replay(druidServer4);
-
- mockCoordinator(coordinator);
+ final ServerHolder serverHolder1 = createHolder(server1, false, allSegments);
+ final ServerHolder serverHolder2 = createHolder(server2, true);
BalancerStrategy strategy = EasyMock.createMock(BalancerStrategy.class);
- EasyMock.expect(strategy.pickSegmentsToMove(EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyInt()))
- .andReturn(ImmutableList.of(new BalancerSegmentHolder(druidServer1, segment1)).iterator())
- .anyTimes();
+ expectPickLoadingSegmentsAndReturnEmpty(strategy, 2);
+ expectPickLoadedSegmentsAndReturn(
+ strategy, 2,
+ new BalancerSegmentHolder(serverHolder1, segment1)
+ );
EasyMock.expect(strategy.findNewSegmentHomeBalancer(EasyMock.anyObject(), EasyMock.anyObject())).andAnswer(() -> {
List<ServerHolder> holders = (List<ServerHolder>) EasyMock.getCurrentArguments()[1];
return holders.get(0);
}).anyTimes();
EasyMock.replay(strategy);
- DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(
- ImmutableList.of(druidServer1, druidServer2),
- ImmutableList.of(peon1, peon2),
- ImmutableList.of(false, true)
- )
- .withBalancerStrategy(strategy)
- .withBroadcastDatasources(broadcastDatasources)
- .build();
+ DruidCoordinatorRuntimeParams params =
+ defaultRuntimeParamsBuilder(serverHolder1, serverHolder2)
+ .withBalancerStrategy(strategy)
+ .withBroadcastDatasources(broadcastDatasources)
+ .build();
- params = new BalanceSegmentsTester(coordinator).run(params);
+ params = new BalanceSegments(stateManager).run(params);
EasyMock.verify(strategy);
- Assert.assertEquals(0, params.getCoordinatorStats().getTieredStat("movedCount", "normal"));
+ Assert.assertEquals(0, params.getCoordinatorStats().getTieredStat(Stats.Segments.MOVED, "normal"));
}
@Test
public void testMoveFromDecommissioningServer()
{
- mockDruidServer(druidServer1, "1", "normal", 30L, 100L, segments);
- mockDruidServer(druidServer2, "2", "normal", 0L, 100L, Collections.emptyList());
-
- EasyMock.replay(druidServer3);
- EasyMock.replay(druidServer4);
-
- mockCoordinator(coordinator);
+ final ServerHolder holder1 = createHolder(server1, allSegments);
+ final ServerHolder holder2 = createHolder(server2);
- ServerHolder holder2 = new ServerHolder(druidServer2, peon2, false);
BalancerStrategy strategy = EasyMock.createMock(BalancerStrategy.class);
- EasyMock.expect(strategy.pickSegmentsToMove(EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyInt()))
- .andReturn(ImmutableList.of(new BalancerSegmentHolder(druidServer1, segment1)).iterator())
- .once();
+ expectPickLoadingSegmentsAndReturnEmpty(strategy, 1);
+ expectPickLoadedSegmentsAndReturn(
+ strategy, 1,
+ new BalancerSegmentHolder(holder1, segment1)
+ );
EasyMock.expect(strategy.findNewSegmentHomeBalancer(EasyMock.anyObject(), EasyMock.anyObject()))
.andReturn(holder2)
.once();
EasyMock.replay(strategy);
- DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(
- ImmutableList.of(druidServer1, druidServer2),
- ImmutableList.of(peon1, peon2),
- ImmutableList.of(true, false)
- )
+ DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(holder1, holder2)
.withDynamicConfigs(CoordinatorDynamicConfig.builder().withMaxSegmentsToMove(1).build())
.withBalancerStrategy(strategy)
.withBroadcastDatasources(broadcastDatasources)
.build();
- params = new BalanceSegmentsTester(coordinator).run(params);
+ params = new BalanceSegments(stateManager).run(params);
EasyMock.verify(strategy);
- Assert.assertEquals(1, params.getCoordinatorStats().getTieredStat("movedCount", "normal"));
- Assert.assertEquals(0, peon1.getNumberOfSegmentsInQueue());
- Assert.assertEquals(1, peon2.getNumberOfSegmentsInQueue());
+ Assert.assertEquals(1, params.getCoordinatorStats().getTieredStat(Stats.Segments.MOVED, "normal"));
+ Assert.assertEquals(0, holder1.getPeon().getNumberOfSegmentsToLoad());
+ Assert.assertEquals(1, holder2.getPeon().getNumberOfSegmentsToLoad());
}
@Test
public void testMoveMaxLoadQueueServerBalancer()
{
- mockDruidServer(druidServer1, "1", "normal", 30L, 100L, segments);
- mockDruidServer(druidServer2, "2", "normal", 0L, 100L, Collections.emptyList());
-
- EasyMock.replay(druidServer3);
- EasyMock.replay(druidServer4);
-
- // Mock stuff that the coordinator needs
- mockCoordinator(coordinator);
-
- BalancerStrategy predefinedPickOrderStrategy = new PredefinedPickOrderBalancerStrategy(
- balancerStrategy,
- ImmutableList.of(
- new BalancerSegmentHolder(druidServer1, segment1),
- new BalancerSegmentHolder(druidServer1, segment2),
- new BalancerSegmentHolder(druidServer1, segment3),
- new BalancerSegmentHolder(druidServer1, segment4)
+ final int maxSegmentsInQueue = 1;
+ final ServerHolder holder1 = createHolder(server1, maxSegmentsInQueue, allSegments);
+ final ServerHolder holder2 = createHolder(server2, maxSegmentsInQueue);
+
+ final CoordinatorDynamicConfig dynamicConfig = CoordinatorDynamicConfig
+ .builder()
+ .withMaxSegmentsToMove(MAX_SEGMENTS_TO_MOVE)
+ .withMaxSegmentsInNodeLoadingQueue(maxSegmentsInQueue)
+ .build();
+ DruidCoordinatorRuntimeParams params = CoordinatorRuntimeParamsTestHelpers
+ .newBuilder()
+ .withDruidCluster(
+ DruidClusterBuilder.newBuilder().addTier("normal", holder1, holder2).build()
)
- );
-
- DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(
- ImmutableList.of(druidServer1, druidServer2),
- ImmutableList.of(peon1, peon2)
- )
- .withBalancerStrategy(predefinedPickOrderStrategy)
- .withBroadcastDatasources(broadcastDatasources)
- .withDynamicConfigs(
- CoordinatorDynamicConfig
- .builder()
- .withMaxSegmentsToMove(MAX_SEGMENTS_TO_MOVE)
- .withMaxSegmentsInNodeLoadingQueue(1)
- .build()
+ .withLoadManagementPeons(
+ ImmutableMap.of(server1.getName(), holder1.getPeon(), server2.getName(), holder2.getPeon())
)
+ .withUsedSegmentsInTest(allSegments)
+ .withBalancerStrategy(balancerStrategy)
+ .withBroadcastDatasources(broadcastDatasources)
+ .withDynamicConfigs(dynamicConfig)
+ .withReplicationManager(createReplicationThrottler(dynamicConfig))
.build();
- params = new BalanceSegmentsTester(coordinator).run(params);
+ params = new BalanceSegments(stateManager).run(params);
// max to move is 5, all segments on server 1, but only expect to move 1 to server 2 since max node load queue is 1
- Assert.assertEquals(1, params.getCoordinatorStats().getTieredStat("movedCount", "normal"));
- }
-
- @Test
- public void testMoveSameSegmentTwice()
- {
- mockDruidServer(druidServer1, "1", "normal", 30L, 100L, segments);
- mockDruidServer(druidServer2, "2", "normal", 0L, 100L, Collections.emptyList());
-
- EasyMock.replay(druidServer3);
- EasyMock.replay(druidServer4);
-
- // Mock stuff that the coordinator needs
- mockCoordinator(coordinator);
-
- BalancerStrategy predefinedPickOrderStrategy = new PredefinedPickOrderBalancerStrategy(
- balancerStrategy,
- ImmutableList.of(
- new BalancerSegmentHolder(druidServer1, segment1)
- )
- );
-
- DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(
- ImmutableList.of(druidServer1, druidServer2),
- ImmutableList.of(peon1, peon2)
- )
- .withBalancerStrategy(predefinedPickOrderStrategy)
- .withBroadcastDatasources(broadcastDatasources)
- .withDynamicConfigs(
- CoordinatorDynamicConfig.builder().withMaxSegmentsToMove(
- 2
- ).build()
- )
- .build();
-
- params = new BalanceSegmentsTester(coordinator).run(params);
- Assert.assertEquals(1, params.getCoordinatorStats().getTieredStat("movedCount", "normal"));
+ Assert.assertEquals(1, params.getCoordinatorStats().getTieredStat(Stats.Segments.MOVED, "normal"));
}
@Test
public void testRun1()
{
// Mock some servers of different usages
- mockDruidServer(druidServer1, "1", "normal", 30L, 100L, segments);
- mockDruidServer(druidServer2, "2", "normal", 0L, 100L, Collections.emptyList());
-
- EasyMock.replay(druidServer3);
- EasyMock.replay(druidServer4);
-
- // Mock stuff that the coordinator needs
- mockCoordinator(coordinator);
-
DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(
- ImmutableList.of(druidServer1, druidServer2),
- ImmutableList.of(peon1, peon2)
+ createHolder(server1, allSegments),
+ createHolder(server2)
).build();
- params = new BalanceSegmentsTester(coordinator).run(params);
- Assert.assertTrue(params.getCoordinatorStats().getTieredStat("movedCount", "normal") > 0);
+ params = new BalanceSegments(stateManager).run(params);
+ Assert.assertTrue(params.getCoordinatorStats().getTieredStat(Stats.Segments.MOVED, "normal") > 0);
}
@Test
public void testRun2()
{
// Mock some servers of different usages
- mockDruidServer(druidServer1, "1", "normal", 30L, 100L, segments);
- mockDruidServer(druidServer2, "2", "normal", 0L, 100L, Collections.emptyList());
- mockDruidServer(druidServer3, "3", "normal", 0L, 100L, Collections.emptyList());
- mockDruidServer(druidServer4, "4", "normal", 0L, 100L, Collections.emptyList());
-
- // Mock stuff that the coordinator needs
- mockCoordinator(coordinator);
-
- DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(druidServers, peons).build();
+ DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(
+ createHolder(server1, allSegments),
+ createHolder(server2),
+ createHolder(server3),
+ createHolder(server4)
+ ).build();
- params = new BalanceSegmentsTester(coordinator).run(params);
- Assert.assertTrue(params.getCoordinatorStats().getTieredStat("movedCount", "normal") > 0);
+ params = new BalanceSegments(stateManager).run(params);
+ Assert.assertTrue(params.getCoordinatorStats().getTieredStat(Stats.Segments.MOVED, "normal") > 0);
}
- /**
- * Testing that the dynamic coordinator config value, percentOfSegmentsToConsiderPerMove, is honored when calling
- * out to pickSegmentToMove. This config limits the number of segments that are considered when looking for a segment
- * to move.
- */
@Test
- public void testThatDynamicConfigIsHonoredWhenPickingSegmentToMove()
+ public void testThatMaxSegmentsToMoveIsHonored()
{
- mockDruidServer(druidServer1, "1", "normal", 50L, 100L, Arrays.asList(segment1, segment2));
- mockDruidServer(druidServer2, "2", "normal", 30L, 100L, Arrays.asList(segment3, segment4));
- mockDruidServer(druidServer3, "3", "normal", 0L, 100L, Collections.emptyList());
-
- EasyMock.replay(druidServer4);
-
- mockCoordinator(coordinator);
+ // Move from non-decomissioning servers
+ final ServerHolder holder1 = createHolder(server1, segment1, segment2);
+ final ServerHolder holder2 = createHolder(server2, segment3, segment4);
+ final ServerHolder holder3 = createHolder(server3);
BalancerStrategy strategy = EasyMock.createMock(BalancerStrategy.class);
-
- // Move from non-decomissioning servers
- EasyMock.expect(
- strategy.pickSegmentsToMove(
- ImmutableList.of(
- new ServerHolder(druidServer3, peon3, false),
- new ServerHolder(druidServer2, peon2, false),
- new ServerHolder(druidServer1, peon1, false)
- ),
- broadcastDatasources,
- 40.0
- )
- )
- .andReturn(ImmutableList.of(new BalancerSegmentHolder(druidServer2, segment3)).iterator());
+ expectPickLoadingSegmentsAndReturnEmpty(strategy, 1);
+ expectPickLoadedSegmentsAndReturn(
+ strategy,
+ 1,
+ new BalancerSegmentHolder(holder2, segment3)
+ );
EasyMock.expect(strategy.findNewSegmentHomeBalancer(EasyMock.anyObject(), EasyMock.anyObject()))
- .andReturn(new ServerHolder(druidServer3, peon3))
+ .andReturn(holder3)
.anyTimes();
EasyMock.replay(strategy);
- DruidCoordinatorRuntimeParams params = defaultRuntimeParamsBuilder(
- ImmutableList.of(druidServer1, druidServer2, druidServer3),
- ImmutableList.of(peon1, peon2, peon3),
- ImmutableList.of(false, false, false)
- )
- .withDynamicConfigs(
- CoordinatorDynamicConfig.builder()
- .withMaxSegmentsToMove(1)
- .withUseBatchedSegmentSampler(false)
- .withPercentOfSegmentsToConsiderPerMove(40)
- .build()
- )
- .withBalancerStrategy(strategy)
- .withBroadcastDatasources(broadcastDatasources)
- .build();
-
- params = new BalanceSegmentsTester(coordinator).run(params);
+ DruidCoordinatorRuntimeParams params =
+ defaultRuntimeParamsBuilder(holder1, holder2, holder3)
+ .withDynamicConfigs(
+ CoordinatorDynamicConfig.builder()
+ .withMaxSegmentsToMove(1)
+ .withUseBatchedSegmentSampler(true)
Review Comment:
## Deprecated method or constructor invocation
Invoking [Builder.withUseBatchedSegmentSampler](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/4745)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org