From aefb5c2522fcff1b2f0db57169e0d0c9f5b5a32f Mon Sep 17 00:00:00 2001
From: javarno <45781895+javarno@users.noreply.github.com>
Date: Tue, 2 Apr 2019 23:55:59 +0200
Subject: [PATCH] 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
---
 .../kinesis/KclMessageDrivenChannelAdapter.java    | 14 +++++++++++++-
 1 file changed, 13 insertions(+), 1 deletion(-)

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 9fa7f4c..d1fe904 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
@@ -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,