diff --git a/README.md b/README.md index fc11f72..d3fdda6 100644 --- a/README.md +++ b/README.md @@ -551,6 +551,13 @@ When `InboundMessageMapper` is used together with the `ListenerMode.batch`, each In this case `AwsHeaders.RECEIVED_PARTITION_KEY` and `AwsHeaders.RECEIVED_SEQUENCE_NUMBER` headers are populated to the particular message for a record. These messages are wrapped as a list payload to one outbound message. +Starting with _version 2.0_, the `KinesisMessageDrivenChannelAdapter` can be configured with the `LockRegistry` for leader selection for the shards in the provided streams. +The container iterates over the shards in its streams and tries to acquire a distributed lock for the shard in its consumer group. +If `LockRegistry` is not provided, no exclusive locking happens. +Also this locking mechanism is not applied when `KinesisShardOffset`-based configuration is provided. +In this case the global [Leader Election][] can be applied. +See also `DynamoDbLockRegistry` for more information. + ### Outbound Channel Adapter The `KinesisMessageHandler` is an `AbstractMessageHandler` to perform put record to the Kinesis stream. @@ -641,4 +648,5 @@ The `com.amazonaws:dynamodb-lock-client` dependency must be present to make a `D [Kinesalite]: https://github.com/mhart/kinesalite [Amazon SQS Message Attributes]: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-message-attributes.html [Amazon SNS Message Attributes]: https://docs.aws.amazon.com/sns/latest/dg/SNSMessageAttributes.html +[Leader Election]: https://docs.spring.io/spring-integration/docs/current/reference/html/messaging-endpoints-chapter.html#leadership-event-handling [LockRegistryLeaderInitiator]: https://docs.spring.io/spring-integration/docs/current/reference/html/messaging-endpoints-chapter.html#leadership-event-handling diff --git a/build.gradle b/build.gradle index 0b8cc4c..4efc614 100644 --- a/build.gradle +++ b/build.gradle @@ -36,7 +36,7 @@ ext { jacksonVersion = '2.9.5' servletApiVersion = '3.1.0' log4jVersion = '2.11.0' - springCloudAwsVersion = '2.0.0.BUILD-SNAPSHOT' + springCloudAwsVersion = '2.0.0.RELEASE' springIntegrationVersion = '5.0.6.RELEASE' idPrefix = 'aws' @@ -107,7 +107,7 @@ dependencies { compile('org.springframework.integration:spring-integration-file', optional) compile('org.springframework.integration:spring-integration-http', optional) -// compile('com.amazonaws:amazon-kinesis-client:1.9.0', optional) +// compile('com.amazonaws:amazon-kinesis-client:1.9.1', optional) compile('com.amazonaws:aws-java-sdk-kinesis', optional) compile('com.amazonaws:aws-java-sdk-dynamodb', optional) diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java index 1afd242..2d66a6b 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java @@ -22,6 +22,7 @@ import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.Date; +import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.List; @@ -37,6 +38,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Lock; import java.util.stream.Collectors; import org.springframework.beans.factory.DisposableBean; @@ -51,6 +53,7 @@ import org.springframework.integration.metadata.SimpleMetadataStore; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.ErrorMessageStrategy; import org.springframework.integration.support.ErrorMessageUtils; +import org.springframework.integration.support.locks.LockRegistry; import org.springframework.integration.support.management.IntegrationManagedResource; import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.jmx.export.annotation.ManagedResource; @@ -59,6 +62,7 @@ import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; import org.springframework.util.StringUtils; +import org.springframework.util.concurrent.SettableListenableFuture; import com.amazonaws.services.kinesis.AmazonKinesis; import com.amazonaws.services.kinesis.model.DescribeStreamRequest; @@ -100,6 +104,16 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i private final List consumerInvokers = new ArrayList<>(); + private final ShardLocksMonitor shardLocksMonitor = new ShardLocksMonitor(); + + private final ExecutorService shardLocksExecutor = + Executors.newSingleThreadExecutor( + new CustomizableThreadFactory( + (getComponentName() == null + ? "" + : getComponentName()) + + "-kinesis-shard-locks-")); + private String consumerGroup = "SpringIntegration"; private ConcurrentMetadataStore checkpointStore = new SimpleMetadataStore(); @@ -140,6 +154,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i private InboundMessageMapper embeddedHeadersMapper; + private LockRegistry lockRegistry; + private volatile boolean active; private volatile int consumerInvokerMaxCapacity; @@ -271,6 +287,16 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i this.embeddedHeadersMapper = embeddedHeadersMapper; } + /** + * Specify a {@link LockRegistry} for an exclusive access to provided streams. + * This is not used when shards-based configuration is provided. + * @param lockRegistry the {@link LockRegistry} to use. + * @since 2.0 + */ + public void setLockRegistry(LockRegistry lockRegistry) { + this.lockRegistry = lockRegistry; + } + @Override protected void onInit() { super.onInit(); @@ -290,6 +316,13 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i : getComponentName()) + "-kinesis-dispatcher-")); } + + if (this.streams == null) { + if (this.lockRegistry != null) { + logger.warn("The LockRegistry is ignored when explicit shards configuration is used."); + } + this.lockRegistry = null; + } } @Override @@ -398,6 +431,11 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i logger.warn("The 'checkpointMode' is overridden from [CheckpointMode.record] to [CheckpointMode.batch] " + "because it does not make sense in case of [ListenerMode.batch]."); } + + if (this.lockRegistry != null) { + this.shardLocksMonitor.start(); + } + if (this.streams != null) { populateShardsForStreams(); } @@ -509,9 +547,9 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i try { for (Shard shard : shards) { + String key = buildCheckpointKeyForShard(stream, shard.getShardId()); String endingSequenceNumber = shard.getSequenceNumberRange().getEndingSequenceNumber(); if (endingSequenceNumber != null) { - String key = buildCheckpointKeyForShard(stream, shard.getShardId()); String checkpoint = this.checkpointStore.get(key); boolean skipClosedShard = checkpoint != null && @@ -531,7 +569,9 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } } - shardsToConsume.add(shard); + if (this.lockRegistry == null || this.shardLocksMonitor.tryLock(key)) { + shardsToConsume.add(shard); + } } } catch (Exception e) { @@ -624,6 +664,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } super.doStop(); stopConsumers(); + + this.shardLocksMonitor.stop(); } private void stopConsumers() { @@ -735,6 +777,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i private final Runnable processTask = processTask(); + private final String key; + private Runnable notifier; private volatile ConsumerState state = ConsumerState.NEW; @@ -747,8 +791,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i ShardConsumer(KinesisShardOffset shardOffset) { this.shardOffset = new KinesisShardOffset(shardOffset); - String key = buildCheckpointKeyForShard(shardOffset.getStream(), shardOffset.getShard()); - this.checkpointer = new ShardCheckpointer(KinesisMessageDrivenChannelAdapter.this.checkpointStore, key); + this.key = buildCheckpointKeyForShard(shardOffset.getStream(), shardOffset.getShard()); + this.checkpointer = new ShardCheckpointer(KinesisMessageDrivenChannelAdapter.this.checkpointStore, this.key); } void setNotifier(Runnable notifier) { @@ -757,6 +801,9 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i void stop() { this.state = ConsumerState.STOP; + if (KinesisMessageDrivenChannelAdapter.this.lockRegistry != null) { + KinesisMessageDrivenChannelAdapter.this.shardLocksMonitor.unlock(this.key); + } if (this.notifier != null) { this.notifier.run(); } @@ -920,78 +967,68 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i ShardConsumer.this.state = ConsumerState.SLEEP; } - private void processRecords(List recordsToProcess) { - List records = this.checkpointer.filterRecords(recordsToProcess); - if (!records.isEmpty()) { - if (logger.isTraceEnabled()) { - logger.trace("Processing records: " + records + " for [" + ShardConsumer.this + "]"); - } + private void processRecords(List records) { + if (logger.isTraceEnabled()) { + logger.trace("Processing records: " + records + " for [" + ShardConsumer.this + "]"); + } - // TODO Reconsider this logic after rebalance and shard leader election implementation - if (CheckpointMode.batch.equals(KinesisMessageDrivenChannelAdapter.this.checkpointMode)) { - if (!this.checkpointer.checkpoint()) { - if (logger.isInfoEnabled()) { - logger.info("The records '" + recordsToProcess + "' are skipped from processing because " + - "their sequence numbers are less than already checkpointed: " + - this.checkpointer.getCheckpoint()); + this.checkpointer.setHighestSequence(records.get(records.size() - 1).getSequenceNumber()); + + switch (KinesisMessageDrivenChannelAdapter.this.listenerMode) { + case record: + for (Record record : records) { + performSend(prepareMessageForRecord(record), record); + + if (CheckpointMode.record.equals(KinesisMessageDrivenChannelAdapter.this.checkpointMode)) { + this.checkpointer.checkpoint(record.getSequenceNumber()); } - return; } - } - switch (KinesisMessageDrivenChannelAdapter.this.listenerMode) { - case record: - for (Record record : records) { - performSend(prepareMessageForRecord(record), record); + break; - if (CheckpointMode.record.equals(KinesisMessageDrivenChannelAdapter.this.checkpointMode)) { - this.checkpointer.checkpoint(record.getSequenceNumber()); - } - } + case batch: + Object payload = records; - break; + if (KinesisMessageDrivenChannelAdapter.this.embeddedHeadersMapper != null) { + payload = records.stream() + .map(this::prepareMessageForRecord) + .collect(Collectors.toList()); + } - case batch: - Object payload = records; + final List partitionKeys; + final List sequenceNumbers; + if (KinesisMessageDrivenChannelAdapter.this.converter != null) { + partitionKeys = new ArrayList<>(); + sequenceNumbers = new ArrayList<>(); - if (KinesisMessageDrivenChannelAdapter.this.embeddedHeadersMapper != null) { - payload = records.stream() - .map(this::prepareMessageForRecord) - .collect(Collectors.toList()); - } + payload = records.stream() + .map(r -> { + partitionKeys.add(r.getPartitionKey()); + sequenceNumbers.add(r.getSequenceNumber()); - final List partitionKeys; - final List sequenceNumbers; - if (KinesisMessageDrivenChannelAdapter.this.converter != null) { - partitionKeys = new ArrayList<>(); - sequenceNumbers = new ArrayList<>(); + return KinesisMessageDrivenChannelAdapter.this.converter + .convert(r.getData().array()); + }) + .collect(Collectors.toList()); + } + else { + partitionKeys = null; + sequenceNumbers = null; + } - payload = records.stream() - .map(r -> { - partitionKeys.add(r.getPartitionKey()); - sequenceNumbers.add(r.getSequenceNumber()); + AbstractIntegrationMessageBuilder messageBuilder = + getMessageBuilderFactory() + .withPayload(payload) + .setHeader(AwsHeaders.RECEIVED_PARTITION_KEY, partitionKeys) + .setHeader(AwsHeaders.RECEIVED_SEQUENCE_NUMBER, sequenceNumbers); - return KinesisMessageDrivenChannelAdapter.this.converter - .convert(r.getData().array()); - }) - .collect(Collectors.toList()); - } - else { - partitionKeys = null; - sequenceNumbers = null; - } + performSend(messageBuilder, records); - AbstractIntegrationMessageBuilder messageBuilder = - getMessageBuilderFactory() - .withPayload(payload) - .setHeader(AwsHeaders.RECEIVED_PARTITION_KEY, partitionKeys) - .setHeader(AwsHeaders.RECEIVED_SEQUENCE_NUMBER, sequenceNumbers); + break; + } - performSend(messageBuilder, records); - - break; - - } + if (CheckpointMode.batch.equals(KinesisMessageDrivenChannelAdapter.this.checkpointMode)) { + this.checkpointer.checkpoint(); } } @@ -1136,4 +1173,125 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } + private final class ShardLocksMonitor implements SchedulingAwareRunnable { + + private final Map> forLocking = + Collections.synchronizedMap(new HashMap<>()); + + private final Queue forUnlocking = new ConcurrentLinkedQueue<>(); + + private volatile boolean active = true; + + boolean tryLock(String lockKey) { + SettableListenableFuture lockedFuture = new SettableListenableFuture<>(); + Lock lock = KinesisMessageDrivenChannelAdapter.this.lockRegistry.obtain(lockKey); + this.forLocking.put(lock, lockedFuture); + try { + return lockedFuture.get(10, TimeUnit.SECONDS); + } + catch (Exception e) { + logger.error("Error during locking: " + lock, e); + return false; + } + } + + void unlock(String lockKey) { + this.forUnlocking.add(KinesisMessageDrivenChannelAdapter.this.lockRegistry.obtain(lockKey)); + } + + void start() { + this.active = true; + KinesisMessageDrivenChannelAdapter.this.shardLocksExecutor.execute(this); + } + + void stop() { + this.active = false; + } + + @Override + public void run() { + Set>> entrySet = this.forLocking.entrySet(); + + try { + while (this.active) { + synchronized (this.forLocking) { + for (Map.Entry> entry : entrySet) { + Lock lock = entry.getKey(); + SettableListenableFuture settableFuture = entry.getValue(); + if (settableFuture != null) { + try { + if (lock.tryLock()) { + settableFuture.set(true); + } + else { + settableFuture.set(false); + } + } + catch (Exception e) { + logger.error("Error during locking: " + lock, e); + } + finally { + entry.setValue(null); + } + } + } + } + + while (true) { + Lock lock = this.forUnlocking.poll(); + if (lock != null) { + try { + lock.unlock(); + } + catch (Exception e) { + logger.error("Error during unlocking: " + lock, e); + } + finally { + this.forLocking.remove(lock); + } + } + else { + break; + } + } + + try { + Thread.sleep(250); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("ShardLocksMonitor Thread [" + + this + "] has been interrupted", e); + } + } + } + finally { + synchronized (this.forLocking) { + for (Iterator>> iterator = entrySet.iterator(); + iterator.hasNext(); ) { + + Map.Entry> next = iterator.next(); + try { + next.getKey().unlock(); + } + catch (Exception e) { + logger.error("Error during unlocking: " + next.getKey(), e); + } + finally { + iterator.remove(); + } + } + } + } + + this.forUnlocking.clear(); + } + + @Override + public boolean isLongLived() { + return true; + } + + } + } diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java index b8d7a16..04a29ca 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java @@ -17,9 +17,6 @@ package org.springframework.integration.aws.inbound.kinesis; import java.math.BigInteger; -import java.util.Iterator; -import java.util.LinkedList; -import java.util.List; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -27,8 +24,6 @@ import org.apache.commons.logging.LogFactory; import org.springframework.integration.metadata.ConcurrentMetadataStore; import org.springframework.integration.metadata.MetadataStore; -import com.amazonaws.services.kinesis.model.Record; - /** * An internal {@link Checkpointer} implementation based on * provided {@link MetadataStore} and {@code key} for shard. @@ -85,30 +80,8 @@ class ShardCheckpointer implements Checkpointer { return false; } - List filterRecords(List records) { - List recordsToProcess = new LinkedList<>(records); - this.lastCheckpointValue = getCheckpoint(); - if (this.lastCheckpointValue != null) { - for (Iterator iterator = recordsToProcess.iterator(); iterator.hasNext(); ) { - Record record = iterator.next(); - String sequenceNumber = record.getSequenceNumber(); - if (new BigInteger(sequenceNumber).compareTo(new BigInteger(this.lastCheckpointValue)) <= 0) { - if (logger.isDebugEnabled()) { - logger.debug("Removing record with sequenceNumber " + sequenceNumber + - " because it is <= checkpoint(" + this.lastCheckpointValue + ")"); - } - iterator.remove(); - } - else { - this.lastCheckpointValue = sequenceNumber; - } - } - - } - else { - this.lastCheckpointValue = recordsToProcess.get(recordsToProcess.size() - 1).getSequenceNumber(); - } - return recordsToProcess; + void setHighestSequence(String highestSequence) { + this.lastCheckpointValue = highestSequence; } String getCheckpoint() { diff --git a/src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java b/src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java index 511e7d1..84e13c4 100644 --- a/src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java +++ b/src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java @@ -22,12 +22,15 @@ import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.atLeast; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; +import static org.springframework.integration.test.matcher.EqualsResultMatcher.equalsResult; +import static org.springframework.integration.test.matcher.EventuallyMatcher.eventually; import java.nio.ByteBuffer; import java.util.List; import java.util.Map; import java.util.Set; +import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -48,6 +51,7 @@ import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.metadata.ConcurrentMetadataStore; import org.springframework.integration.metadata.MetadataStore; import org.springframework.integration.metadata.SimpleMetadataStore; +import org.springframework.integration.support.locks.DefaultLockRegistry; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; @@ -149,9 +153,17 @@ public class KinesisMessageDrivenChannelAdapterTests { assertThat(this.checkpointStore.get("SpringIntegration" + ":" + STREAM1 + ":" + "1")).isEqualTo("2"); this.kinesisMessageDrivenChannelAdapter.stop(); + + Map forLocking = + TestUtils.getPropertyValue(this.kinesisMessageDrivenChannelAdapter, + "shardLocksMonitor.forLocking", Map.class); + + Assert.assertThat(0, eventually(equalsResult(forLocking::size))); + this.kinesisMessageDrivenChannelAdapter.setListenerMode(ListenerMode.batch); this.kinesisMessageDrivenChannelAdapter.setCheckpointMode(CheckpointMode.record); this.checkpointStore.put("SpringIntegration" + ":" + STREAM1 + ":" + "1", "1"); + this.kinesisMessageDrivenChannelAdapter.start(); message = this.kinesisChannel.receive(10000); @@ -169,19 +181,10 @@ public class KinesisMessageDrivenChannelAdapterTests { Object sequenceNumberHeader = message.getHeaders().get(AwsHeaders.RECEIVED_SEQUENCE_NUMBER); assertThat(sequenceNumberHeader).isInstanceOf(List.class); assertThat((List) sequenceNumberHeader).contains("2"); - int n = 0; - while (n++ < 100) { - if (!this.checkpointStore.get("SpringIntegration" + ":" + STREAM1 + ":" + "1").equals("2")) { - Thread.sleep(100); - } - else { - break; - } - } - - assertThat(n).isLessThan(100); - assertThat(this.checkpointStore.get("SpringIntegration" + ":" + STREAM1 + ":" + "1")).isEqualTo("2"); + Assert.assertThat("2", + eventually(equalsResult(() -> + this.checkpointStore.get("SpringIntegration" + ":" + STREAM1 + ":" + "1")))); List consumerInvoker = TestUtils.getPropertyValue(this.kinesisMessageDrivenChannelAdapter, "consumerInvokers", List.class); @@ -291,10 +294,22 @@ public class KinesisMessageDrivenChannelAdapterTests { .willReturn(new GetRecordsResult() .withNextShardIterator(shard1Iterator3)); + String shard1Iterator4 = "shard1Iterator4"; + given(amazonKinesis.getShardIterator(KinesisShardOffset.afterSequenceNumber(STREAM1, "1", "1") .toShardIteratorRequest())) .willReturn(new GetShardIteratorResult() - .withShardIterator(shard1Iterator2)); + .withShardIterator(shard1Iterator4)); + + given(amazonKinesis.getRecords(new GetRecordsRequest() + .withShardIterator(shard1Iterator4) + .withLimit(25))) + .willReturn(new GetRecordsResult() + .withNextShardIterator(shard1Iterator3) + .withRecords(new Record() + .withPartitionKey("partition1") + .withSequenceNumber("2") + .withData(ByteBuffer.wrap(serializingConverter.convert("bar"))))); return amazonKinesis; } @@ -315,6 +330,7 @@ public class KinesisMessageDrivenChannelAdapterTests { adapter.setOutputChannel(kinesisChannel()); adapter.setCheckpointStore(checkpointStore()); adapter.setCheckpointMode(CheckpointMode.manual); + adapter.setLockRegistry(new DefaultLockRegistry()); adapter.setStartTimeout(10000); adapter.setDescribeStreamRetries(1); adapter.setConcurrency(10); diff --git a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java index e5493d9..fa9cd84 100644 --- a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java +++ b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java @@ -44,6 +44,8 @@ import org.springframework.integration.metadata.ConcurrentMetadataStore; import org.springframework.integration.metadata.SimpleMetadataStore; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper; +import org.springframework.integration.support.locks.DefaultLockRegistry; +import org.springframework.integration.support.locks.LockRegistry; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -154,6 +156,11 @@ public class KinesisIntegrationTests { return new SimpleMetadataStore(); } + @Bean + public LockRegistry lockRegistry() { + return new DefaultLockRegistry(); + } + private KinesisMessageDrivenChannelAdapter kinesisMessageDrivenChannelAdapter() { KinesisMessageDrivenChannelAdapter adapter = new KinesisMessageDrivenChannelAdapter(KINESIS_LOCAL_RUNNING.getKinesis(), TEST_STREAM); @@ -161,6 +168,7 @@ public class KinesisIntegrationTests { adapter.setErrorChannel(errorChannel()); adapter.setErrorMessageStrategy(new KinesisMessageHeaderErrorMessageStrategy()); adapter.setCheckpointStore(checkpointStore()); + adapter.setLockRegistry(lockRegistry()); adapter.setEmbeddedHeadersMapper(new EmbeddedJsonHeadersMessageMapper("foo")); return adapter; }