From e267297972e1fc8c36329dd2a3c71b5a9ef8d5b6 Mon Sep 17 00:00:00 2001 From: "pj.vauthier" Date: Mon, 31 Aug 2026 14:49:01 -0400 Subject: [PATCH] aws_dynamodb_batcher: extract `hasUnpersistedFrontier` for readability and reuse The same condition was repeated multiple times in the file. Replaced the expression with a readable wording `hasUnpersistedFrontier` --- internal/impl/aws/dynamodb/batcher.go | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/internal/impl/aws/dynamodb/batcher.go b/internal/impl/aws/dynamodb/batcher.go index 9f17b05fd4..1402d15c6a 100644 --- a/internal/impl/aws/dynamodb/batcher.go +++ b/internal/impl/aws/dynamodb/batcher.go @@ -407,6 +407,12 @@ func (st *shardAckTracker) advanceFrontier(frontier string) { } } +// hasUnpersistedFrontier reports whether the shard has a contiguous acked +// sequence frontier that has not yet been persisted. +func (st *shardAckTracker) hasUnpersistedFrontier() bool { + return st.frontier != "" && st.frontier != st.persisted +} + // AckBatch marks a tracked batch as acknowledged, advances the shard's // contiguous frontier, and persists a checkpoint once enough messages have // been acked since the last persisted position. Settlement goes through the @@ -475,7 +481,7 @@ func (b *RecordBatcher) maybePersist(ctx context.Context, cp checkpointer, st *s for { b.mu.Lock() - due := st.pending >= cp.CheckpointLimit() && st.frontier != "" && st.frontier != st.persisted + due := st.pending >= cp.CheckpointLimit() && st.hasUnpersistedFrontier() persistSeq := st.frontier persistTime := st.frontierTime pendingAtCompute := st.pending @@ -522,7 +528,7 @@ func (b *RecordBatcher) PendingCheckpoints() map[string]CheckpointValue { checkpoints := make(map[string]CheckpointValue, len(b.shards)) for shardID, st := range b.shards { - if st.frontier != "" && st.frontier != st.persisted { + if st.hasUnpersistedFrontier() { checkpoints[shardID] = CheckpointValue{ SequenceNumber: st.frontier, ApproxCreationTime: st.frontierTime, @@ -631,7 +637,7 @@ func (b *RecordBatcher) LastCheckpointsCount() int { defer b.mu.Unlock() n := 0 for _, st := range b.shards { - if st.frontier != "" && st.frontier != st.persisted { + if st.hasUnpersistedFrontier() { n++ } }