fix: ensure closed shards with pending data are still claimed

[matus-tomlein]

Summary

Fixes #643

Closed shards (with EndingSequenceNumber) were being skipped during rebalancing, even if they still had unprocessed records. This caused data loss when a consumer restarted before fully draining a closed shard.

The Problem

When a shard is closed, the old logic would skip it entirely during rebalancing:

for _, s := range allShards {
    if !isShardFinished(s) {  // ← Closed shards skipped!
        unclaimedShards[*s.ShardId] = ""
    }
}

This caused a race condition:

1. Consumer is reading from Shard-0
2. Kinesis scales down → Shard-0 gets CLOSED (EndingSequenceNumber set)
3. Consumer hasn't finished draining Shard-0 yet (still has records)
4. Rebalance runs:
   • isShardFinished(Shard-0) = true
   • Shard-0 is NOT added to unclaimedShards
   • No consumer gets spawned for it
5. Consumer pod restarts (or crashes)
6. Shard-0 is orphaned — nobody will ever finish consuming it

The Fix

Now we also check if a closed shard has a checkpoint in DynamoDB. If a checkpoint exists, the shard hasn't been fully consumed yet (since checkpointer.Delete() removes it upon completion), so we include it in unclaimedShards:

if !isShardFinished(s) || shardsWithCheckpoints[shardID] {
    unclaimedShards[shardID] = ""
}

Test plan

Added unit tests for isShardFinished() covering open, closed, and edge cases

[Copilot]

Pull request overview

This pull request fixes a race condition where DynamoDB lease table entries for closed Kinesis shards are not cleaned up when Bento pods are terminated during Kubernetes scaling events, leading to orphaned entries and false positive latency alerts.

• Adds conditional cleanup logic during pod shutdown to delete checkpoints for finished shards instead of saving them
• Implements periodic background cleanup that removes DynamoDB entries for shards that have been closed by Kinesis

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 1 out of 1 changed files in this pull request and generated 4 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Copilot encountered an error and was unable to review this pull request. You can try again by re-requesting a review.

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated no new comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[matus-tomlein]

@jem-davies @gregfurman We have simplified the PR and addressed all the comments above – could you please give this another look?

[jem-davies]

Closed as included in #680