From 377e2a42f13fb12a4416506bd0c2ed6ebe479519 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 10 Sep 2019 13:36:24 -0400 Subject: [PATCH] GH-151: Recreate KCL Worker on each restart Fixes https://github.com/spring-projects/spring-integration-aws/issues/151 **Cherry-pick to 2.2.x** --- .../KclMessageDrivenChannelAdapter.java | 66 ++++++++++++++----- 1 file changed, 48 insertions(+), 18 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 59f1b31..46036eb 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 @@ -83,6 +83,8 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { private static final ThreadLocal attributesHolder = new ThreadLocal<>(); + private final RecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory(); + private final String stream; private final AmazonKinesis kinesisClient; @@ -99,7 +101,7 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { private InboundMessageMapper embeddedHeadersMapper; - private Worker scheduler; + private KinesisClientLibConfiguration config; private InitialPositionInStream streamInitialSequence = InitialPositionInStream.LATEST; @@ -119,6 +121,8 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { private boolean bindSourceRecord; + private volatile Worker scheduler; + public KclMessageDrivenChannelAdapter(String streams) { this(streams, AmazonKinesisClientBuilder.defaultClient(), AmazonCloudWatchClientBuilder.defaultClient(), AmazonDynamoDBClientBuilder.defaultClient(), new DefaultAWSCredentialsProviderChain()); @@ -229,22 +233,31 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { protected void onInit() { super.onInit(); - KinesisClientLibConfiguration config = new KinesisClientLibConfiguration(this.consumerGroup, this.stream, null, - this.streamInitialSequence, this.kinesisProxyCredentialsProvider, null, null, - KinesisClientLibConfiguration.DEFAULT_FAILOVER_TIME_MILLIS, this.workerId, - KinesisClientLibConfiguration.DEFAULT_MAX_RECORDS, this.idleBetweenPolls, false, - KinesisClientLibConfiguration.DEFAULT_PARENT_SHARD_POLL_INTERVAL_MILLIS, - KinesisClientLibConfiguration.DEFAULT_SHARD_SYNC_INTERVAL_MILLIS, - KinesisClientLibConfiguration.DEFAULT_CLEANUP_LEASES_UPON_SHARDS_COMPLETION, new ClientConfiguration(), - new ClientConfiguration(), new ClientConfiguration(), this.consumerBackoff, - KinesisClientLibConfiguration.DEFAULT_METRICS_BUFFER_TIME_MILLIS, - KinesisClientLibConfiguration.DEFAULT_METRICS_MAX_QUEUE_SIZE, - KinesisClientLibConfiguration.DEFAULT_VALIDATE_SEQUENCE_NUMBER_BEFORE_CHECKPOINTING, null, - KinesisClientLibConfiguration.DEFAULT_SHUTDOWN_GRACE_MILLIS); - - this.scheduler = new Worker.Builder().kinesisClient(this.kinesisClient).dynamoDBClient(this.dynamoDBClient) - .cloudWatchClient(this.cloudWatchClient).recordProcessorFactory(new RecordProcessorFactory()) - .execService(new ExecutorServiceAdapter(this.executor)).config(config).build(); + this.config = + new KinesisClientLibConfiguration(this.consumerGroup, + this.stream, + null, + this.streamInitialSequence, + this.kinesisProxyCredentialsProvider, + null, + null, + KinesisClientLibConfiguration.DEFAULT_FAILOVER_TIME_MILLIS, + this.workerId, + KinesisClientLibConfiguration.DEFAULT_MAX_RECORDS, + this.idleBetweenPolls, + false, + KinesisClientLibConfiguration.DEFAULT_PARENT_SHARD_POLL_INTERVAL_MILLIS, + KinesisClientLibConfiguration.DEFAULT_SHARD_SYNC_INTERVAL_MILLIS, + KinesisClientLibConfiguration.DEFAULT_CLEANUP_LEASES_UPON_SHARDS_COMPLETION, + new ClientConfiguration(), + new ClientConfiguration(), + new ClientConfiguration(), + this.consumerBackoff, + KinesisClientLibConfiguration.DEFAULT_METRICS_BUFFER_TIME_MILLIS, + KinesisClientLibConfiguration.DEFAULT_METRICS_MAX_QUEUE_SIZE, + KinesisClientLibConfiguration.DEFAULT_VALIDATE_SEQUENCE_NUMBER_BEFORE_CHECKPOINTING, + null, + KinesisClientLibConfiguration.DEFAULT_SHUTDOWN_GRACE_MILLIS); } @Override @@ -257,6 +270,17 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { + "because it does not make sense in case of [ListenerMode.batch]."); } + this.scheduler = + new Worker + .Builder() + .kinesisClient(this.kinesisClient) + .dynamoDBClient(this.dynamoDBClient) + .cloudWatchClient(this.cloudWatchClient) + .recordProcessorFactory(this.recordProcessorFactory) + .execService(new ExecutorServiceAdapter(this.executor)) + .config(this.config) + .build(); + this.executor.execute(this.scheduler); } @@ -271,6 +295,12 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { } + @Override + public void destroy() { + super.destroy(); + this.scheduler.shutdown(); + } + @Override protected AttributeAccessor getErrorMessageAttributes(org.springframework.messaging.Message message) { AttributeAccessor attributes = attributesHolder.get(); @@ -414,7 +444,7 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { } private void performSend(AbstractIntegrationMessageBuilder messageBuilder, Object rawRecord, - IRecordProcessorCheckpointer checkpointer) { + IRecordProcessorCheckpointer checkpointer) { messageBuilder.setHeader(AwsHeaders.RECEIVED_STREAM, KclMessageDrivenChannelAdapter.this.stream) .setHeader(AwsHeaders.SHARD, this.shardId);