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

GH-151: Recreate KCL Worker on each restart

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

**Cherry-pick to 2.2.x**
This commit is contained in:
Artem Bilan
2019-09-10 13:36:24 -04:00
parent 12ce221513
commit 377e2a42f1
1 changed files with 48 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -83,6 +83,8 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport {
private static final ThreadLocal<AttributeAccessor> 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<byte[]> 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);