From 0e02b0d47067ade4698f5268dd0ed947b573fb5f Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 16 Jul 2018 13:17:13 -0400 Subject: [PATCH] GH-90: Rework locking logic in the KinesisMDChA Fixes https://github.com/spring-projects/spring-integration-aws/issues/90 * Add an internal `ShardConsumerManager` which is responsible to initiate a locking for the shard key in the provided consumer group and populating a `ShardConsumer` if `tryLock()` is successful or no `LockRegistry` at all * Additional logic is added to always iterate over candidate shards if `tryLock()` on the matter is not successful. This way the current `KinesisMessageDrivenChannelAdapter` picks up those shards which have been locked by the consumer which has just left a cluster and unlocked distributed locks * Now all the shards are considered as candidates independently of the `streams` or `shardOffsets` configuration * Improve Kinesis tests performance --- README.md | 8 +- .../KinesisMessageDrivenChannelAdapter.java | 177 ++++++++---------- ...nesisMessageDrivenChannelAdapterTests.java | 2 +- .../aws/kinesis/KinesisIntegrationTests.java | 7 + 4 files changed, 84 insertions(+), 110 deletions(-) diff --git a/README.md b/README.md index f9c8961..5a9994e 100644 --- a/README.md +++ b/README.md @@ -552,11 +552,9 @@ 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. +Starting with _version 2.0_, the `KinesisMessageDrivenChannelAdapter` can be configured with the `LockRegistry` for leader selection for the provided shards or derived from the provided streams. +The `KinesisMessageDrivenChannelAdapter` iterates over its shards and tries to acquire a distributed lock for the shard in its consumer group. +If `LockRegistry` is not provided, no exclusive locking happens and all the shards are consumed by this `KinesisMessageDrivenChannelAdapter`. See also `DynamoDbLockRegistry` for more information. ### Outbound Channel Adapter 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 3eec15b..39e0663 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 @@ -36,6 +36,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; @@ -62,7 +63,6 @@ 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; @@ -104,7 +104,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i private final List consumerInvokers = new ArrayList<>(); - private final ShardLocksMonitor shardLocksMonitor = new ShardLocksMonitor(); + private final ShardConsumerManager shardConsumerManager = new ShardConsumerManager(); private final ExecutorService shardLocksExecutor = Executors.newSingleThreadExecutor( @@ -160,6 +160,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i private volatile int consumerInvokerMaxCapacity; + private volatile Future shardConsumerManagerFuture; + public KinesisMessageDrivenChannelAdapter(AmazonKinesis amazonKinesis, String... streams) { Assert.notNull(amazonKinesis, "'amazonKinesis' must not be null."); Assert.notEmpty(streams, "'streams' must not be null."); @@ -362,7 +364,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i synchronized (this.shardOffsets) { for (KinesisShardOffset shardOffset : this.shardOffsets) { if (shardOffsetForSearch.equals(shardOffset)) { - populateConsumer(shardOffset); + this.shardConsumerManager.addShardToConsume(shardOffset); break; } } @@ -410,7 +412,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i oldShardConsumer.close(); } shardOffset.setReset(true); - populateConsumer(shardOffset); + this.shardConsumerManager.addShardToConsume(shardOffset); } } @@ -432,10 +434,6 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i "because it does not make sense in case of [ListenerMode.batch]."); } - if (this.lockRegistry != null) { - this.shardLocksMonitor.start(); - } - if (this.streams != null) { populateShardsForStreams(); } @@ -446,15 +444,9 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i this.concurrency = Math.min(this.maxConcurrency, this.shardOffsets.size()); - for (int i = 0; i < this.concurrency; i++) { - Collection shardConsumers = shardConsumerSubset(i); - this.consumerInvokerMaxCapacity = Math.max(this.consumerInvokerMaxCapacity, shardConsumers.size()); - ConsumerInvoker consumerInvoker = new ConsumerInvoker(shardConsumers); - this.consumerInvokers.add(consumerInvoker); - this.consumerExecutor.execute(consumerInvoker); - } - this.dispatcherExecutor.execute(new ConsumerDispatcher()); + + this.shardConsumerManagerFuture = this.shardLocksExecutor.submit(this.shardConsumerManager); } private Collection shardConsumerSubset(int i) { @@ -569,9 +561,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } } - if (this.lockRegistry == null || this.shardLocksMonitor.tryLock(key)) { - shardsToConsume.add(shard); - } + shardsToConsume.add(shard); } } catch (Exception e) { @@ -598,7 +588,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i addedOffset = this.shardOffsets.add(shardOffset); } if (addedOffset && shardsGatherLatch == null && this.active) { - populateConsumer(shardOffset); + this.shardConsumerManager.addShardToConsume(shardOffset); } } } @@ -614,8 +604,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i private void populateConsumers() { synchronized (this.shardOffsets) { for (KinesisShardOffset shardOffset : this.shardOffsets) { - shardOffset.setReset(this.resetCheckpoints); - populateConsumer(shardOffset); + this.shardConsumerManager.addShardToConsume(shardOffset); } } @@ -623,6 +612,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } private void populateConsumer(KinesisShardOffset shardOffset) { + shardOffset.setReset(this.resetCheckpoints); ShardConsumer shardConsumer = new ShardConsumer(shardOffset); if (this.active) { @@ -664,7 +654,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i super.doStop(); stopConsumers(); - this.shardLocksMonitor.stop(); + this.shardConsumerManagerFuture.cancel(true); this.active = false; } @@ -801,7 +791,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i void stop() { this.state = ConsumerState.STOP; if (KinesisMessageDrivenChannelAdapter.this.lockRegistry != null) { - KinesisMessageDrivenChannelAdapter.this.shardLocksMonitor.unlock(this.key); + KinesisMessageDrivenChannelAdapter.this.shardConsumerManager.unlock(this.key); } if (this.notifier != null) { this.notifier.run(); @@ -1172,85 +1162,70 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } - private final class ShardLocksMonitor implements SchedulingAwareRunnable { + private final class ShardConsumerManager implements SchedulingAwareRunnable { - private final Map> forLocking = - Collections.synchronizedMap(new HashMap<>()); + private final Map shardOffsetsToConsumer = new ConcurrentHashMap<>(); - private final Queue forUnlocking = new ConcurrentLinkedQueue<>(); + private final Map locks = new HashMap<>(); - private volatile boolean active = true; + private final Queue forUnlocking = new ConcurrentLinkedQueue<>(); - 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; - } - finally { - this.forLocking.remove(lock); - } + ShardConsumerManager() { + } + + void addShardToConsume(KinesisShardOffset kinesisShardOffset) { + String lockKey = buildCheckpointKeyForShard(kinesisShardOffset.getStream(), kinesisShardOffset.getShard()); + this.shardOffsetsToConsumer.put(lockKey, kinesisShardOffset); } 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; + this.forUnlocking.add(lockKey); } @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) { + while (!Thread.currentThread().isInterrupted()) { + + this.shardOffsetsToConsumer.entrySet() + .removeIf(entry -> { + boolean remove = true; + if (KinesisMessageDrivenChannelAdapter.this.lockRegistry != null) { + String key = entry.getKey(); + Lock lock = + KinesisMessageDrivenChannelAdapter.this.lockRegistry.obtain(key); + try { + if (lock.tryLock()) { + this.locks.put(key, lock); + } + else { + remove = false; + } + + } + catch (Exception e) { + logger.error("Error during locking: " + lock, e); + } + } + + if (remove) { + populateConsumer(entry.getValue()); + } + + return remove; + }); + + while (KinesisMessageDrivenChannelAdapter.this.lockRegistry != null) { + String lockKey = this.forUnlocking.poll(); + if (lockKey != null) { + Lock lock = this.locks.remove(lockKey); + if (lock != null) { try { - if (lock.tryLock()) { - settableFuture.set(true); - } - else { - settableFuture.set(false); - } + lock.unlock(); } catch (Exception e) { - logger.error("Error during locking: " + lock, e); - settableFuture.set(false); + logger.error("Error during unlocking: " + 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 { @@ -1263,31 +1238,25 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new IllegalStateException("ShardLocksMonitor Thread [" + + throw new IllegalStateException("ShardConsumerManager 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(); - } + for (Iterator iterator = this.locks.values().iterator(); iterator.hasNext(); ) { + Lock lock = iterator.next(); + try { + lock.unlock(); + } + catch (Exception e) { + logger.error("Error during unlocking: " + lock, e); + } + finally { + iterator.remove(); } } } - - this.forUnlocking.clear(); } @Override 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 88db315..3cc8329 100644 --- a/src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java +++ b/src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java @@ -156,7 +156,7 @@ public class KinesisMessageDrivenChannelAdapterTests { Map forLocking = TestUtils.getPropertyValue(this.kinesisMessageDrivenChannelAdapter, - "shardLocksMonitor.forLocking", Map.class); + "shardConsumerManager.locks", Map.class); Assert.assertThat(0, eventually(100, 100, equalsResult(forLocking::size))); 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 607d375..38595d8 100644 --- a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java +++ b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java @@ -29,6 +29,7 @@ import org.junit.ClassRule; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -170,6 +171,12 @@ public class KinesisIntegrationTests { adapter.setCheckpointStore(checkpointStore()); adapter.setLockRegistry(lockRegistry()); adapter.setEmbeddedHeadersMapper(new EmbeddedJsonHeadersMessageMapper("foo")); + + DirectFieldAccessor dfa = new DirectFieldAccessor(adapter); + dfa.setPropertyValue("describeStreamBackoff", 10); + dfa.setPropertyValue("consumerBackoff", 10); + dfa.setPropertyValue("idleBetweenPolls", 1); + return adapter; }