From c99698dfb6e0eb4da5883aa06238df9397895dff Mon Sep 17 00:00:00 2001 From: Refal <4848778+Refal@users.noreply.github.com> Date: Tue, 1 Dec 2020 16:05:42 +0100 Subject: [PATCH] GH-184: Correct default values in KclMDChAdapter 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 --- .../KclMessageDrivenChannelAdapter.java | 24 ++++++++++++++++--- 1 file changed, 21 insertions(+), 3 deletions(-) diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java index e407570..fb243ff 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java @@ -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 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();