Reading from Kinesis stream with more than 1 shard in Spark Streaming

Question

Getting following error:

com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException: Can't update checkpoint - instance doesn't hold the lease for this shard

What could be the cause of this?

Answer 1

It seems like and kinesis issue. For debugging reason, you can enable logging for LeaseManager. There are some additional messages that should indicate the state of the checkpoint request. An important thing to take into consideration is failoverTimeMillis. You could try to increase the value of failoverTimeMillis this could help to avoid the following issue. More detailed explanation is in this GitHub issue.

The root cause of this issue is that sometimes a KCL Worker loses a lease and then reacquires a new lease on a particular shard before discarding the old lease for that shard, it then continues to hold the old lease which still holds the old concurrency token. The mismatched concurrency token prevents checkpointing but the Worker never discards the lease because the lease cleanup logic only compares the shard-id.

one way to alleviate the impact of this issue is to increase the failoverTimeMillis which is set to 1000ms by default in the connector library. Increasing the failoverTimeMillis will reduce the chance of a Worker losing and then immediately reacquiring its own lease.