GH-158: Remove CLOSED shards from locking

Fixes https://github.com/spring-projects/spring-integration-aws/issues/158

It turns out that closed shards are still candidates for consumption
when we have concurrency
* Remove closed shards from the `shardConsumerManager.shardOffsetsToConsumer`
* Upgrade to Spring Integration 5.2.3, Jackson 2.10.2
This commit is contained in:
Artem Bilan
2020-02-04 16:00:13 -05:00
parent 55b12dc4a8
commit 6b82fde8a1
2 changed files with 7 additions and 3 deletions

View File

@@ -26,13 +26,13 @@ ext {
assertjVersion = '3.14.0'
awaitilityVersion = '4.0.1'
dynamodbLockClientVersion = '1.1.0'
jacksonVersion = '2.10.1'
jacksonVersion = '2.10.2'
junitVersion = '5.5.2'
servletApiVersion = '4.0.1'
localstackVersion = '0.1.22'
log4jVersion = '2.13.0'
springCloudAwsVersion = '2.2.1.RELEASE'
springIntegrationVersion = '5.2.2.RELEASE'
springIntegrationVersion = '5.2.3.RELEASE'
kinesisClientVersion = '1.13.0'
kinesisProducerVersion = '0.14.0'

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2019 the original author or authors.
* Copyright 2017-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -927,6 +927,10 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
this.shardIterator = result.getNextShardIterator();
if (this.shardIterator == null) {
if (KinesisMessageDrivenChannelAdapter.this.lockRegistry != null) {
KinesisMessageDrivenChannelAdapter.this.shardConsumerManager.shardOffsetsToConsumer
.remove(this.key);
}
// Shard is closed: nothing to consume any more.
// Resharding is possible.
if (KinesisMessageDrivenChannelAdapter.this.applicationEventPublisher != null) {