Spring Aws Kinesis Binder 在使用消息时获取和释放 Dynamo DB 中的锁问题
Spring Aws Kinesis Binder Acquiring and Releasing lock issues in Dynamo DB while consuming messages
有时候，当我们突然停止应用程序时，会出现解锁失败的异常。那么同一组将永远不会收到消息。其他组正在接收消息。
我正在使用 aws kinesis binder 快照版本。
这是应用程序停止时的错误。
2018-07-19 22:21:21.371 ERROR 60981 --- [s-shard-locks-1] a.i.k.KinesisMessageDrivenChannelAdapter : Error during unlocking: DynamoDbLock [lockKey=aaaa:myStream:shardId-000000000000,lockedAt=2018-07-19@22:21:11.145, lockItem=null]
org.springframework.dao.DataAccessResourceFailureException: Failed to release lock at aaaa:myStream:shardId-000000000000; nested exception is java.util.concurrent.RejectedExecutionException: Task org.springframework.integration.aws.lock.DynamoDbLockRegistry$DynamoDbLock$$Lambda1/685035750@7835c79e rejected from java.util.concurrent.ThreadPoolExecutor@650635c5[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 1]
at org.springframework.integration.aws.lock.DynamoDbLockRegistry$DynamoDbLock.unlock(DynamoDbLockRegistry.java:538) ~[spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
at org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter$ShardConsumerManager.run(KinesisMessageDrivenChannelAdapter.java:1250) ~[spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_172]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_172]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_172]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_172]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_172]
Caused by: java.util.concurrent.RejectedExecutionException: Task org.springframework.integration.aws.lock.DynamoDbLockRegistry$DynamoDbLock$$Lambda1/685035750@7835c79e rejected from java.util.concurrent.ThreadPoolExecutor@650635c5[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 1]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063) ~[na:1.8.0_172]
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830) [na:1.8.0_172]
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379) [na:1.8.0_172]
at org.springframework.integration.aws.lock.DynamoDbLockRegistry$DynamoDbLock.unlock(DynamoDbLockRegistry.java:529) ~[spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
... 6 common frames omitted
然后我们启动应用程序，这是我们得到的错误。
2018-07-19 22:15:25.912 INFO 60969 --- [ Thread-3] ConfigServletWebServerApplicationContext : Closing org.springframework.boot.web.servlet.context.AnnotationConfigServletWebServerApplicationContext@62150f9e: startup date [Thu Jul 19 22:14:26 IST 2018]; root of context hierarchy
2018-07-19 22:15:25.914 INFO 60969 --- [ Thread-3] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@29b40b3: startup date [Thu Jul 19 22:14:33 IST 2018]; parent: org.springframework.boot.web.servlet.context.AnnotationConfigServletWebServerApplicationContext@62150f9e
2018-07-19 22:15:25.924 INFO 60969 --- [b-lock-client-1] c.a.s.d.AmazonDynamoDBLockClient : Heartbeat thread recieved interrupt, exiting run() (possibly exiting thread)
java.lang.InterruptedException: sleep interrupted
at java.lang.Thread.sleep(Native Method) [na:1.8.0_172]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient.run(AmazonDynamoDBLockClient.java:955) ~[dynamodb-lock-client-1.0.0.jar:na]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_172]
2018-07-19 22:15:25.928 INFO 60969 --- [ Thread-3] o.s.c.support.DefaultLifecycleProcessor : Stopping beans in phase 2147482647
2018-07-19 22:15:25.929 INFO 60969 --- [ Thread-3] a.i.k.KinesisMessageDrivenChannelAdapter : stopped KinesisMessageDrivenChannelAdapter{shardOffsets=[KinesisShardOffset{iteratorType=TRIM_HORIZON, sequenceNumber='null', timestamp=null, stream='myStream', shard='shardId-000000000000', reset=false}], consumerGroup='aaaa'}
2018-07-19 22:15:25.930 INFO 60969 --- [ Thread-3] o.s.c.stream.binder.BinderErrorChannel : Channel 'application.myStream.aaaa.errors' has 1 subscriber(s).
2018-07-19 22:15:25.931 INFO 60969 --- [ Thread-3] o.s.c.stream.binder.BinderErrorChannel : Channel 'application.myStream.aaaa.errors' has 0 subscriber(s).
2018-07-19 22:15:25.931 WARN 60969 --- [s-shard-locks-1] c.a.s.d.AmazonDynamoDBLockClient : Could not acquire lock because of a client side failure in talking to DDB
com.amazonaws.AbortedException:
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleInterruptedException(AmazonHttpClient.java:795) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.execute(AmazonHttpClient.java:701) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.access0(AmazonHttpClient.java:667) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutionBuilderImpl.execute(AmazonHttpClient.java:649) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:513) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient.doInvoke(AmazonDynamoDBClient.java:3452) ~[aws-java-sdk-dynamodb-1.11.336.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient.invoke(AmazonDynamoDBClient.java:3428) ~[aws-java-sdk-dynamodb-1.11.336.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient.executeGetItem(AmazonDynamoDBClient.java:1789) ~[aws-java-sdk-dynamodb-1.11.336.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient.getItem(AmazonDynamoDBClient.java:1764) ~[aws-java-sdk-dynamodb-1.11.336.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient.readFromDynamoDB(AmazonDynamoDBLockClient.java:997) [dynamodb-lock-client-1.0.0.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient.getLockFromDynamoDB(AmazonDynamoDBLockClient.java:743) [dynamodb-lock-client-1.0.0.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient.acquireLock(AmazonDynamoDBLockClient.java:402) [dynamodb-lock-client-1.0.0.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient.tryAcquireLock(AmazonDynamoDBLockClient.java:567) [dynamodb-lock-client-1.0.0.jar:na]
at org.springframework.integration.aws.lock.DynamoDbLockRegistry$DynamoDbLock.doLock(DynamoDbLockRegistry.java:504) [spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
at org.springframework.integration.aws.lock.DynamoDbLockRegistry$DynamoDbLock.tryLock(DynamoDbLockRegistry.java:478) [spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
at org.springframework.integration.aws.lock.DynamoDbLockRegistry$DynamoDbLock.tryLock(DynamoDbLockRegistry.java:451) [spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
at org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter$ShardConsumerManager.lambda$run[=12=](KinesisMessageDrivenChannelAdapter.java:1198) [spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
at java.util.Collection.removeIf(Collection.java:414) ~[na:1.8.0_172]
at org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter$ShardConsumerManager.run(KinesisMessageDrivenChannelAdapter.java:1191) [spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[na:1.8.0_172]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[na:1.8.0_172]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[na:1.8.0_172]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[na:1.8.0_172]
at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_172]
Caused by: java.lang.InterruptedException: sleep interrupted
at java.lang.Thread.sleep(Native Method) ~[na:1.8.0_172]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.doPauseBeforeRetry(AmazonHttpClient.java:1671) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.pauseBeforeRetry(AmazonHttpClient.java:1645) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeOneRequest(AmazonHttpClient.java:1191) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeHelper(AmazonHttpClient.java:1058) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.doExecute(AmazonHttpClient.java:743) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeWithTimer(AmazonHttpClient.java:717) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.execute(AmazonHttpClient.java:699) ~[aws-java-sdk-core-1.11.336.jar:na]
... 22 common frames omitted
我会说这是任何分布式锁实现的正常行为：当持有锁的线程被破坏时，在数据库级别上没有任何钩子可以解锁。
你需要学习一个leaseDuration属性的AmazonDynamoDBLockClient：https://aws.amazon.com/blogs/database/building-distributed-locks-with-the-dynamodb-lock-client/
虽然你在应用程序停止的情况下显示了两个堆栈跟踪...
有没有机会做一个简单的项目让我们一起玩？
更新
问题已通过以下方式解决：https://github.com/spring-projects/spring-integration-aws/commit/dd52e8f304d20bbdc46d3f9da77e26dd7977f8b7
再次感谢您的好评！
有时候，当我们突然停止应用程序时，会出现解锁失败的异常。那么同一组将永远不会收到消息。其他组正在接收消息。
我正在使用 aws kinesis binder 快照版本。
这是应用程序停止时的错误。
2018-07-19 22:21:21.371 ERROR 60981 --- [s-shard-locks-1] a.i.k.KinesisMessageDrivenChannelAdapter : Error during unlocking: DynamoDbLock [lockKey=aaaa:myStream:shardId-000000000000,lockedAt=2018-07-19@22:21:11.145, lockItem=null]
org.springframework.dao.DataAccessResourceFailureException: Failed to release lock at aaaa:myStream:shardId-000000000000; nested exception is java.util.concurrent.RejectedExecutionException: Task org.springframework.integration.aws.lock.DynamoDbLockRegistry$DynamoDbLock$$Lambda1/685035750@7835c79e rejected from java.util.concurrent.ThreadPoolExecutor@650635c5[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 1]
at org.springframework.integration.aws.lock.DynamoDbLockRegistry$DynamoDbLock.unlock(DynamoDbLockRegistry.java:538) ~[spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
at org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter$ShardConsumerManager.run(KinesisMessageDrivenChannelAdapter.java:1250) ~[spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_172]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_172]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_172]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_172]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_172]
Caused by: java.util.concurrent.RejectedExecutionException: Task org.springframework.integration.aws.lock.DynamoDbLockRegistry$DynamoDbLock$$Lambda1/685035750@7835c79e rejected from java.util.concurrent.ThreadPoolExecutor@650635c5[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 1]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063) ~[na:1.8.0_172]
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830) [na:1.8.0_172]
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379) [na:1.8.0_172]
at org.springframework.integration.aws.lock.DynamoDbLockRegistry$DynamoDbLock.unlock(DynamoDbLockRegistry.java:529) ~[spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
... 6 common frames omitted
然后我们启动应用程序，这是我们得到的错误。
2018-07-19 22:15:25.912 INFO 60969 --- [ Thread-3] ConfigServletWebServerApplicationContext : Closing org.springframework.boot.web.servlet.context.AnnotationConfigServletWebServerApplicationContext@62150f9e: startup date [Thu Jul 19 22:14:26 IST 2018]; root of context hierarchy
2018-07-19 22:15:25.914 INFO 60969 --- [ Thread-3] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@29b40b3: startup date [Thu Jul 19 22:14:33 IST 2018]; parent: org.springframework.boot.web.servlet.context.AnnotationConfigServletWebServerApplicationContext@62150f9e
2018-07-19 22:15:25.924 INFO 60969 --- [b-lock-client-1] c.a.s.d.AmazonDynamoDBLockClient : Heartbeat thread recieved interrupt, exiting run() (possibly exiting thread)
java.lang.InterruptedException: sleep interrupted
at java.lang.Thread.sleep(Native Method) [na:1.8.0_172]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient.run(AmazonDynamoDBLockClient.java:955) ~[dynamodb-lock-client-1.0.0.jar:na]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_172]
2018-07-19 22:15:25.928 INFO 60969 --- [ Thread-3] o.s.c.support.DefaultLifecycleProcessor : Stopping beans in phase 2147482647
2018-07-19 22:15:25.929 INFO 60969 --- [ Thread-3] a.i.k.KinesisMessageDrivenChannelAdapter : stopped KinesisMessageDrivenChannelAdapter{shardOffsets=[KinesisShardOffset{iteratorType=TRIM_HORIZON, sequenceNumber='null', timestamp=null, stream='myStream', shard='shardId-000000000000', reset=false}], consumerGroup='aaaa'}
2018-07-19 22:15:25.930 INFO 60969 --- [ Thread-3] o.s.c.stream.binder.BinderErrorChannel : Channel 'application.myStream.aaaa.errors' has 1 subscriber(s).
2018-07-19 22:15:25.931 INFO 60969 --- [ Thread-3] o.s.c.stream.binder.BinderErrorChannel : Channel 'application.myStream.aaaa.errors' has 0 subscriber(s).
2018-07-19 22:15:25.931 WARN 60969 --- [s-shard-locks-1] c.a.s.d.AmazonDynamoDBLockClient : Could not acquire lock because of a client side failure in talking to DDB
com.amazonaws.AbortedException:
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleInterruptedException(AmazonHttpClient.java:795) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.execute(AmazonHttpClient.java:701) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.access0(AmazonHttpClient.java:667) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutionBuilderImpl.execute(AmazonHttpClient.java:649) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:513) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient.doInvoke(AmazonDynamoDBClient.java:3452) ~[aws-java-sdk-dynamodb-1.11.336.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient.invoke(AmazonDynamoDBClient.java:3428) ~[aws-java-sdk-dynamodb-1.11.336.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient.executeGetItem(AmazonDynamoDBClient.java:1789) ~[aws-java-sdk-dynamodb-1.11.336.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient.getItem(AmazonDynamoDBClient.java:1764) ~[aws-java-sdk-dynamodb-1.11.336.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient.readFromDynamoDB(AmazonDynamoDBLockClient.java:997) [dynamodb-lock-client-1.0.0.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient.getLockFromDynamoDB(AmazonDynamoDBLockClient.java:743) [dynamodb-lock-client-1.0.0.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient.acquireLock(AmazonDynamoDBLockClient.java:402) [dynamodb-lock-client-1.0.0.jar:na]
at com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient.tryAcquireLock(AmazonDynamoDBLockClient.java:567) [dynamodb-lock-client-1.0.0.jar:na]
at org.springframework.integration.aws.lock.DynamoDbLockRegistry$DynamoDbLock.doLock(DynamoDbLockRegistry.java:504) [spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
at org.springframework.integration.aws.lock.DynamoDbLockRegistry$DynamoDbLock.tryLock(DynamoDbLockRegistry.java:478) [spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
at org.springframework.integration.aws.lock.DynamoDbLockRegistry$DynamoDbLock.tryLock(DynamoDbLockRegistry.java:451) [spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
at org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter$ShardConsumerManager.lambda$run[=12=](KinesisMessageDrivenChannelAdapter.java:1198) [spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
at java.util.Collection.removeIf(Collection.java:414) ~[na:1.8.0_172]
at org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter$ShardConsumerManager.run(KinesisMessageDrivenChannelAdapter.java:1191) [spring-integration-aws-2.0.0.BUILD-SNAPSHOT.jar:na]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[na:1.8.0_172]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[na:1.8.0_172]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[na:1.8.0_172]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[na:1.8.0_172]
at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_172]
Caused by: java.lang.InterruptedException: sleep interrupted
at java.lang.Thread.sleep(Native Method) ~[na:1.8.0_172]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.doPauseBeforeRetry(AmazonHttpClient.java:1671) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.pauseBeforeRetry(AmazonHttpClient.java:1645) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeOneRequest(AmazonHttpClient.java:1191) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeHelper(AmazonHttpClient.java:1058) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.doExecute(AmazonHttpClient.java:743) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeWithTimer(AmazonHttpClient.java:717) ~[aws-java-sdk-core-1.11.336.jar:na]
at com.amazonaws.http.AmazonHttpClient$RequestExecutor.execute(AmazonHttpClient.java:699) ~[aws-java-sdk-core-1.11.336.jar:na]
... 22 common frames omitted
我会说这是任何分布式锁实现的正常行为：当持有锁的线程被破坏时，在数据库级别上没有任何钩子可以解锁。
你需要学习一个leaseDuration属性的AmazonDynamoDBLockClient：https://aws.amazon.com/blogs/database/building-distributed-locks-with-the-dynamodb-lock-client/
虽然你在应用程序停止的情况下显示了两个堆栈跟踪...
有没有机会做一个简单的项目让我们一起玩？
更新
问题已通过以下方式解决：https://github.com/spring-projects/spring-integration-aws/commit/dd52e8f304d20bbdc46d3f9da77e26dd7977f8b7
再次感谢您的好评！