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

Code for backward compatibility check

Browse source
This commit is contained in:
eha sah 2025-04-11 14:34:30 -07:00
parent d01ed3e527
commit af99b4d743
3 changed files with 17 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -80,6 +80,12 @@ public class LeaseAssignmentManager {
private static final String FORCE_LEADER_RELEASE_METRIC_NAME = "ForceLeaderRelease"; private static final String FORCE_LEADER_RELEASE_METRIC_NAME = "ForceLeaderRelease";
/**
* Default multiplier for LAM frequency with respect to leaseDurationMillis (lease failover millis).
* If leaseDurationMillis is 10000 millis, default LAM frequency is 20000 millis.
*/
private static final int DEFAULT_LEASE_ASSIGNMENT_MANAGER_FREQ_MULTIPLIER = 2;
/** /**
* Default retry attempt for loading leases and workers before giving up. * Default retry attempt for loading leases and workers before giving up.
*/ */
@ -140,7 +146,7 @@ public class LeaseAssignmentManager {
this.nanoTimeProvider = nanoTimeProvider; this.nanoTimeProvider = nanoTimeProvider;
this.maxLeasesForWorker = maxLeasesForWorker; this.maxLeasesForWorker = maxLeasesForWorker;
this.gracefulLeaseHandoffConfig = gracefulLeaseHandoffConfig; this.gracefulLeaseHandoffConfig = gracefulLeaseHandoffConfig;
this.leaseAssignmentIntervalMillis = 2 * leaseDurationMillis; this.leaseAssignmentIntervalMillis = leaseDurationMillis * DEFAULT_LEASE_ASSIGNMENT_MANAGER_FREQ_MULTIPLIER;
} }
public LeaseAssignmentManager( public LeaseAssignmentManager(

1
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/dynamodb/DynamoDBLeaseManagementFactory.java
View file

@ -146,6 +146,7 @@ public class DynamoDBLeaseManagementFactory implements LeaseManagementFactory {
* @param workerUtilizationAwareAssignmentConfig * @param workerUtilizationAwareAssignmentConfig
* @param gracefulLeaseHandoffConfig * @param gracefulLeaseHandoffConfig
*/ */
@Deprecated
public DynamoDBLeaseManagementFactory( public DynamoDBLeaseManagementFactory(
final @NotNull KinesisAsyncClient kinesisClient, final @NotNull KinesisAsyncClient kinesisClient,
final @NotNull DynamoDbAsyncClient dynamoDBClient, final @NotNull DynamoDbAsyncClient dynamoDBClient,

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

@ -433,7 +433,6 @@ class LeaseAssignmentManagerTest {
.anyMatch(lease -> lease.leaseOwner().equals(TEST_YIELD_WORKER_ID + "2"))); .anyMatch(lease -> lease.leaseOwner().equals(TEST_YIELD_WORKER_ID + "2")));
} }
// 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 =
@ -476,6 +475,15 @@ 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());
} }
private void setupConditionForVarianceBalancing() throws Exception { private void setupConditionForVarianceBalancing() throws Exception {