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 43383c7 refactor BalanceSegments#balanceServers to exit early if
there is no work to be done (#11768)
43383c7 is described below
commit 43383c73a84006d54022629475053b122994d4f0
Author: Lucas Capistrant <[email protected]>
AuthorDate: Mon Oct 25 10:06:35 2021 -0500
refactor BalanceSegments#balanceServers to exit early if there is no work
to be done (#11768)
* remove useless call to balanceServers for move from decom servers when
there are no decom servers
* refactor approach to this PR but accomplish the same thing
---
.../server/coordinator/duty/BalanceSegments.java | 10 ++++++++
.../server/coordinator/BalanceSegmentsTest.java | 27 ++++++++--------------
2 files changed, 20 insertions(+), 17 deletions(-)
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/duty/BalanceSegments.java
b/server/src/main/java/org/apache/druid/server/coordinator/duty/BalanceSegments.java
index 218cf69..12de017 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/duty/BalanceSegments.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/duty/BalanceSegments.java
@@ -144,6 +144,8 @@ public class BalanceSegments implements CoordinatorDuty
}
final int maxSegmentsToMove =
Math.min(params.getCoordinatorDynamicConfig().getMaxSegmentsToMove(),
numSegments);
+
+ // Prioritize moving segments from decomissioning servers.
int decommissioningMaxPercentOfMaxSegmentsToMove =
params.getCoordinatorDynamicConfig().getDecommissioningMaxPercentOfMaxSegmentsToMove();
int maxSegmentsToMoveFromDecommissioningNodes =
@@ -155,6 +157,7 @@ public class BalanceSegments implements CoordinatorDuty
Pair<Integer, Integer> decommissioningResult =
balanceServers(params, decommissioningServers, activeServers,
maxSegmentsToMoveFromDecommissioningNodes);
+ // After moving segments from decomissioning servers, move the remaining
segments from the rest of the servers.
int maxGeneralSegmentsToMove = maxSegmentsToMove -
decommissioningResult.lhs;
log.info("Processing %d segments for balancing between active servers",
maxGeneralSegmentsToMove);
Pair<Integer, Integer> generalResult =
@@ -184,6 +187,13 @@ public class BalanceSegments implements CoordinatorDuty
)
{
if (maxSegmentsToMove <= 0) {
+ log.debug("maxSegmentsToMove is 0; no balancing work can be performed.");
+ return new Pair<>(0, 0);
+ } else if (toMoveFrom.isEmpty()) {
+ log.debug("toMoveFrom is empty; no balancing work can be performed.");
+ return new Pair<>(0, 0);
+ } else if (toMoveTo.isEmpty()) {
+ log.debug("toMoveTo is empty; no balancing work can be peformed.");
return new Pair<>(0, 0);
}
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/BalanceSegmentsTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/BalanceSegmentsTest.java
index 76b1104..0221a37 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/BalanceSegmentsTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/BalanceSegmentsTest.java
@@ -236,7 +236,7 @@ public class BalanceSegmentsTest
EasyMock.expect(
strategy.pickSegmentsToMove(
ImmutableList.of(
- new ServerHolder(druidServer2, peon2, false)
+ new ServerHolder(druidServer2, peon2, true)
),
broadcastDatasources,
1,
@@ -276,6 +276,7 @@ public class BalanceSegmentsTest
.build();
params = new BalanceSegmentsTester(coordinator).run(params);
+ EasyMock.verify(strategy);
Assert.assertEquals(3L,
params.getCoordinatorStats().getTieredStat("movedCount", "normal"));
Assert.assertThat(
peon3.getSegmentsToLoad(),
@@ -316,9 +317,8 @@ public class BalanceSegmentsTest
mockCoordinator(coordinator);
BalancerStrategy strategy = EasyMock.createMock(BalancerStrategy.class);
- EasyMock.expect(strategy.pickSegmentsToMove(EasyMock.anyObject(),
EasyMock.anyObject(), EasyMock.anyInt(), EasyMock.anyInt()))
- .andReturn(ImmutableList.of(new
BalancerSegmentHolder(druidServer1, segment1)).iterator());
- EasyMock.expect(strategy.pickSegmentsToMove(EasyMock.anyObject(),
EasyMock.anyObject(), EasyMock.anyInt(), EasyMock.anyInt()))
+ EasyMock.expect(
+ strategy.pickSegmentsToMove(EasyMock.anyObject(),
EasyMock.anyObject(), EasyMock.anyInt(), EasyMock.anyInt()))
.andReturn(
ImmutableList.of(
new BalancerSegmentHolder(druidServer1, segment2),
@@ -346,10 +346,11 @@ public class BalanceSegmentsTest
.build();
params = new BalanceSegmentsTester(coordinator).run(params);
+ EasyMock.verify(strategy);
Assert.assertEquals(3L,
params.getCoordinatorStats().getTieredStat("movedCount", "normal"));
Assert.assertThat(
peon3.getSegmentsToLoad(),
- Matchers.is(Matchers.equalTo(ImmutableSet.of(segment1, segment2,
segment3)))
+ Matchers.is(Matchers.equalTo(ImmutableSet.of(segment2, segment3,
segment4)))
);
}
@@ -387,6 +388,7 @@ public class BalanceSegmentsTest
.build();
params = new BalanceSegmentsTester(coordinator).run(params);
+ EasyMock.verify(strategy);
Assert.assertEquals(0,
params.getCoordinatorStats().getTieredStat("movedCount", "normal"));
}
@@ -422,6 +424,7 @@ public class BalanceSegmentsTest
.build();
params = new BalanceSegmentsTester(coordinator).run(params);
+ EasyMock.verify(strategy);
Assert.assertEquals(1,
params.getCoordinatorStats().getTieredStat("movedCount", "normal"));
Assert.assertEquals(0, peon1.getNumberOfSegmentsInQueue());
Assert.assertEquals(1, peon2.getNumberOfSegmentsInQueue());
@@ -564,18 +567,7 @@ public class BalanceSegmentsTest
BalancerStrategy strategy = EasyMock.createMock(BalancerStrategy.class);
- // The first call for decommissioning servers
- EasyMock.expect(
- strategy.pickSegmentsToMove(
- ImmutableList.of(),
- broadcastDatasources,
- 1,
- 40
- )
- )
- .andReturn(Collections.emptyIterator());
-
- // The second call for the single non decommissioning server move
+ // Move from non-decomissioning servers
EasyMock.expect(
strategy.pickSegmentsToMove(
ImmutableList.of(
@@ -611,6 +603,7 @@ public class BalanceSegmentsTest
.build();
params = new BalanceSegmentsTester(coordinator).run(params);
+ EasyMock.verify(strategy);
Assert.assertEquals(1L,
params.getCoordinatorStats().getTieredStat("movedCount", "normal"));
Assert.assertThat(
peon3.getSegmentsToLoad(),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
