Repository navigation
Conversation
… stream ended Add an in-process fake Pub/Sub server and two regression tests. The fake's first streaming pull delivers one message and is then closed by the server with UNAVAILABLE, as Pub/Sub does when it rebalances streams; later streams deliver nothing. - The acknowledgement test finalizes the event only after the source has opened its next stream and expects the acknowledgement to reach the server. - The shutdown test begins shutdown while that batch is pending and expects the source to wait for the batch, then to finish once the event is delivered.
Each streaming pull created its own finalizer and acknowledgement channel, and dropped both when the stream ended. Pub/Sub closes streaming pulls routinely (UNAVAILABLE while rebalancing), the source restarts them when its authentication token is refreshed, and any stream error ends them. Every batch still in the pipeline at that moment lost its acknowledgement, so its messages were redelivered once the ack deadline expired. Behind a slow sink this redelivered tens of thousands of already-delivered messages per stream restart. Pub/Sub accepts acknowledgement IDs on any stream of the subscription. Move the finalizer, the pending batch count and the acknowledgement queue from the streaming pull into its task, so they outlive the individual streams. Each new streaming pull takes over the queue, so acknowledgements for batches that are finalized after their stream ended, and updates the ended stream had not yet sent, are sent on the next stream. A new stream's request stream is sent before the server answers, which it does only once it has messages to deliver, so the startup wait now also forwards finalized batches and, like the main loop, waits for pending batches before shutting down. The finalizer stream ends after shutdown, and an ended stream is always ready, so poll it only while batches are pending; otherwise the biased select would spin on it instead of reaching its shutdown branch. When the source fails to emit a batch, release the leases of its messages with a zero ack deadline instead of holding them until they expire.
…ding_bytes The source opened streaming pulls without flow control, so the server sent messages as fast as it could and every one of them held a lease until the pipeline acknowledged it. Behind a sink slower than the subscription's fill rate, the source leased far more messages than it could deliver within the ack deadline, and those messages were redelivered when their leases expired. Add `max_outstanding_messages` and `max_outstanding_bytes`, sent on the first request of each streaming pull as the StreamingPullRequest flow control fields. Both default to 0, which the server treats as no limit, so existing configurations behave as before.
Contributor
|
Thank you for your contribution! Before we can merge this PR, please sign our Contributor License Agreement. To sign, copy and post the phrase below as a new comment on this PR.
I have read the CLA Document and I hereby sign the CLA You can retrigger this bot by commenting recheck in this Pull Request. Posted by the CLA Assistant Lite bot. |
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.
Summary
The
gcp_pubsubsource creates its batch finalizer and acknowledgement channel per streamingpull and drops both when the stream ends. Pub/Sub ends streaming pulls routinely (
UNAVAILABLEwhile it rebalances), the source restarts them when its token is refreshed, and any error ends
them. Every batch still in the pipeline at that moment loses its acknowledgement, and its
messages are redelivered once their ack deadline expires. The source also opens streams with
no flow control, so behind a slow sink it leases far more messages than it can deliver within
the ack deadline.
In a 1.2M-message run behind a 1,000 events/s sink (ack deadline 600 s), this produced 34,176
duplicate deliveries, all expiring together 600 s after a stream restart.
This PR:
of per stream. Pub/Sub accepts ack ids on any stream of the subscription, so acks for
batches finalized after their stream ended, and updates the ended stream had not sent, go
out on the next stream. A stream counter kept under the queue's lock ends the previous
request stream when a new one takes over the queue, so an ended stream cannot take updates
meant for its successor. The server answers a new streaming pull only once it has messages
to deliver, so the startup wait now forwards finalized batches too, and waits for pending
batches before shutting down, as the main loop already did;
an ended stream would always win the biased select;
modify_deadline_seconds = 0) of messages the source failed to emit;max_outstanding_messagesandmax_outstanding_bytes, sent in the firstStreamingPullRequest. Both default to 0 (no limit), so existing configurations are unchanged.The vendored proto has only the
StreamingPullRPC, so carrying acks to the next stream issmaller than adding a unary
Acknowledgecall.The change was developed and qualified against v0.54.0, where we run it, and ported to
masterfor this PR. The port kept the mechanism unchanged and adapted it to the pedanticclippy lints
masternow enables.References
None (no existing issue).
Vector configuration
How did you test this PR?
New unit tests run against an in-process tonic implementation of the vendored
Subscriberservice, whose first streaming pull delivers one message and then ends with
UNAVAILABLE:acknowledges_batches_finalized_after_their_stream_ended: the event is finalized only afterthe source opened its next stream, and the ack must reach the server;
shuts_down_after_its_pending_batches_are_finalized: shutdown begins while the batch ispending and the next stream is starting, and the source must wait for the batch, then finish;
requests_the_configured_flow_controlandflow_control_is_unlimited_by_default.On
master(7fd2b32), x86_64 Linux, Rust 1.96.1:timed out waiting for an acknowledgement;the source shut down with a batch pending);cargo test --no-default-features --features sources-gcp_pubsub --lib -- sources::gcp_pubsub::passes all 7 tests, and 10 more consecutive runs pass;cargo clippy --no-default-features --features sources-gcp_pubsub --lib --tests -- -D warningsand
cargo fmt --checkare clean;vdev check changelog-fragmentsaccepts both fragments;vector generate-schemaandvdev build component-docs(cue 0.17.1) produce the committedmax_outstanding_messagesandmax_outstanding_bytesentries inwebsite/cue/reference/components/sources/generated/gcp_pubsub.cue. That run used asingle-component build, so I did not run
make check-generated-docsover the full schema.On v0.54.0 the same tests showed the same red/green result, and the fake-server tests passed 20
more consecutive runs. The Pub/Sub emulator integration tests were not run (they are ignored
upstream, #24133).
Change type
Does this PR include user facing changes?
changelog.d/.no-changeloglabel to this PR.