fix(plugins): add opt-in exactly-once delivery to BigQueryAgentAnalyticsPlugin#6466
Draft
addenergyx wants to merge 6 commits into
Draft
fix(plugins): add opt-in exactly-once delivery to BigQueryAgentAnalyticsPlugin#6466addenergyx wants to merge 6 commits into
addenergyx wants to merge 6 commits into
Conversation
…icsPlugin BatchProcessor retries an ambiguous write (timeout, UNAVAILABLE, INTERNAL) by resubmitting the same batch to BigQuery's `_default` stream, which is at-least-once and rejects an append offset -- so a retry after a lost ack can write a byte-identical duplicate row. Add an opt-in `exactly_once_delivery` config flag that instead writes to a dedicated COMMITTED stream per event loop with a tracked append offset, so a retry resends the same offset and BigQuery rejects the duplicate (ALREADY_EXISTS) instead of applying it twice. Default is unchanged. Fixes google#6465
…ests - Drop the redundant offset_for_batch local in BatchProcessor._write_rows_with_retry; self._next_offset is only ever mutated on a confirmed outcome within a single (serial) call, so reading it directly is equivalent and removes four "is not None" checks. - Move the OUT_OF_RANGE branch above the generic error-code log so it no longer double-logs (it previously logged both the generic warning and its own error message for the same event). - Extract the repeated table-resource-path f-string into _table_resource_path(). - Hoist _stub_arrow_prep/_make_batch_processor to module level in the test file; TestDropStats and TestExactlyOnceDelivery previously carried byte-identical/near-identical copies. No behavior change; full suite still green.
adk-bot
added
the
tracing
… in BQ exactly-once path An external correctness review (Codex) surfaced two real bugs in the exactly_once_delivery path introduced by the prior commits: 1. The offset_for_batch removal in the last refactor made self._next_offset accumulate via `+=` instead of being assigned from a frozen base. If a batch's offset got resolved twice within one write (a confirmed success followed by a retry that then sees ALREADY_EXISTS for the same batch), the offset advanced twice, permanently skipping a range the server never actually used and wedging the processor on the next real batch. Restored offset_for_batch as an immutable per-batch value and switched both mutation sites back to assignment. Also return immediately after the single response AppendRows is expected to produce per request, instead of waiting on the response iterator for a stream-close that could time out after a success was already recorded. 2. If every retry attempt for an ambiguous failure (timeout, UNAVAILABLE, INTERNAL) also fails, the batch's fate is unknown -- it may have landed server-side despite every ack being lost. The previous code left _next_offset untouched in that case, so a later, unrelated batch could reuse the same offset, get ALREADY_EXISTS from the server, and be mistaken for its own successful delivery -- silently losing that batch's rows with nothing counted as dropped. Added an _offset_desynced flag, set on both this case and the existing OUT_OF_RANGE case, that makes the processor refuse further writes (loud, counted under the existing offset_conflict stat) until it is recreated with a fresh stream. Also trims several comments added in the prior two commits down to match the file's existing terse style. Adds 3 regression tests covering: offset correctness for a batch sent after a dedup, retry-exhaustion-after-ambiguous-failure poisoning the processor, and OUT_OF_RANGE poisoning future batches. Full suite: 364 passed.
…finalize streams - Poison the offset only on genuinely ambiguous failures: post-send ack-lost errors and zero-response streams. In-band rejections and pre-send client failures leave the offset valid so later batches keep writing. - Poison on unexpected mid-RPC exceptions (previously unguarded silent-loss path). - Log poisoning once at error level on state entry; per-batch drops log at debug. - Finalize the dedicated COMMITTED stream on shutdown/close, best-effort, once, within the caller's remaining timeout budget. - Consolidate offset bookkeeping into _confirm_delivered/_mark_offset_ambiguous, dedupe test fixtures, and add coverage for the new failure classifications. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
… COMMITTED stream Instead of permanently dropping every batch after an ambiguous failure, the processor now lazily rotates on the next write: it finalizes the desynced stream best-effort, creates a fresh COMMITTED stream via a plugin-supplied factory, resets the offset, and resumes exactly-once writes. Failed rotation drops that batch and backs off 30s before retrying so CreateWriteStream is not hammered during outages. Without a factory (direct construction), the previous fail-stop behavior is unchanged. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
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
3 participants
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.
Closes: #6465
BatchProcessor._write_rows_with_retrywrites to BigQuery's shared_defaultStorage Write API stream, which is at-least-once and rejects an append offset. It resubmits the same batch after a client-side timeout orUNAVAILABLE/INTERNAL, so if the first append actually landed but the ack was lost, the retry can write a byte-identical duplicate row. Evidence and repro in #6465.Adds an opt-in
exactly_once_deliveryflag toBigQueryLoggerConfig(defaultFalse, no behavior change for existing users). When enabled, each event loop'sBatchProcessorwrites to its ownCOMMITTEDstream with a tracked append offset instead of_default. A retry after an ambiguous failure resends the same offset; BigQuery returnsALREADY_EXISTSinstead of writing it again, which is treated as a confirmed no-op rather than an error. An unexpectedOUT_OF_RANGE(offset desync) drops the batch loudly rather than guessing a corrective offset. Similar in spirit to Apache Beam'suse_at_least_once=FalseonWriteToBigQuery.Added
TestExactlyOnceDelivery(5 tests) covering the offset lifecycle, including the exact timeout-then-ALREADY_EXISTSsequence from production. Full suite (tox, py310-py314): 9231 passed, 0 failed on every version. Not yet tested against a live BigQuery project.