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

Minor comment updates. Update test. Rename config variable.

Browse source
This commit is contained in:
Lucien Luc 2024-04-19 15:25:03 -07:00
parent c7957d9a86
commit 99794ab2ab
8 changed files with 42 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
amazon-kinesis-client-multilang/src/main/java/software/amazon/kinesis/coordinator/KinesisClientLibConfiguration.java
View file

@ -217,7 +217,7 @@ public class KinesisClientLibConfiguration {
private AwsCredentialsProvider dynamoDBCredentialsProvider; private AwsCredentialsProvider dynamoDBCredentialsProvider;
private AwsCredentialsProvider cloudWatchCredentialsProvider; private AwsCredentialsProvider cloudWatchCredentialsProvider;
private long failoverTimeMillis; private long failoverTimeMillis;
private boolean doPriorityLeaseTaking; private boolean enablePriorityLeaseAssignment;
private String workerIdentifier; private String workerIdentifier;
private long shardSyncIntervalMillis; private long shardSyncIntervalMillis;
private int maxRecords; private int maxRecords;
@ -960,8 +960,8 @@ public class KinesisClientLibConfiguration {
return this; return this;
} }
public KinesisClientLibConfiguration withDoPriorityLeaseTaking(boolean doPriorityLeaseTaking) { public KinesisClientLibConfiguration withEnablePriorityLeaseAssignment(boolean enablePriorityLeaseAssignment) {
this.doPriorityLeaseTaking = doPriorityLeaseTaking; this.enablePriorityLeaseAssignment = enablePriorityLeaseAssignment;
return this; return this;
} }

2
amazon-kinesis-client-multilang/src/main/java/software/amazon/kinesis/multilang/config/MultiLangDaemonConfiguration.java
View file

@ -87,7 +87,7 @@ public class MultiLangDaemonConfiguration {
@ConfigurationSettable(configurationClass = LeaseManagementConfig.class) @ConfigurationSettable(configurationClass = LeaseManagementConfig.class)
private long failoverTimeMillis; private long failoverTimeMillis;
@ConfigurationSettable(configurationClass = LeaseManagementConfig.class) @ConfigurationSettable(configurationClass = LeaseManagementConfig.class)
private Boolean doPriorityLeaseTaking; private Boolean enablePriorityLeaseAssignment;
@ConfigurationSettable(configurationClass = LeaseManagementConfig.class) @ConfigurationSettable(configurationClass = LeaseManagementConfig.class)
private long shardSyncIntervalMillis; private long shardSyncIntervalMillis;
@ConfigurationSettable(configurationClass = LeaseManagementConfig.class) @ConfigurationSettable(configurationClass = LeaseManagementConfig.class)

6
amazon-kinesis-client-multilang/src/test/java/software/amazon/kinesis/multilang/config/MultiLangDaemonConfigurationTest.java
View file

@ -91,14 +91,14 @@ public class MultiLangDaemonConfigurationTest {
} }
@Test @Test
public void testSetDoPriorityLeaseTaking() { public void testSetEnablePriorityLeaseAssignment() {
MultiLangDaemonConfiguration configuration = baseConfiguration(); MultiLangDaemonConfiguration configuration = baseConfiguration();
configuration.setDoPriorityLeaseTaking(Boolean.FALSE); configuration.setEnablePriorityLeaseAssignment(false);
MultiLangDaemonConfiguration.ResolvedConfiguration resolvedConfiguration = configuration.resolvedConfiguration( MultiLangDaemonConfiguration.ResolvedConfiguration resolvedConfiguration = configuration.resolvedConfiguration(
shardRecordProcessorFactory); shardRecordProcessorFactory);
assertThat(resolvedConfiguration.leaseManagementConfig.doPriorityLeaseTaking(), equalTo(false)); assertThat(resolvedConfiguration.leaseManagementConfig.enablePriorityLeaseAssignment(), equalTo(false));
} }
@Test @Test

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

@ -57,7 +57,7 @@ public class LeaseManagementConfig {
public static final long DEFAULT_COMPLETED_LEASE_CLEANUP_INTERVAL_MILLIS = Duration.ofMinutes(5).toMillis(); public static final long DEFAULT_COMPLETED_LEASE_CLEANUP_INTERVAL_MILLIS = Duration.ofMinutes(5).toMillis();
public static final long DEFAULT_GARBAGE_LEASE_CLEANUP_INTERVAL_MILLIS = Duration.ofMinutes(30).toMillis(); public static final long DEFAULT_GARBAGE_LEASE_CLEANUP_INTERVAL_MILLIS = Duration.ofMinutes(30).toMillis();
public static final long DEFAULT_PERIODIC_SHARD_SYNC_INTERVAL_MILLIS = 2 * 60 * 1000L; public static final long DEFAULT_PERIODIC_SHARD_SYNC_INTERVAL_MILLIS = 2 * 60 * 1000L;
public static final boolean DEFAULT_DO_PRIORITY_LEASE_TAKING = true; public static final boolean DEFAULT_ENABLE_PRIORITY_LEASE_ASSIGNMENT = true;
public static final int DEFAULT_CONSECUTIVE_HOLES_FOR_TRIGGERING_LEASE_RECOVERY = 3; public static final int DEFAULT_CONSECUTIVE_HOLES_FOR_TRIGGERING_LEASE_RECOVERY = 3;
@ -107,11 +107,12 @@ public class LeaseManagementConfig {
* Whether workers should take very expired leases at priority. A very expired lease is when a worker does not * Whether workers should take very expired leases at priority. A very expired lease is when a worker does not
* renew its lease in 3 * {@link LeaseManagementConfig#failoverTimeMillis}. Very expired leases will be taken at * renew its lease in 3 * {@link LeaseManagementConfig#failoverTimeMillis}. Very expired leases will be taken at
* priority for a worker which disregards the target leases for the worker but obeys * priority for a worker which disregards the target leases for the worker but obeys
* {@link LeaseManagementConfig#maxLeasesForWorker} * {@link LeaseManagementConfig#maxLeasesForWorker}. New leases for new shards due to shard mutation are
* considered to be very expired and taken with priority.
* *
* <p>Default value: true </p> * <p>Default value: true </p>
*/ */
private boolean doPriorityLeaseTaking = DEFAULT_DO_PRIORITY_LEASE_TAKING; private boolean enablePriorityLeaseAssignment = DEFAULT_ENABLE_PRIORITY_LEASE_ASSIGNMENT;
/** /**
* Shard sync interval in milliseconds - e.g. wait for this long between shard sync tasks. * Shard sync interval in milliseconds - e.g. wait for this long between shard sync tasks.
@ -381,7 +382,7 @@ public class LeaseManagementConfig {
workerIdentifier(), workerIdentifier(),
executorService(), executorService(),
failoverTimeMillis(), failoverTimeMillis(),
doPriorityLeaseTaking(), enablePriorityLeaseAssignment(),
epsilonMillis(), epsilonMillis(),
maxLeasesForWorker(), maxLeasesForWorker(),
maxLeasesToStealAtOneTime(), maxLeasesToStealAtOneTime(),

10
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/dynamodb/DynamoDBLeaseCoordinator.java
View file

@ -153,7 +153,7 @@ public class DynamoDBLeaseCoordinator implements LeaseCoordinator {
final long initialLeaseTableWriteCapacity, final long initialLeaseTableWriteCapacity,
final MetricsFactory metricsFactory) { final MetricsFactory metricsFactory) {
this(leaseRefresher, workerIdentifier, leaseDurationMillis, this(leaseRefresher, workerIdentifier, leaseDurationMillis,
LeaseManagementConfig.DEFAULT_DO_PRIORITY_LEASE_TAKING, epsilonMillis, maxLeasesForWorker, LeaseManagementConfig.DEFAULT_ENABLE_PRIORITY_LEASE_ASSIGNMENT, epsilonMillis, maxLeasesForWorker,
maxLeasesToStealAtOneTime, maxLeaseRenewerThreadCount, maxLeasesToStealAtOneTime, maxLeaseRenewerThreadCount,
TableConstants.DEFAULT_INITIAL_LEASE_TABLE_READ_CAPACITY, TableConstants.DEFAULT_INITIAL_LEASE_TABLE_READ_CAPACITY,
TableConstants.DEFAULT_INITIAL_LEASE_TABLE_WRITE_CAPACITY, metricsFactory); TableConstants.DEFAULT_INITIAL_LEASE_TABLE_WRITE_CAPACITY, metricsFactory);
@ -168,8 +168,8 @@ public class DynamoDBLeaseCoordinator implements LeaseCoordinator {
* Identifies the worker (e.g. useful to track lease ownership) * Identifies the worker (e.g. useful to track lease ownership)
* @param leaseDurationMillis * @param leaseDurationMillis
* Duration of a lease * Duration of a lease
* @param doPriorityLeaseTaking * @param enablePriorityLeaseAssignment
* Whether to do priority lease taking for very expired leases * Whether to do priority lease assignment for very expired leases
* @param epsilonMillis * @param epsilonMillis
* Allow for some variance when calculating lease expirations * Allow for some variance when calculating lease expirations
* @param maxLeasesForWorker * @param maxLeasesForWorker
@ -186,7 +186,7 @@ public class DynamoDBLeaseCoordinator implements LeaseCoordinator {
public DynamoDBLeaseCoordinator(final LeaseRefresher leaseRefresher, public DynamoDBLeaseCoordinator(final LeaseRefresher leaseRefresher,
final String workerIdentifier, final String workerIdentifier,
final long leaseDurationMillis, final long leaseDurationMillis,
final boolean doPriorityLeaseTaking, final boolean enablePriorityLeaseAssignment,
final long epsilonMillis, final long epsilonMillis,
final int maxLeasesForWorker, final int maxLeasesForWorker,
final int maxLeasesToStealAtOneTime, final int maxLeasesToStealAtOneTime,
@ -199,7 +199,7 @@ public class DynamoDBLeaseCoordinator implements LeaseCoordinator {
this.leaseTaker = new DynamoDBLeaseTaker(leaseRefresher, workerIdentifier, leaseDurationMillis, metricsFactory) this.leaseTaker = new DynamoDBLeaseTaker(leaseRefresher, workerIdentifier, leaseDurationMillis, metricsFactory)
.withMaxLeasesForWorker(maxLeasesForWorker) .withMaxLeasesForWorker(maxLeasesForWorker)
.withMaxLeasesToStealAtOneTime(maxLeasesToStealAtOneTime) .withMaxLeasesToStealAtOneTime(maxLeasesToStealAtOneTime)
.withDoPriorityLeaseTaking(doPriorityLeaseTaking); .withEnablePriorityLeaseAssignment(enablePriorityLeaseAssignment);
this.leaseRenewer = new DynamoDBLeaseRenewer( this.leaseRenewer = new DynamoDBLeaseRenewer(
leaseRefresher, workerIdentifier, leaseDurationMillis, leaseRenewalThreadpool, metricsFactory); leaseRefresher, workerIdentifier, leaseDurationMillis, leaseRenewalThreadpool, metricsFactory);
this.renewerIntervalMillis = getRenewerTakerIntervalMillis(leaseDurationMillis, epsilonMillis); this.renewerIntervalMillis = getRenewerTakerIntervalMillis(leaseDurationMillis, epsilonMillis);

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

@ -71,7 +71,7 @@ public class DynamoDBLeaseManagementFactory implements LeaseManagementFactory {
private Function<StreamConfig, ShardDetector> customShardDetectorProvider; private Function<StreamConfig, ShardDetector> customShardDetectorProvider;
private final long failoverTimeMillis; private final long failoverTimeMillis;
private final boolean doPriorityLeaseTaking; private final boolean enablePriorityLeaseAssignment;
private final long epsilonMillis; private final long epsilonMillis;
private final int maxLeasesForWorker; private final int maxLeasesForWorker;
private final int maxLeasesToStealAtOneTime; private final int maxLeasesToStealAtOneTime;
@ -563,7 +563,7 @@ public class DynamoDBLeaseManagementFactory implements LeaseManagementFactory {
LeaseCleanupConfig leaseCleanupConfig) { LeaseCleanupConfig leaseCleanupConfig) {
this(kinesisClient, dynamoDBClient, tableName, this(kinesisClient, dynamoDBClient, tableName,
workerIdentifier, executorService, failoverTimeMillis, workerIdentifier, executorService, failoverTimeMillis,
LeaseManagementConfig.DEFAULT_DO_PRIORITY_LEASE_TAKING, epsilonMillis, maxLeasesForWorker, LeaseManagementConfig.DEFAULT_ENABLE_PRIORITY_LEASE_ASSIGNMENT, epsilonMillis, maxLeasesForWorker,
maxLeasesToStealAtOneTime, maxLeaseRenewalThreads, cleanupLeasesUponShardCompletion, maxLeasesToStealAtOneTime, maxLeaseRenewalThreads, cleanupLeasesUponShardCompletion,
ignoreUnexpectedChildShards, shardSyncIntervalMillis, consistentReads, listShardsBackoffTimeMillis, ignoreUnexpectedChildShards, shardSyncIntervalMillis, consistentReads, listShardsBackoffTimeMillis,
maxListShardsRetryAttempts, maxCacheMissesBeforeReload, listShardsCacheAllowedAgeInSeconds, maxListShardsRetryAttempts, maxCacheMissesBeforeReload, listShardsCacheAllowedAgeInSeconds,
@ -581,7 +581,7 @@ public class DynamoDBLeaseManagementFactory implements LeaseManagementFactory {
* @param workerIdentifier * @param workerIdentifier
* @param executorService * @param executorService
* @param failoverTimeMillis * @param failoverTimeMillis
* @param doPriorityLeaseTaking * @param enablePriorityLeaseAssignment
* @param epsilonMillis * @param epsilonMillis
* @param maxLeasesForWorker * @param maxLeasesForWorker
* @param maxLeasesToStealAtOneTime * @param maxLeasesToStealAtOneTime
@ -610,7 +610,7 @@ public class DynamoDBLeaseManagementFactory implements LeaseManagementFactory {
public DynamoDBLeaseManagementFactory(final KinesisAsyncClient kinesisClient, public DynamoDBLeaseManagementFactory(final KinesisAsyncClient kinesisClient,
final DynamoDbAsyncClient dynamoDBClient, final String tableName, final String workerIdentifier, final DynamoDbAsyncClient dynamoDBClient, final String tableName, final String workerIdentifier,
final ExecutorService executorService, final long failoverTimeMillis, final ExecutorService executorService, final long failoverTimeMillis,
final boolean doPriorityLeaseTaking, final long epsilonMillis, final boolean enablePriorityLeaseAssignment, final long epsilonMillis,
final int maxLeasesForWorker, final int maxLeasesToStealAtOneTime, final int maxLeaseRenewalThreads, final int maxLeasesForWorker, final int maxLeasesToStealAtOneTime, final int maxLeaseRenewalThreads,
final boolean cleanupLeasesUponShardCompletion, final boolean ignoreUnexpectedChildShards, final boolean cleanupLeasesUponShardCompletion, final boolean ignoreUnexpectedChildShards,
final long shardSyncIntervalMillis, final boolean consistentReads, final long listShardsBackoffTimeMillis, final long shardSyncIntervalMillis, final boolean consistentReads, final long listShardsBackoffTimeMillis,
@ -628,7 +628,7 @@ public class DynamoDBLeaseManagementFactory implements LeaseManagementFactory {
this.workerIdentifier = workerIdentifier; this.workerIdentifier = workerIdentifier;
this.executorService = executorService; this.executorService = executorService;
this.failoverTimeMillis = failoverTimeMillis; this.failoverTimeMillis = failoverTimeMillis;
this.doPriorityLeaseTaking = doPriorityLeaseTaking; this.enablePriorityLeaseAssignment = enablePriorityLeaseAssignment;
this.epsilonMillis = epsilonMillis; this.epsilonMillis = epsilonMillis;
this.maxLeasesForWorker = maxLeasesForWorker; this.maxLeasesForWorker = maxLeasesForWorker;
this.maxLeasesToStealAtOneTime = maxLeasesToStealAtOneTime; this.maxLeasesToStealAtOneTime = maxLeasesToStealAtOneTime;
@ -660,8 +660,7 @@ public class DynamoDBLeaseManagementFactory implements LeaseManagementFactory {
public LeaseCoordinator createLeaseCoordinator(@NonNull final MetricsFactory metricsFactory) { public LeaseCoordinator createLeaseCoordinator(@NonNull final MetricsFactory metricsFactory) {
return new DynamoDBLeaseCoordinator(this.createLeaseRefresher(), return new DynamoDBLeaseCoordinator(this.createLeaseRefresher(),
workerIdentifier, workerIdentifier,
failoverTimeMillis, failoverTimeMillis, enablePriorityLeaseAssignment,
doPriorityLeaseTaking,
epsilonMillis, epsilonMillis,
maxLeasesForWorker, maxLeasesForWorker,
maxLeasesToStealAtOneTime, maxLeasesToStealAtOneTime,

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

@ -69,7 +69,7 @@ public class DynamoDBLeaseTaker implements LeaseTaker {
// TODO: Remove these defaults and use the defaults in the config // TODO: Remove these defaults and use the defaults in the config
private int maxLeasesForWorker = Integer.MAX_VALUE; private int maxLeasesForWorker = Integer.MAX_VALUE;
private int maxLeasesToStealAtOneTime = 1; private int maxLeasesToStealAtOneTime = 1;
private boolean doPriorityLeaseTaking = true; private boolean enablePriorityLeaseAssignment = true;
private int veryOldLeaseDurationNanosMultiplier = 3; private int veryOldLeaseDurationNanosMultiplier = 3;
private long lastScanTimeNanos = 0L; private long lastScanTimeNanos = 0L;
@ -124,8 +124,8 @@ public class DynamoDBLeaseTaker implements LeaseTaker {
return this; return this;
} }
public DynamoDBLeaseTaker withDoPriorityLeaseTaking(boolean doPriorityLeaseTaking) { public DynamoDBLeaseTaker withEnablePriorityLeaseAssignment(boolean enablePriorityLeaseAssignment) {
this.doPriorityLeaseTaking = doPriorityLeaseTaking; this.enablePriorityLeaseAssignment = enablePriorityLeaseAssignment;
return this; return this;
} }
@ -446,7 +446,7 @@ public class DynamoDBLeaseTaker implements LeaseTaker {
// If there are leases that have been expired for an extended period of // If there are leases that have been expired for an extended period of
// time, take them with priority, disregarding the target (computed // time, take them with priority, disregarding the target (computed
// later) but obeying the maximum limit per worker. // later) but obeying the maximum limit per worker.
if (doPriorityLeaseTaking) { if (enablePriorityLeaseAssignment) {
long currentNanoTime; long currentNanoTime;
try { try {
currentNanoTime = timeProvider.call(); currentNanoTime = timeProvider.call();

26
amazon-kinesis-client/src/test/java/software/amazon/kinesis/leases/dynamodb/DynamoDBLeaseTakerTest.java
View file

@ -153,28 +153,32 @@ public class DynamoDBLeaseTakerTest {
} }
@Test @Test
public void test_disableDoPriorityLeaseTakingGetsCorrectLeases() throws Exception { public void test_disableEnablePriorityLeaseAssignmentGetsCorrectLeases() throws Exception {
long veryOldThreshold = MOCK_CURRENT_TIME - long veryOldThreshold = MOCK_CURRENT_TIME -
(TimeUnit.MILLISECONDS.toNanos(LEASE_DURATION_MILLIS) * DEFAULT_VERY_OLD_LEASE_DURATION_MULTIPLIER); (TimeUnit.MILLISECONDS.toNanos(LEASE_DURATION_MILLIS) * DEFAULT_VERY_OLD_LEASE_DURATION_MULTIPLIER);
DynamoDBLeaseTaker dynamoDBLeaseTakerWithCustomMultiplier = DynamoDBLeaseTaker dynamoDBLeaseTakerWithDisabledPriorityLeaseAssignment =
new DynamoDBLeaseTaker(leaseRefresher, WORKER_IDENTIFIER, LEASE_DURATION_MILLIS, metricsFactory) new DynamoDBLeaseTaker(leaseRefresher, WORKER_IDENTIFIER, LEASE_DURATION_MILLIS, metricsFactory)
.withDoPriorityLeaseTaking(false); .withEnablePriorityLeaseAssignment(false);
final List<Lease> allLeases = new ArrayList<>(); final List<Lease> allLeases = new ArrayList<>();
allLeases.add(createLease("foo", "2", MOCK_CURRENT_TIME)); allLeases.add(createLease("bar", "2", MOCK_CURRENT_TIME));
allLeases.add(createLease("bar", "3", veryOldThreshold - 1)); allLeases.add(createLease("bar", "3", MOCK_CURRENT_TIME));
allLeases.add(createLease("baz", "4", veryOldThreshold + 1)); allLeases.add(createLease("bar", "4", MOCK_CURRENT_TIME));
final List<Lease> expiredLeases = allLeases.subList(1, 3); allLeases.add(createLease("baz", "5", veryOldThreshold - 1));
allLeases.add(createLease("baz", "6", veryOldThreshold + 1));
allLeases.add(createLease(null, "7"));
final List<Lease> expiredLeases = allLeases.subList(3, 6);
dynamoDBLeaseTakerWithCustomMultiplier.allLeases.putAll( dynamoDBLeaseTakerWithDisabledPriorityLeaseAssignment.allLeases.putAll(
allLeases.stream().collect(Collectors.toMap(Lease::leaseKey, Function.identity()))); allLeases.stream().collect(Collectors.toMap(Lease::leaseKey, Function.identity())));
when(leaseRefresher.listLeases()).thenReturn(allLeases); when(leaseRefresher.listLeases()).thenReturn(allLeases);
when(metricsFactory.createMetrics()).thenReturn(new NullMetricsScope()); when(metricsFactory.createMetrics()).thenReturn(new NullMetricsScope());
when(timeProvider.call()).thenReturn(MOCK_CURRENT_TIME); when(timeProvider.call()).thenReturn(MOCK_CURRENT_TIME);
Set<Lease> output = dynamoDBLeaseTakerWithCustomMultiplier.computeLeasesToTake(expiredLeases, timeProvider); Set<Lease> output = dynamoDBLeaseTakerWithDisabledPriorityLeaseAssignment.computeLeasesToTake(expiredLeases, timeProvider);
final Set<Lease> expectedOutput = new HashSet<>(); final Set<Lease> expectedOutput = new HashSet<>();
expectedOutput.add(createLease("bar", "3", veryOldThreshold - 1)); expectedOutput.add(createLease("baz", "5", veryOldThreshold - 1));
expectedOutput.add(createLease("baz", "4", veryOldThreshold + 1)); expectedOutput.add(createLease("baz", "6", veryOldThreshold + 1));
expectedOutput.add(createLease(null, "7"));
assertEquals(expectedOutput, output); assertEquals(expectedOutput, output);
} }