fix(bigtable): report swallowed batch flush errors and unacknowledged entries

[mutianf]

report swallowed batch flush errors and unacknowledged mutate_rows entries.

Double check if there's any active requests left to avoid incomplete mutations.

[gemini-code-assist]

Code Review

This pull request improves error handling in Google Cloud Bigtable by surfacing exceptions raised during asynchronous flushes in the batcher and ensuring that mutations without response entries in successfully closed streams are treated as incomplete rather than silently dropped. The review feedback points out a potential CancelledError when calling future.exception() on a cancelled future in _batch_completed_callback, suggesting a check for future.cancelled() to avoid unhandled exceptions.

[daniel-sanche]

LGTM

[parthea]

The unit tests seem to be running for more than an hour. Please can you check if it is related to the changes in this PR?

https://github.com/googleapis/google-cloud-python/actions/runs/31826948380/attempts/1?pr=18122

[parthea]

Adding do not merge to investigate the delay with the unit tests. In the last PR that was merged to packages/google-cloud-bigtable, tests only took 7 minutes

#18112

https://github.com/googleapis/google-cloud-python/actions/runs/31780116767

[parthea]

Something is wrong with tests. The duration of unit tests increased from ~ 7 minutes to > 1 hour

[daniel-sanche]

Hmm I wonder if the fork could be causing the issue? I'm looking into this

[mutianf]

The errors seems to be:

 raise exceptions.from_grpc_error(exc) from exc
E           google.api_core.exceptions.ResourceExhausted: 429 Quota exceeded for quota metric 'Instance and cluster metadata write requests' and limit 'Instance and cluster metadata write requests per day' of service 'bigtableadmin.googleapis.com' for consumer 'project_number:'. [reason: "RATE_LIMIT_EXCEEDED"

[mutianf]

I think my fix on the async client broke the test:

The async MutateRows operation retries by tracking remaining_indices — the set of entries still needing a response. At the start of each attempt it resets remaining_indices = [] and only re-adds an entry when the server explicitly returns a non-OK status for it. This means that if the stream closes cleanly (gRPC OK) but simply omits the response entry for a mutation, that entry is neither re-queued nor recorded as an error — it's silently treated as successful. Since the server sends one response entry per request entry and the client cannot know the true outcome of an omitted one, assuming success risks reporting a write as applied when it may not have been (silent data loss). The correct behavior is to treat an omitted entry as an unknown outcome: retry it if idempotent, or surface it as a failure otherwise.

The catch is that the existing V3 unit tests bake in the old assumption — their mock responses return an entry only for the mutations they care about and let the rest be "omitted = success" (e.g. _mock_response([DeadlineExceeded]) returns a single entry for a 3-entry request). So fixing the omitted-entry handling changes established behavior and breaks
those tests, which is what surfaced in CI.