luciano/amazon-kinesis-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

updated logic to count based variance balancing

Browse source
This commit is contained in:
eha sah 2025-04-10 09:08:21 -07:00
parent 6b19b6e996
commit f9465637ae
3 changed files with 14 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/assignment/LeaseAssignmentManager.java
View file

@ -279,16 +279,14 @@ public final class LeaseAssignmentManager {
} }
private boolean shouldRunVarianceBalancing() { private boolean shouldRunVarianceBalancing() {
final long nowNanos = nanoTimeProvider.get(); final boolean response = this.lamRunCounter == 0;
final long intervalMillis = leaseDurationMillis * config.varianceBalancingFrequency(); /*
To avoid lamRunCounter grow large, keep it within [0,varianceBalancingFrequency).
final long elapsedMillis = Math.abs(nowNanos - varianceBasedBalancingLastRunTime) / 1_000_000; If varianceBalancingFrequency is 5 lamRunCounter value will be within 0 to 4 and method return true when
lamRunCounter is 0.
if (elapsedMillis >= intervalMillis) { */
varianceBasedBalancingLastRunTime = nowNanos; this.lamRunCounter = (this.lamRunCounter + 1) % config.varianceBalancingFrequency();
return true; return response;
}
return false;
} }
/** /**

8
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseManagementConfig.java
View file

@ -579,11 +579,11 @@ public class LeaseManagementConfig {
private WorkerMetricsTableConfig workerMetricsTableConfig; private WorkerMetricsTableConfig workerMetricsTableConfig;
/** /**
* Frequency to perform worker variance balancing. This value is used with respect to the failoverTimeMillis, * Frequency to perform worker variance balancing. This value is used with respect to the LAM frequency,
* that is every six (as default) * failoverTimeMillis the worker variance balancing will be performed. * that is every sixth (as default) iteration of LAM the worker variance balancing will be performed.
* Setting it to 1 will make varianceBalancing run on every failoverTimeMillis and 2 on every 2 * failoverTimeMillis * Setting it to 1 will make varianceBalancing run on every iteration of LAM and 2 on every 2nd iteration
* and so on. * and so on.
* NOTE: LAM frequency = {@link LeaseManagementConfig#leaseAssignmentIntervalMillis} * NOTE: LAM frequency = failoverTimeMillis
*/ */
private int varianceBalancingFrequency = 6; private int varianceBalancingFrequency = 6;

13
amazon-kinesis-client/src/test/java/software/amazon/kinesis/coordinator/assignment/LeaseAssignmentManagerTest.java
View file

@ -434,7 +434,7 @@ class LeaseAssignmentManagerTest {
} }
// no needed since variance based load balancing is no longer tied to LAM run // no needed since variance based load balancing is no longer tied to LAM run
// @Test @Test
void performAssignment_varianceBalanceFreq3_asserLoadBalancingEvery3Iteration() throws Exception { void performAssignment_varianceBalanceFreq3_asserLoadBalancingEvery3Iteration() throws Exception {
final LeaseManagementConfig.WorkerUtilizationAwareAssignmentConfig config = final LeaseManagementConfig.WorkerUtilizationAwareAssignmentConfig config =
getWorkerUtilizationAwareAssignmentConfig(Double.MAX_VALUE, 10); getWorkerUtilizationAwareAssignmentConfig(Double.MAX_VALUE, 10);
@ -476,18 +476,9 @@ class LeaseAssignmentManagerTest {
leaseRefresher.listLeases().stream() leaseRefresher.listLeases().stream()
.filter(lease -> lease.leaseOwner().equals(TEST_TAKE_WORKER_ID)) .filter(lease -> lease.leaseOwner().equals(TEST_TAKE_WORKER_ID))
.count()); .count());
setupConditionForVarianceBalancing();
// 5th Run, expect no re-balance
leaseAssignmentManagerRunnable.run();
assertEquals(
1L,
leaseRefresher.listLeases().stream()
.filter(lease -> lease.leaseOwner().equals(TEST_TAKE_WORKER_ID))
.count());
} }
@Test // @Test
void performAssignment_varianceBalanceFreq3_asserLoadBalancingEveryVarianceBalancingFrequencyLeaseDuration() void performAssignment_varianceBalanceFreq3_asserLoadBalancingEveryVarianceBalancingFrequencyLeaseDuration()
throws Exception { throws Exception {