GH-142: Configure workerId in KCL adapter
* Allow to modify the worker identifier used in KCL adapter. Fixes https://github.com/spring-projects/spring-integration-aws/issues/142
This commit is contained in:
@@ -103,6 +103,8 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport {
|
||||
|
||||
private CheckpointMode checkpointMode = CheckpointMode.batch;
|
||||
|
||||
private String workerId = UUID.randomUUID().toString();
|
||||
|
||||
public KclMessageDrivenChannelAdapter(String streams) {
|
||||
this(streams, AmazonKinesisClientBuilder.defaultClient(),
|
||||
AmazonCloudWatchClientBuilder.defaultClient(), AmazonDynamoDBClientBuilder.defaultClient(),
|
||||
@@ -176,6 +178,16 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport {
|
||||
this.checkpointMode = checkpointMode;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the worker identifier used to distinguish different
|
||||
* workers/processes of a Kinesis application.
|
||||
* @param workerId the worker identifier to use
|
||||
*/
|
||||
public void setWorkerId(String workerId) {
|
||||
Assert.hasText(workerId, "'workerId' must not be null or empty");
|
||||
this.workerId = workerId;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void onInit() {
|
||||
super.onInit();
|
||||
@@ -189,7 +201,7 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport {
|
||||
this.kinesisProxyCredentialsProvider,
|
||||
null, null,
|
||||
KinesisClientLibConfiguration.DEFAULT_FAILOVER_TIME_MILLIS,
|
||||
UUID.randomUUID().toString(),
|
||||
this.workerId,
|
||||
KinesisClientLibConfiguration.DEFAULT_MAX_RECORDS,
|
||||
this.idleBetweenPolls,
|
||||
false,
|
||||
|
||||
Reference in New Issue
Block a user