fix(kinesis): Fix shard end coordination race condition

[huajiang-tubi]

Summary

• Fix race condition in ShardProcessor where premature shard checkpoint occurs when KCL delivers an empty final batch
• Replace single semaphore coordination with per-batch last-record tracking
• Add ShardProcessorSpec with comprehensive test coverage for the coordination scenarios

Problem

The original implementation used a single Semaphore(1) that was acquired when isAtShardEnd was true. This created a race condition:

1. When KCL delivers a batch with records where isAtShardEnd=false, the records are processed normally
2. KCL then delivers an empty final batch with isAtShardEnd=true (this is valid KCL behavior)
3. The shardEnded method is called, but since no records were in the final batch, no semaphore coordination happened
4. shardEnded immediately checkpoints the shard with SHARD_END
5. In-flight records from the previous batch fail to checkpoint because the shard is already marked as ended

Solution

Replace the single semaphore approach with a per-batch tracking mechanism:

• Create a new Semaphore(0) for the last record of each batch (empty batches create no semaphore)
• Store it in Option[Semaphore] (volatile) so it persists across batches
• When shardEnded is called, wait on the most recent semaphore (which correctly points to the previous batch's last record if the final batch is empty)
• If no records were ever processed, proceed without waiting

Test Coverage

Added ShardProcessorSpec with three test scenarios:

1. Normal case: Block shardEnded until the last record is checkpointed
2. Empty final batch: Block shardEnded until the previous batch's last record is checkpointed (the critical bug fix)
3. No records: Complete shardEnded immediately when no records were ever processed

Related

This bug was discovered and fixed in https://github.com/adRise/hyades/pull/3668 and has been ported to alpakka.

🤖 Generated with Claude Code

[lightbend-cla-validator]

Hi @huajiang-tubi,

Thank you for your contribution! We really value the time you've taken to put this together.

Before we proceed with reviewing this pull request, please sign the Akka Contributors License Agreement:

https://www.lightbend.com/contribute/cla

[huajiang-tubi]

I have signed sign the CLA.

[sebastian-alfers]

@huajiang-tubi I have closed and re-opened this PR and now the CLA validator looks ok.

[sebastian-alfers]

This bug was discovered and fixed in https://github.com/adRise/hyades/pull/3668 and has been ported to alpakka.

Can you share more details around this? I am not able to open this change.

[huajiang-tubi]

This bug was discovered and fixed in adRise/hyades#3668 and has been ported to alpakka.

Can you share more details around this? I am not able to open this change.

We observed checkpoint failures during scaling up of a kinesis stream. It was caused by checkpoint with sequence number after SHARD_END had been checkpointed.

The current implementation of ShardProcessor is using a semaphore to prevent it. It is based on the assumption that we can get notified that the shard is ended via batchData.isAtShardEnd in processRecords. But it's not true when the shard ends with an empty batch. KCL skips processRecords in this case, and when shardEnded is called, the semaphore still has the initial permit (the last record was not known as the last one because the shard was still alive when it was processed). So SHARD_END is checkpointed immediately without waiting for the previous records to be checkpointed.

@sebastian-alfers

[johanandren]

LGTM

I pushed a tiny improvement to the test on top of the changes.