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

Add lock renewal for KinesisChannelAdapter

Browse Source 
Related to https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis/issues/148

The current implementation has a flaw when it uses a distributed lock
for an exclusive access to shard for consuming only once at start up.
Such a behavior cause the problem when we have a network glitch at
runtime, so the lock is broken, but consumer is still active to retry
consumption attempts

* Add `renewLockIfAny()` logic ot the `ShardConsumer`, so we ensure
that we are still a lock holder and don't consume otherwise
* Add `unlockFuture` logic to block the `ShardConsumer.stop()`
until we really got lock unlocked.
Otherwise we end up with the race condition when we are still
stopping, but already ready to start a new consumer for the same
shard
This commit is contained in:
Artem Bilan
2020-12-18 15:57:44 -05:00
parent c15e345470
commit ac74dfd236
1 changed files with 104 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -29,6 +29,7 @@ import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentSkipListSet;
@@ -915,7 +916,17 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
void stop() {
this.state = ConsumerState.STOP;
if (KinesisMessageDrivenChannelAdapter.this.lockRegistry != null) {
KinesisMessageDrivenChannelAdapter.this.shardConsumerManager.unlock(this.key);
LockCompletableFuture unlockFuture = new LockCompletableFuture(this.key);
KinesisMessageDrivenChannelAdapter.this.shardConsumerManager.unlock(unlockFuture);
try {
unlockFuture.get(1, TimeUnit.SECONDS);
}
catch (Exception ex) {
if (ex instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
logger.info("The lock for key '" + this.key + "' was not unlocked in time", ex);
}
}
if (this.notifier != null) {
this.notifier.run();
@@ -929,6 +940,10 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
void execute() {
if (this.task == null) {
if (!renewLockIfAny()) {
return;
}
switch (this.state) {
case NEW:
case EXPIRED:
@@ -1007,6 +1022,36 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
}
}
private boolean renewLockIfAny() {
if (KinesisMessageDrivenChannelAdapter.this.lockRegistry != null && this.state == ConsumerState.CONSUME) {
LockCompletableFuture renewLockFuture = new LockCompletableFuture(this.key);
KinesisMessageDrivenChannelAdapter.this.shardConsumerManager.renewLock(renewLockFuture);
boolean lockRenewed = false;
try {
lockRenewed = renewLockFuture.get(1, TimeUnit.SECONDS);
}
catch (Exception ex) {
if (ex instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
logger.info("The lock for key '" + this.key + "' was not renewed in time", ex);
}
if (!lockRenewed && this.state == ConsumerState.CONSUME) {
this.state = ConsumerState.STOP;
this.checkpointer.close();
if (this.notifier != null) {
this.notifier.run();
}
if (KinesisMessageDrivenChannelAdapter.this.active) {
KinesisMessageDrivenChannelAdapter.this.shardConsumerManager.addShardToConsume(this.shardOffset);
}
return false;
}
}
return true;
}
private Runnable processTask() {
return () -> {
GetRecordsRequest getRecordsRequest = new GetRecordsRequest();
@@ -1368,7 +1413,9 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
private final Map<String, Lock> locks = new HashMap<>();
private final Queue<String> forUnlocking = new ConcurrentLinkedQueue<>();
private final Queue<LockCompletableFuture> forUnlocking = new ConcurrentLinkedQueue<>();
private final Queue<LockCompletableFuture> forRenewing = new ConcurrentLinkedQueue<>();
ShardConsumerManager() {
}
@@ -1379,15 +1426,18 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
this.shardOffsetsToConsumer.put(lockKey, kinesisShardOffset);
}
void unlock(String lockKey) {
this.forUnlocking.add(lockKey);
void unlock(LockCompletableFuture unlockFuture) {
this.forUnlocking.add(unlockFuture);
}
void renewLock(LockCompletableFuture renewLockFuture) {
this.forRenewing.add(renewLockFuture);
}
@Override
public void run() {
try {
while (!Thread.currentThread().isInterrupted()) {
this.shardOffsetsToConsumer
.entrySet()
.removeIf(
@@ -1418,9 +1468,9 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
});
while (KinesisMessageDrivenChannelAdapter.this.lockRegistry != null) {
String lockKey = this.forUnlocking.poll();
if (lockKey != null) {
Lock lock = this.locks.remove(lockKey);
LockCompletableFuture forUnlocking = this.forUnlocking.poll();
if (forUnlocking != null) {
Lock lock = this.locks.remove(forUnlocking.lockKey);
if (lock != null) {
try {
lock.unlock();
@@ -1429,13 +1479,48 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
logger.error("Error during unlocking: " + lock, e);
}
}
forUnlocking.complete(true);
}
else {
break;
}
}
sleep(250, new IllegalStateException("ShardConsumerManager Thread [" + this + "] has been interrupted"), true);
while (KinesisMessageDrivenChannelAdapter.this.lockRegistry != null) {
LockCompletableFuture lockFuture = this.forRenewing.poll();
if (lockFuture != null) {
Lock lock = this.locks.get(lockFuture.lockKey);
if (lock != null) {
try {
if (lock.tryLock()) {
try {
lockFuture.complete(true);
}
finally {
lock.unlock();
}
}
else {
lockFuture.complete(false);
this.locks.remove(lockFuture.lockKey);
}
}
catch (Exception e) {
logger.error("Error during locking: " + lock, e);
}
}
else {
lockFuture.complete(false);
}
}
else {
break;
}
}
sleep(250,
new IllegalStateException("ShardConsumerManager Thread [" + this + "] has been interrupted"),
true);
}
}
finally {
@@ -1461,4 +1546,14 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
}
private final static class LockCompletableFuture extends CompletableFuture<Boolean> {
private final String lockKey;
LockCompletableFuture(String lockKey) {
this.lockKey = lockKey;
}
}
}