Repository navigation
Conversation
Leward
added this pull request to stack #4990
October 19, 2026 19:31
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
Leward
changed the title
Close waited for the acks of all batches that the shard readers sent to the message channel. Batches that stay in the channel are never delivered, so their acks never come, and Close waited the full shutdown timeout (10s). Benthos resolves the acks of delivered batches before it calls Close, thus only the undelivered batches caused the wait. Close now removes the batches that stay in the channel and calls their ack functions after it marks the input as closed. These calls release the tracking and the pending ack. They do not write a checkpoint. The integration tests now ack the batches that they read, as a pipeline does. The ack-gate test still does not ack, because it simulates a crash.
The function name and its godoc give the reason for the drain and its precondition (d.closed is set). Close does not need a labeled loop.
The drain in Close did not fix the real shutdown. Benthos drops a batch that it read at shutdown without its ack when the batch did not get into the pipeline. Close waited the full shutdown timeout (10s) for that ack. Close now waits only for the ack functions that are running. Each ack function holds a read lock while it checks closed and does its work, and Close sets closed under the write lock. The pending acks wait group and the drain are removed. Acks after close were already dropped without a checkpoint, so the wait did not save a position. A new integration test stops a real stream with a slow output while the input has batches in its buffer. The stop took 10.1s before this change and takes about 2s now. A restart reads all records that the first run did not ack.
Only a restart that resumes from a checkpoint reads them again. Without a saved checkpoint, start_from: latest skips them.
The Benthos auto-retry wrapper calls ReadBatch from its own goroutine, so ReadBatch can run while Close runs. ReadBatch reads msgChan and shutSig without a lock, and Close set them to nil. The new shutdown integration test found this data race under -race. Close now keeps both fields. After Close, the soft stop signal makes ReadBatch return ErrNotConnected.
Use testcontainers.CleanupContainer, report stream.Run errors with t.Error, and log the Given, When, and Then steps.
Leward
force-pushed
the
dynamodb-cdc-close-soft-stop
branch
from
October 20, 2026 18:45
5100b1c to
7a0c432
Compare
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stacked on #4988.
Summary
Stopping a pipeline with an
aws_dynamodb_cdcinput in streaming mode took about 10 seconds too long.On close, the input waited for an ack of every batch it had read. Some of these acks never come:
Close now waits only for the acks that are running.
Records that were not acked are not checkpointed, so a restart from a checkpoint reads them again, as before.
This also affects production: the wait used half of the default 20s
shutdown_timeout, before the final checkpoints are saved.Close also no longer sets
msgChanandshutSigto nil. This fixes a data race with the Benthos auto-retry wrapper, which can callReadBatchwhileCloseruns.Evidence
New
TestIntegrationDynamoDBShutdownWithBufferedBatches: a real stream with a slow output stops after the first message, then restarts.After: pass. The restart reads all 100 items.
Package integration wall time, on top of #4988: 69.5s before, 10.0s after.
Merge Danger
Door: two-way
Blast Radius: shutdown
Acks after close were already dropped without a checkpoint, so the removed wait never saved a position.
Ref: CON-565