Worker lost leases on shards after error

[tyger]

Just recently we encountered strange behaviour of KCL in our system.
Our configuration:

• 2 machines in cluster
• Kinesis stream with 5 shards

In short: We encountered short network blip or something which caused timeout for some leases and they were lost and neither of two workers picked up them again.

Before error first worker had been processing one shard and second worker had been processing four shards.

14:59:02,383	rid:	worker-Some(dd557a7b-7294-4ac9-bb9b-c98edd3e9aa5)-run-18dcd724-d03a-4b32-adb7-3bd9d646d41b-0	c.a.s.k.c.l.worker.Worker	- Current stream shard assignments: shardId-000000000004, shardId-000000000003, shardId-000000000002, shardId-000000000001
14:59:02,487	rid:	worker-Some(1209e181-10b3-4302-b702-b3df6598b3db)-run-1b0c931a-ab2b-427a-b9fe-bb34cbab1304-0	c.a.s.k.c.l.worker.Worker	- Current stream shard assignments: shardId-000000000000

then error happened (repeating stacktraces have been skipped), as it is seen from first three records second worker lost leases for three shards.

15:00:08,160	rid:	LeaseRenewer-27	c.a.s.k.l.i.LeaseRenewer	- Worker 26a0eef5-c51e-46e7-92aa-4e547f14ec8f lost lease with key shardId-000000000002
15:00:08,160	rid:	LeaseRenewer-26	c.a.s.k.l.i.LeaseRenewer	- Worker 26a0eef5-c51e-46e7-92aa-4e547f14ec8f lost lease with key shardId-000000000003
15:00:08,160	rid:	LeaseRenewer-25	c.a.s.k.l.i.LeaseRenewer	- Worker 26a0eef5-c51e-46e7-92aa-4e547f14ec8f lost lease with key shardId-000000000004
15:00:12,590	rid:	RecordProcessor-0001	c.a.s.k.c.l.w.KinesisClientLibLeaseCoordinator	- Worker 26a0eef5-c51e-46e7-92aa-4e547f14ec8f could not update checkpoint for shard shardId-000000000003 because it does not hold the lease
15:00:12,591	ERROR	rid:	RecordProcessor-0001	c.s.s.a.a.c.UserSegmentsEventConsumer	- Cannot process records
com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException: Can't update checkpoint - instance doesn't hold the lease for this shard
	at com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibLeaseCoordinator.setCheckpoint(KinesisClientLibLeaseCoordinator.java:173) ~[ate-an-user-segs-publisher.jar:1.0.0]
	at com.amazonaws.services.kinesis.clientlibrary.lib.worker.RecordProcessorCheckpointer.advancePosition(RecordProcessorCheckpointer.java:216) ~[ate-an-user-segs-publisher.jar:1.0.0]
	at com.amazonaws.services.kinesis.clientlibrary.lib.worker.RecordProcessorCheckpointer.checkpoint(RecordProcessorCheckpointer.java:77) ~[ate-an-user-segs-publisher.jar:1.0.0]
	at com.companyname.kinesis.consumer.KinesisConsumerBase.checkpoint(KinesisConsumerBase.scala:105) ~[ate-an-user-segs-publisher.jar:1.0.0]
	at com.companyname.kinesis.consumer.KinesisConsumerBase.processRecords(KinesisConsumerBase.scala:58) ~[ate-an-user-segs-publisher.jar:1.0.0]
	at com.amazonaws.services.kinesis.clientlibrary.lib.worker.ProcessTask.call(ProcessTask.java:176) [ate-an-user-segs-publisher.jar:1.0.0]
	at com.amazonaws.services.kinesis.clientlibrary.lib.worker.MetricsCollectingTaskDecorator.call(MetricsCollectingTaskDecorator.java:49) [ate-an-user-segs-publisher.jar:1.0.0]
	at com.amazonaws.services.kinesis.clientlibrary.lib.worker.MetricsCollectingTaskDecorator.call(MetricsCollectingTaskDecorator.java:24) [ate-an-user-segs-publisher.jar:1.0.0]
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_66]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [na:1.8.0_66]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [na:1.8.0_66]
	at java.lang.Thread.run(Thread.java:745) [na:1.8.0_66]
15:00:13,013	rid:	RecordProcessor-0000	c.a.s.k.c.l.w.KinesisClientLibLeaseCoordinator	- Worker 26a0eef5-c51e-46e7-92aa-4e547f14ec8f could not update checkpoint for shard shardId-000000000002 because it does not hold the lease
15:00:13,013	ERROR	rid:	RecordProcessor-0000	c.s.s.a.a.c.UserSegmentsEventConsumer	- Cannot process records
com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException: Can't update checkpoint - instance doesn't hold the lease for this shard
	<skipped>
15:00:13,645	rid:	RecordProcessor-0002	c.a.s.k.c.l.w.KinesisClientLibLeaseCoordinator	- Worker 26a0eef5-c51e-46e7-92aa-4e547f14ec8f could not update checkpoint for shard shardId-000000000004 because it does not hold the lease
15:00:13,661	rid:	RecordProcessor-0002	c.s.s.a.a.c.UserSegmentsEventConsumer	- Cannot process records
com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException: Can't update checkpoint - instance doesn't hold the lease for this shard
	<skipped>

then system kept working with such assignments:

15:02:05,517	rid:	worker-Some(1209e181-10b3-4302-b702-b3df6598b3db)-run-1b0c931a-ab2b-427a-b9fe-bb34cbab1304-0	c.a.s.k.c.l.worker.Worker	- Current stream shard assignments: shardId-000000000000
15:02:11,617	rid:	worker-Some(dd557a7b-7294-4ac9-bb9b-c98edd3e9aa5)-run-18dcd724-d03a-4b32-adb7-3bd9d646d41b-0	c.a.s.k.c.l.worker.Worker	- Current stream shard assignments: shardId-000000000001

So, obviously, second worker lost leases for three shards and nobody were processing it.

After we noticed it the second machine was restarted:

16:24:38,348	rid:	worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0	c.a.s.k.c.l.worker.Worker	- Initialization attempt 1
[INFO] [02/14/2017 16:24:38.348] [worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0] [akka://WatchdogSystem/user/$a/$a] Starting the worker f5ecf23e-72c9-4bec-a81b-3a03fe12e42a
16:24:38,348	rid:	worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0	c.a.s.k.c.l.worker.Worker	- Initializing LeaseCoordinator
16:24:38,494	rid:	worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0	c.a.s.k.c.l.worker.Worker	- Syncing Kinesis shard info
16:24:38,758	rid:	worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0	c.a.s.k.c.l.worker.Worker	- Starting LeaseCoordinator
16:24:38,813	rid:	LeaseCoordinator-1	c.a.s.k.l.impl.LeaseTaker	- Worker 97998ed6-183e-4225-9a76-dc1a31e081b6 needed 2 leases but none were expired, so it will steal lease shardId-000000000002 from 26a0eef5-c51e-46e7-92aa-4e547f14ec8f
16:24:38,827	rid:	LeaseCoordinator-1	c.a.s.k.l.impl.LeaseTaker	- Worker 97998ed6-183e-4225-9a76-dc1a31e081b6 saw 5 total leases, 0 available leases, 3 workers. Target is 2 leases, I have 0 leases, I will take 1 leases
16:24:38,861	rid:	LeaseCoordinator-1	c.a.s.k.l.impl.LeaseTaker	- Worker 97998ed6-183e-4225-9a76-dc1a31e081b6 successfully took 1 leases: shardId-000000000002
16:25:08,789	rid:	worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0	c.a.s.k.c.l.worker.Worker	- Initialization complete. Starting worker loop.
16:25:08,797	rid:	worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0	c.a.s.k.c.l.worker.Worker	- Created new shardConsumer for : ShardInfo [shardId=shardId-000000000002, concurrencyToken=4924acea-cea5-4d04-a9fe-a621b84a805b, parentShardIds=[], checkpoint={SequenceNumber: 49570092336489771390945783788019523776720431277106266146,SubsequenceNumber: 0}]
16:25:08,799	rid:	RecordProcessor-0000	c.a.s.k.c.l.w.BlockOnParentShardTask	- No need to block on parents [] of shard shardId-000000000002
16:25:09,810	rid:	RecordProcessor-0000	c.a.s.k.c.l.w.KinesisDataFetcher	- Initializing shard shardId-000000000002 with 49570092336489771390945783788019523776720431277106266146
16:25:38,823	rid:	worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0	c.a.s.k.c.l.worker.Worker	- Current stream shard assignments: shardId-000000000002
16:25:38,823	rid:	worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0	c.a.s.k.c.l.worker.Worker	- Sleeping ...
16:25:38,921	rid:	LeaseCoordinator-1	c.a.s.k.l.impl.LeaseTaker	- Worker 97998ed6-183e-4225-9a76-dc1a31e081b6 saw 5 total leases, 3 available leases, 2 workers. Target is 3 leases, I have 1 leases, I will take 2 leases
16:25:38,934	rid:	LeaseCoordinator-1	c.a.s.k.l.impl.LeaseTaker	- Worker 97998ed6-183e-4225-9a76-dc1a31e081b6 successfully took 2 leases: shardId-000000000004, shardId-000000000003
16:25:39,823	rid:	worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0	c.a.s.k.c.l.worker.Worker	- Created new shardConsumer for : ShardInfo [shardId=shardId-000000000004, concurrencyToken=e8c7044e-3033-4106-bc28-0f57ee24cab9, parentShardIds=[], checkpoint={SequenceNumber: 49569843781348688138435101669211274695657945615095038018,SubsequenceNumber: 0}]
16:25:39,824	rid:	worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0	c.a.s.k.c.l.worker.Worker	- Created new shardConsumer for : ShardInfo [shardId=shardId-000000000003, concurrencyToken=796491bf-2e70-462f-9051-9b9aff89ff06, parentShardIds=[], checkpoint={SequenceNumber: 49567923954416924365771252318481282427997188154562969650,SubsequenceNumber: 0}]
16:25:39,825	rid:	RecordProcessor-0001	c.a.s.k.c.l.w.BlockOnParentShardTask	- No need to block on parents [] of shard shardId-000000000004
16:24:38,328	rid:	WatchdogSystem-akka.actor.default-dispatcher-3	c.a.s.k.l.i.LeaseCoordinator	- With failover time 30000 ms and epsilon 25 ms, LeaseCoordinator will renew leases every 9975 ms, takeleases every 60050 ms, process maximum of 2147483647 leases and steal 1 lease(s) at a time.
16:25:39,825	rid:	RecordProcessor-0002	c.a.s.k.c.l.w.BlockOnParentShardTask	- No need to block on parents [] of shard shardId-000000000003
16:25:40,842	rid:	RecordProcessor-0000	c.a.s.k.c.l.w.KinesisDataFetcher	- Initializing shard shardId-000000000004 with 49569843781348688138435101669211274695657945615095038018
16:25:40,874	rid:	RecordProcessor-0002	c.a.s.k.c.l.w.KinesisDataFetcher	- Initializing shard shardId-000000000003 with 49567923954416924365771252318481282427997188154562969650

And what looks strange in this listing is that line: c.a.s.k.l.impl.LeaseTaker - Worker 97998ed6-183e-4225-9a76-dc1a31e081b6 saw 5 total leases, 0 available leases, 3 workers. Target is 2 leases, I have 0 leases, I will take 1 leases after restart of second machine, as if there weren't available leases at all, though at that moment one lease should be freed by restarting machine and three should be free long time ago.

after that system settled to the next assignment (first machine still processed one shard, second machine with new worker processed three shard, shard number 1 was abandoned):

16:26:39,844	rid:	worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0	c.a.s.k.c.l.worker.Worker	- Current stream shard assignments: shardId-000000000004, shardId-000000000003, shardId-000000000002
16:27:12,794	rid:	worker-Some(1209e181-10b3-4302-b702-b3df6598b3db)-run-1b0c931a-ab2b-427a-b9fe-bb34cbab1304-0	c.a.s.k.c.l.worker.Worker	- Current stream shard assignments: shardId-000000000000

Then we decided to restart the first machine.
The second machine this time took all leases. And only at this moment all shards were being processed.

17:08:42,021	rid:	LeaseCoordinator-1	c.a.s.k.l.impl.LeaseTaker	- Worker 97998ed6-183e-4225-9a76-dc1a31e081b6 saw 5 total leases, 2 available leases, 1 workers. Target is 5 leases, I have 3 leases, I will take 2 leases
17:08:42,035	rid:	LeaseCoordinator-1	c.a.s.k.l.impl.LeaseTaker	- Worker 97998ed6-183e-4225-9a76-dc1a31e081b6 successfully took 2 leases: shardId-000000000001, shardId-000000000000
17:08:42,103	rid:	worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0	c.a.s.k.c.l.worker.Worker	- Created new shardConsumer for : ShardInfo [shardId=shardId-000000000001, concurrencyToken=5c1f84f6-cdff-4a20-9349-a0984c44dceb, parentShardIds=[], checkpoint={SequenceNumber: 49562821942073005333784227425598100663899624075432034322,SubsequenceNumber: 0}]
17:08:42,105	rid:	worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0	c.a.s.k.c.l.worker.Worker	- Created new shardConsumer for : ShardInfo [shardId=shardId-000000000000, concurrencyToken=48b6c208-7e33-4f75-84c8-4c98fc2dc977, parentShardIds=[], checkpoint={SequenceNumber: 49566429842600596358334071735832074185052611066978107394,SubsequenceNumber: 0}]
17:08:42,105	rid:	RecordProcessor-0003	c.a.s.k.c.l.w.BlockOnParentShardTask	- No need to block on parents [] of shard shardId-000000000001
17:08:42,125	rid:	RecordProcessor-0004	c.a.s.k.c.l.w.BlockOnParentShardTask	- No need to block on parents [] of shard shardId-000000000000
17:08:43,110	rid:	RecordProcessor-0003	c.a.s.k.c.l.w.KinesisDataFetcher	- Initializing shard shardId-000000000000 with 49566429842600596358334071735832074185052611066978107394
17:08:43,111	rid:	RecordProcessor-0004	c.a.s.k.c.l.w.KinesisDataFetcher	- Initializing shard shardId-000000000001 with 49562821942073005333784227425598100663899624075432034322
17:09:36,174	rid:	worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0	c.a.s.k.c.l.worker.Worker	- Current stream shard assignments: shardId-000000000004, shardId-000000000003, shardId-000000000002, shardId-000000000001, shardId-000000000000

Then first machine started and get lease for shard number 4:

[INFO] [02/14/2017 17:09:52.374] [worker-Some(f4980ea0-8c17-4886-a0d9-3c98df845b12)-run-307e0cf9-c633-45e8-b3f4-19d9a0aa1f2d-0] [akka://WatchdogSystem/user/$a/$a] Starting the worker f4980ea0-8c17-4886-a0d9-3c98df845b12
17:09:52,343	rid:	WatchdogSystem-akka.actor.default-dispatcher-2	c.a.s.k.l.i.LeaseCoordinator	- With failover time 30000 ms and epsilon 25 ms, LeaseCoordinator will renew leases every 9975 ms, takeleases every 60050 ms, process maximum of 2147483647 leases and steal 1 lease(s) at a time.
17:09:52,375	rid:	worker-Some(f4980ea0-8c17-4886-a0d9-3c98df845b12)-run-307e0cf9-c633-45e8-b3f4-19d9a0aa1f2d-0	c.a.s.k.c.l.worker.Worker	- Initialization attempt 1
17:09:52,375	rid:	worker-Some(f4980ea0-8c17-4886-a0d9-3c98df845b12)-run-307e0cf9-c633-45e8-b3f4-19d9a0aa1f2d-0	c.a.s.k.c.l.worker.Worker	- Initializing LeaseCoordinator
17:09:52,500	rid:	worker-Some(f4980ea0-8c17-4886-a0d9-3c98df845b12)-run-307e0cf9-c633-45e8-b3f4-19d9a0aa1f2d-0	c.a.s.k.c.l.worker.Worker	- Syncing Kinesis shard info
17:09:52,782	rid:	worker-Some(f4980ea0-8c17-4886-a0d9-3c98df845b12)-run-307e0cf9-c633-45e8-b3f4-19d9a0aa1f2d-0	c.a.s.k.c.l.worker.Worker	- Starting LeaseCoordinator
17:09:52,818	rid:	LeaseCoordinator-1	c.a.s.k.l.impl.LeaseTaker	- Worker e2aa8983-e49f-4a6c-b314-07fd65b5a366 needed 3 leases but none were expired, so it will steal lease shardId-000000000004 from 97998ed6-183e-4225-9a76-dc1a31e081b6
17:09:52,818	rid:	LeaseCoordinator-1	c.a.s.k.l.impl.LeaseTaker	- Worker e2aa8983-e49f-4a6c-b314-07fd65b5a366 saw 5 total leases, 0 available leases, 2 workers. Target is 3 leases, I have 0 leases, I will take 1 leases
17:09:52,840	rid:	LeaseCoordinator-1	c.a.s.k.l.impl.LeaseTaker	- Worker e2aa8983-e49f-4a6c-b314-07fd65b5a366 successfully took 1 leases: shardId-000000000004
17:10:01,997	rid:	LeaseRenewer-522	c.a.s.k.l.i.LeaseRenewer	- Worker 97998ed6-183e-4225-9a76-dc1a31e081b6 lost lease with key shardId-000000000004
[WARN] [02/14/2017 17:10:10.070] [WatchdogSystem-akka.actor.default-dispatcher-2] [akka://WatchdogSystem/user/$a] The worker f4980ea0-8c17-4886-a0d9-3c98df845b12 is stopping
[WARN] [02/14/2017 17:10:10.065] [WatchdogSystem-akka.actor.default-dispatcher-2] [akka://WatchdogSystem/user/$a] Request shutdown of worker Some(f4980ea0-8c17-4886-a0d9-3c98df845b12)
17:10:22,800	rid:	worker-Some(f4980ea0-8c17-4886-a0d9-3c98df845b12)-run-307e0cf9-c633-45e8-b3f4-19d9a0aa1f2d-0	c.a.s.k.c.l.worker.Worker	- Initialization complete. Starting worker loop.
17:10:22,806	rid:	worker-Some(f4980ea0-8c17-4886-a0d9-3c98df845b12)-run-307e0cf9-c633-45e8-b3f4-19d9a0aa1f2d-0	c.a.s.k.c.l.worker.Worker	- Created new shardConsumer for : ShardInfo [shardId=shardId-000000000004, concurrencyToken=f94d17ce-b72a-42d2-b032-65595c52d8c7, parentShardIds=[], checkpoint={SequenceNumber: 49569843781348688138435892750655967103509458816235208770,SubsequenceNumber: 0}]
[WARN] [02/14/2017 17:10:23.074] [WatchdogSystem-akka.actor.default-dispatcher-5] [akka://WatchdogSystem/user/$a] The worker f4980ea0-8c17-4886-a0d9-3c98df845b12 is stopping
[WARN] [02/14/2017 17:10:23.073] [WatchdogSystem-akka.actor.default-dispatcher-5] [akka://WatchdogSystem/user/$a] Request shutdown of worker Some(f4980ea0-8c17-4886-a0d9-3c98df845b12)
17:10:22,808	rid:	RecordProcessor-0000	c.a.s.k.c.l.w.BlockOnParentShardTask	- No need to block on parents [] of shard shardId-000000000004
[WARN] [02/14/2017 17:10:24.074] [WatchdogSystem-akka.actor.default-dispatcher-8] [akka://WatchdogSystem/user/$a] Request shutdown of worker Some(f4980ea0-8c17-4886-a0d9-3c98df845b12)
17:10:23,818	rid:	RecordProcessor-0000	c.a.s.k.c.l.w.KinesisDataFetcher	- Initializing shard shardId-000000000004 with 49569843781348688138435892750655967103509458816235208770
17:10:37,992	rid:	worker-Some(f5ecf23e-72c9-4bec-a81b-3a03fe12e42a)-run-458a2eac-2f41-4ac3-b4b8-aedf9bfdbf26-0	c.a.s.k.c.l.worker.Worker	- Current stream shard assignments: shardId-000000000003, shardId-000000000002, shardId-000000000001, shardId-000000000000

[DominicMCN]

@tyger we're experiencing a similar issue, that is no other worker tried stealing the lease when one instance went down (due to auto-scaling). Have you found out the reason behind this?

[jonchase]

We're also experiencing this issue. We have restart all of our consumers periodically to ensure we catch any of these "dropped shards".

[david-mcneil]

We just recently started experiencing this behavior as well.

[sahilpalvia]

@tyger this is what I gather from your comments and logs, it seems like your workers were restarted. The KCL on restarts does not get the leases that it previously had, it tries to steal leases. The leases don't tell you the state of the worker if they are alive or not, they have an expiration after which they can be stolen. When machine 2 restarted, the leases that it previously had may not have expired/timed out yet. That is why in the log message you see 0 leases available. During the reboot of the first machine all the leases expired and machine 2 was able to acquire them. When machine 1 came back up, it saw that there are 5 leases present and machine 2 had it. To load balance it, it would have tried to steal 2 leases. If you could provide some more logs, I would be able to give you a more concrete answer.