BigQueryAgentAnalyticsPlugin can duplicate rows when writes are retried

Problem

BigQueryAgentAnalyticsPlugin can write duplicate rows when its internal BigQuery Storage Write API request is retried after an ambiguous failure.

This concerns retries of the BigQuery write, not retries by the agent, model, or tools.

The plugin writes to BigQuery's _default stream. BatchProcessor._write_rows_with_retry constructs one AppendRowsRequest and resends the same serialized batch following transient errors or its 30-second
client-side timeout.

If BigQuery accepts the first append but its acknowledgement is lost or delayed, the plugin retries the batch. Because _default provides at-least-once delivery, BigQuery can accept it again and produce duplicate rows.

Relevant code:

Impact

We observed byte-identical duplicate analytics rows in production, including identical timestamps, trace/span IDs, token usage, status, and content.

The issue is rare. Over seven days, six duplicate LLM_RESPONSE rows inflated a naive SUM(usage.total) by 182,555 tokens, approximately 0.31% of the total. This can silently over-count tokens, costs, latency, and other metrics derived from the analytics table. However this was for a poc agent with few users and can easily inflate with more use.

Proposed fix

BigQuery does not allow append offsets on _default, and BigQueryWriteAsyncClient has no exactly_once option.

Exactly-once append behaviour requires an application-created COMMITTED stream with offsets:

Create one committed stream per BatchProcessor.
Include the current offset in each AppendRowsRequest.
Retry an ambiguous append using the same stream and offset.
Advance the offset after success.
Treat ALREADY_EXISTS at that offset as confirmation that the earlier
attempt succeeded.
Handle unexpected OUT_OF_RANGE without blindly resending on a new stream.

A committed stream preserves immediate row visibility while making Storage Write API retries idempotent.

This fix would prevent duplicates caused by BigQuery write retries. It would not deduplicate genuinely separate events emitted by the agent or application.

For anyone familiar with Apache beam this is similar at a high level to their use_at_least_once=False option for BigQuery Storage Write API sinks. Beam provides the stream, offset, and durable checkpoint management behind that abstraction; because ADK uses BigQueryWriteAsyncClient directly, the plugin must manage its committed stream and append offsets itself.

References:

@addenergyx

Hi @addenergyx , Thank you for providing these details. To help us fully understand the issue, could you please share a minimal reproducible example? This will allow us to recreate the problem exactly as you are experiencing it.

@sanketpatil06

Hi @sanketpatil06, thanks for looking into this. Sorry for the delay, took me abit to work out a way to reliably reproduce this issue. From my experience with our poc agent work, if you'd want to reproduce this with an actual agent you'd need a loop of it doing actions to create logs and hopefully eventually get a lost ack to cause a duplicate.

The race needs an append that BigQuery applies but whose acknowledgement the client never sees. Rather than wait for that, the harness injects it, forward the append to BigQuery, consume the real success response so the rows are genuinely committed, then raise ServiceUnavailable instead of returning it.

class AckSuppressingClient:
    """Real BigQuery, but the first successful ack never reaches the caller."""

    def __init__(self, real):
        self.real = real
        self.suppress = True
        self.attempts = 0

    async def append_rows(self, requests, **kw):
        self.attempts += 1
        responses = await self.real.append_rows(requests, **kw)
        collected = [r async for r in responses]   # BigQuery has now committed
        if self.suppress:
            self.suppress = False
            raise ServiceUnavailable("ack suppressed by test harness")

        async def replay():
            for r in collected:
                yield r

        return replay()

Then drive BatchProcessor with it, append one row tagged with a unique id, and
count:

bp = BatchProcessor(
    write_client=AckSuppressingClient(BigQueryWriteAsyncClient()),
    arrow_schema=SCHEMA,
    write_stream=f"projects/{project}/datasets/{dataset}/tables/{table}/_default",
    batch_size=1, flush_interval=0.05,
    retry_config=RetryConfig(max_retries=3, initial_delay=0.5),
    queue_max_size=10, shutdown_timeout=30.0,
)
await bp.start()
await bp.append({"id": run_id, "tokens": 1234})   # ONE event
await bp.flush()
await bp.shutdown(timeout=30.0)
# SELECT COUNT(*) FROM table WHERE id = run_id

run	appends sent	rows in table
ack suppressed	2	2
control, ack delivered	1	1
ack suppressed (repeat)	2	2

_write_rows_with_retry builds one AppendRowsRequest before the retry loop and resends that same object, and on _default it carries no offset — so the client can't tell "never applied" from "applied, ack lost", and BigQuery has nothing to deduplicate against, since _default is documented as at-least-once (https://docs.cloud.google.com/bigquery/docs/write-api-streaming) and permits duplicate destination rows.

i've started a fix in here #6466

I'm working on this issue for now, would keep you posted

Confirmed at HEAD. _write_rows_with_retry builds the AppendRowsRequest once and resends the identical batch on transient errors and on the 30s client-side timeout, against _default with no offset — so an applied-but-unacked append duplicates rows exactly as your harness demonstrates. Thanks for the reliable repro; the ack-suppression wrapper is a faithful model of the race.

Plan, in two parts:

1. Near-term (unconditional): per-row event_id dedup key. Every row gets a UUID stamped before enqueue, so Storage Write API retry resends carry the identical id. That makes transport-level duplicates exactly identifiable (QUALIFY ROW_NUMBER() OVER (PARTITION BY event_id ...) = 1, or an opt-in deduplicating view). Existing tables pick the column up via the additive schema-upgrade path. This doesn't prevent duplicates, but it makes your token/cost SUMs correct immediately and at near-zero risk.

2. Your PR #6466 as the opt-in exactly-once mode. The offset/stream design is right, and the ambiguity classification (request_sent/in-band-rejection) and ALREADY_EXISTS-as-confirmation handling are careful. Before we can merge it:

The cla/google check is failing — that's a hard blocker, please complete the CLA.
NOT_FOUND isn't handled: if the committed stream is reclaimed/finalized externally, batches drop forever with no rotation (rotation currently only triggers on OUT_OF_RANGE desync). Suggest treating NOT_FOUND as desync-equivalent → rotate.
Real gapic clients can surface AlreadyExists/NotFound as raised google.api_core exceptions from the response iterator, not only as in-band response.error codes; the current handling would poison instead of confirm on the raised form. Since you noted it hasn't run against live BigQuery yet, a live smoke test of the error-surfacing forms would settle this.
Please document the CreateWriteStream quota exposure (relevant for high-churn serverless fleets) and that during a desync window the mode degrades to at-most-once (batches drop while rotating).

Happy to review the updated PR — with those addressed I think this is mergeable as a default-off feature.

Follow-up after running #6466 locally:

TestExactlyOnceDelivery: 17 passed.
Full BigQuery Agent Analytics plugin test file: 453 passed, 6 skipped.

That is a solid unit-test baseline, but it does not yet validate the Storage Write API’s live committed-stream behavior. Before merging, I recommend these gates:

Add a live opt-in smoke test using an actual COMMITTED stream and an ack-suppressing client: retry the same frozen batch/offset, verify exactly one copy is stored, then verify the next batch is accepted at the next offset.
Handle both in-band AppendRowsResponse.error statuses and raised google.api_core exceptions. In particular, AlreadyExists for the same offset should be treated as delivered; NotFound/invalidated stream and OutOfRange need explicit resynchronization or stream rotation rather than indefinite retry/drop behavior.
Finalize the committed stream on every shutdown path, including early returns after a worker becomes terminal.
Keep the feature default-off and document CreateWriteStream quota usage, ordering assumptions, and that this provides idempotent append retries—not durable end-to-end exactly-once processing across process loss.

I would also split the proposed per-row event_id mitigation into a separate small change. It is useful for query-time deduplication, but it does not prevent duplicate writes and it introduces its own physical-schema/view contract. It should not be coupled to #6466’s transport semantics.

The remaining administrative blocker on #6466 is the failing CLA check.

A few lifecycle notes to add to the #6466 review checklist (from tracing the PR against current HEAD):

Finalize flag is set before the attempt. _stream_finalized flips ahead of the finalize_write_stream call, so a failed finalize is never retried. Impact is low (COMMITTED rows stay visible without finalization), but it deserves a code comment so nobody "fixes" it into a retry loop later.
Orphaned streams rely on idle expiry. Each event loop creates its own COMMITTED stream; abandoned ones are only reclaimed by BigQuery's ~24h idle-stream expiry. Worth stating next to the CreateWriteStream quota note already requested — high-churn serverless fleets (many short-lived loops) will accumulate streams against quota between expiries.
First create_write_stream usage in the repo. Nothing else in the codebase exercises stream creation/finalization, so the mock-based unit tests are the only coverage until a live smoke test confirms the error-surfacing forms (raised gapic exceptions vs in-band response.error codes) called out above.
Test anchor for the ack-lost sequence: the shared mock_write_client fixture (tests/unittests/plugins/test_bigquery_agent_analytics_plugin.py:140-156) already models append_rows as an async generator; a side-effect sequence of asyncio.TimeoutError then an in-band code-6 response reproduces the production race deterministically — the PR's TestExactlyOnceDelivery covers this, and the same pattern should guard the rotation path.

None of these change the design; they're the merge-gate details alongside the CLA and the NOT_FOUND-as-desync handling already flagged.

@addenergyx

Implementation follow-up: the independent implementation is now in draft PR caohy1988#12.

The draft keeps both promised layers:

unconditional per-row event_id, stamped before enqueue so a transport retry preserves the ID; it is included in the physical schema, additive upgrade, and views;
default-off exactly_once_delivery, using loop-local COMMITTED streams and frozen batch offsets.

The offset path handles both raised and in-band ALREADY_EXISTS as delivery confirmation, treats raised/in-band NOT_FOUND and OUT_OF_RANGE as desynchronization, rotates before subsequent batches, does not replay an uncertain exhausted batch onto a new stream, and retries failed finalization on later shutdown/close paths.

I also ran the live probe requested above against test-project-0728-467323: offset 0 succeeded with code 0; resending offset 0 surfaced as an in-band response error code 6, not a raised exception. The temporary dataset/table was finalized and deleted. Unit coverage retains both forms because client/library behavior may differ.

CLA handling: this implementation was written independently from the public issue design; it does not copy/cherry-pick #6466 and has no non-signer Co-authored-by trailer. The PR prose credits @addenergyx for the report and design discussion.

fix(plugins): Make BigQuery retry duplicates identifiable

feat(plugins): add offset-aware BigQuery delivery

fix(plugins): address BigQuery analytics review findings

@caohy1988

Thanks @caohy1988, and sorry for the slow reply. Really appreciate you picking this up! Happy to help review once it's ready

fix(plugins): address BigQuery analytics review findings

@addenergyx

Correction / final status: this is fixed at upstream HEAD. The change merged to main on 2026-08-10 in 04b8b72 (feat(plugins): add BigQuery Agent Analytics delivery and termination observability), landing via the internal review + Copybara flow. My earlier note describing it as still pending on a fork draft was out of date — apologies for the confusion.

What shipped (re-verified line-by-line at current HEAD), both layers:

Unconditional per-row event_id: assigned before enqueue, so a Storage Write API retry resends the identical id. It's in the physical schema (additive upgrade, schema v2) and all typed views — transport-level duplicates are now exactly identifiable (QUALIFY ROW_NUMBER() OVER (PARTITION BY event_id ...) = 1), which makes the token/cost SUMs from your report correct immediately in default mode.
Opt-in exactly_once_delivery (default off): one loop-local COMMITTED stream with explicit offsets. The batch offset is frozen before the retry loop; ambiguity is tracked per attempt, so ALREADY_EXISTS at the batch's offset confirms delivery only after an ambiguous send, while a first-attempt ALREADY_EXISTS desynchronizes and rotates. Both raised gapic exceptions and in-band response.error codes are handled for ALREADY_EXISTS / NOT_FOUND / OUT_OF_RANGE; ambiguous retry exhaustion poisons the stream rather than guessing; an uncertain batch is never replayed onto a new stream; rotation is non-blocking with a 30s backoff and an explicit offset_conflict drop bucket; finalization is retried across shutdown/close within the caller's timeout budget. Quota and loss boundaries (CreateWriteStream on rotation; duplicate-free but not lossless) are documented on the config.

This matches the committed-stream/offset design you proposed and reliably neutralizes the applied-but-unacked race your ack-suppression harness demonstrated — thanks for the report, the repro, and the design groundwork in #6466.

Availability: the merge postdates v2.6.3, so it ships in the next release; until then it's available from main.

Closing as fixed. @addenergyx if you get a chance to run your harness against main (with and without exactly_once_delivery=True), confirmation from the original environment would be a great final validation.

Navigation Menu

Repository navigation

BigQueryAgentAnalyticsPlugin can duplicate rows when writes are retried #6465

Problem

Impact

Proposed fix

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

BigQueryAgentAnalyticsPlugin can duplicate rows when writes are retried #6465

Description

Problem

Impact

Proposed fix

Activity

sanketpatil06 commented on Jul 28, 2026

addenergyx commented on Jul 29, 2026

haiyuan-eng-google commented on Aug 5, 2026

caohy1988 commented on Aug 5, 2026

caohy1988 commented on Aug 5, 2026

caohy1988 commented on Aug 5, 2026

caohy1988 commented on Aug 5, 2026

addenergyx commented on Aug 6, 2026

caohy1988 commented on Aug 11, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions