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,