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

GH-90: Add shard locking support to KinesisMDChA

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

* The `KinesisMessageDrivenChannelAdapter` can now be supplied with the
`LockRegistry` (e.g. `DynamoDbLockRegistry`) and when stream-based
configuration is used, the channel adapter performs `tryLock()` for the
shard in the channel adapter consumer group.
Therefor only one listener in the group is able to consume from the shard

Note: there is no yet full support for rebalance functionality.
And such a feature can be implemented using Spring Cloud Bus with the
command to stop and start channel adapters when a new
`KinesisMessageDrivenChannelAdapter` arrives to the cluster
This commit is contained in:
Artem Bilan
2018-06-19 13:05:00 -04:00
parent fca661404a
commit ebd562d6da
6 changed files with 270 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
README.md
View File

@@ -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

4
build.gradle
View File

@@ -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)

284
src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java
View File

@@ -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<ConsumerInvoker> 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<byte[]> 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<Record> recordsToProcess) {
List<Record> records = this.checkpointer.filterRecords(recordsToProcess);
if (!records.isEmpty()) {
if (logger.isTraceEnabled()) {
logger.trace("Processing records: " + records + " for [" + ShardConsumer.this + "]");
}
private void processRecords(List<Record> 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<String> partitionKeys;
final List<String> 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<String> partitionKeys;
final List<String> 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<Lock, SettableListenableFuture<Boolean>> forLocking =
Collections.synchronizedMap(new HashMap<>());
private final Queue<Lock> forUnlocking = new ConcurrentLinkedQueue<>();
private volatile boolean active = true;
boolean tryLock(String lockKey) {
SettableListenableFuture<Boolean> 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<Map.Entry<Lock, SettableListenableFuture<Boolean>>> entrySet = this.forLocking.entrySet();
try {
while (this.active) {
synchronized (this.forLocking) {
for (Map.Entry<Lock, SettableListenableFuture<Boolean>> entry : entrySet) {
Lock lock = entry.getKey();
SettableListenableFuture<Boolean> 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<Map.Entry<Lock, SettableListenableFuture<Boolean>>> iterator = entrySet.iterator();
iterator.hasNext(); ) {
Map.Entry<Lock, SettableListenableFuture<Boolean>> 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;
}
}
}

31
src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java
View File

@@ -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<Record> filterRecords(List<Record> records) {
List<Record> recordsToProcess = new LinkedList<>(records);
this.lastCheckpointValue = getCheckpoint();
if (this.lastCheckpointValue != null) {
for (Iterator<Record> 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() {

42
src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java
View File

@@ -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<String>) 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);

8
src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java
View File

@@ -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;
}