Spring/spring-integration-aws
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-184: Correct default values in KclMDChAdapter

Browse Source 
Fixes https://github.com/spring-projects/spring-integration-aws/issues/184

During creating `KclMessageDrivenChannelAdapter`
if `config == null` a default `KinesisClientLibConfiguration` is created 
with parameter `leaseCleanupIntervalMillis == 0` which causes an `IllegalArgumentExcpetoin` 
when KPL is initialized

* Use default reasonable values for lease clean up intervals.
* Copy respective constants from the `KinesisClientLibConfiguration` into the `KclMessageDrivenChannelAdapter`
since they are `private` over there
This commit is contained in:
Refal
2020-12-01 16:05:42 +01:00
committed by GitHub
parent a0d0752cdc
commit c99698dfb6
1 changed files with 21 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.aws.inbound.kinesis;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -84,6 +85,23 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport {
private static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<>();
/**
* Interval to run lease cleanup thread in {@link LeaseCleanupManager}.
*/
private static final long DEFAULT_LEASE_CLEANUP_INTERVAL_MILLIS = Duration.ofMinutes(1).toMillis();
/**
* Threshold in millis at which to check if there are any completed leases (leases for shards which have been
* closed as a result of a resharding operation) that need to be cleaned up.
*/
private static final long DEFAULT_COMPLETED_LEASE_CLEANUP_THRESHOLD_MILLIS = Duration.ofMinutes(5).toMillis();
/**
* Threshold in millis at which to check if there are any garbage leases (leases for shards which no longer exist
* in the stream) that need to be cleaned up.
*/
private static final long DEFAULT_GARBAGE_LEASE_CLEANUP_THRESHOLD_MILLIS = Duration.ofMinutes(30).toMillis();
private final RecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory();
private final String stream;
@@ -298,9 +316,9 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport {
KinesisClientLibConfiguration.DEFAULT_SHUTDOWN_GRACE_MILLIS,
KinesisClientLibConfiguration.DEFAULT_DDB_BILLING_MODE,
new SimpleRecordsFetcherFactory(),
0,
0,
0);
DEFAULT_LEASE_CLEANUP_INTERVAL_MILLIS,
DEFAULT_COMPLETED_LEASE_CLEANUP_THRESHOLD_MILLIS,
DEFAULT_GARBAGE_LEASE_CLEANUP_THRESHOLD_MILLIS);
}
this.consumerGroup = this.config.getApplicationName();