From db44971710291c27448e2e02ee0c8558c70a6b02 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Fri, 14 Aug 2026 17:45:03 +0000 Subject: [PATCH 1/3] fix(bigtable): report swallowed batch flush errors and unacknowledged mutate_rows entries Change-Id: I61a1444656d46f5b8e62aea0d16670dc9115659f --- .../google/cloud/bigtable/batcher.py | 13 +++++ .../bigtable/data/_async/_mutate_rows.py | 12 ++++ .../data/_sync_autogen/_mutate_rows.py | 7 +++ .../unit/data/_async/test__mutate_rows.py | 55 ++++++++++++++++++- .../data/_sync_autogen/test__mutate_rows.py | 49 ++++++++++++++++- .../tests/unit/v2_client/test_batcher.py | 30 ++++++++++ 6 files changed, 160 insertions(+), 6 deletions(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py index 4643c3402af4..32c6820c5978 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py @@ -350,6 +350,19 @@ def _batch_completed_callback(self, future): processed_rows = self.futures_mapping[future] self.flow_control.release(processed_rows) del self.futures_mapping[future] + # Surface any exception raised inside the async flush. Without this, an + # exception raised by ``_flush_rows`` (e.g. a non-retryable RPC error, a + # retry deadline, or a response-count mismatch) would be stored on the + # future and silently discarded, so the failed mutations would never be + # reported to the user -- effectively silent data loss. Per-row errors + # from a successful RPC are already recorded in ``self.exceptions`` by + # ``_flush_rows``; here the whole batch failed with a single exception, + # so record it once per row in the batch to keep the reported error + # count aligned with the number of affected mutations. + exc = future.exception() + if exc is not None: + for _ in range(processed_rows.rows_count): + self.exceptions.put(exc) def _row_fits_in_batch(self, row, batch_info): """Checks if a row can fit in the current batch. diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py index 974e450d232b..c54f3c97bfb3 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py @@ -214,6 +214,18 @@ async def _run_attempt(self): self._handle_entry_error(idx, exc) # bubble up exception to be handled by retry wrapper raise + # Any entries that were sent but never received a response entry (a + # successfully-closed but incomplete stream) must not be treated as + # successful. Record a retryable error so idempotent entries are retried + # and non-idempotent entries surface as failures instead of being + # silently dropped. + for idx in active_request_indices.values(): + self._handle_entry_error( + idx, + bt_exceptions._MutateRowsIncomplete( + "no response entry received for mutation" + ), + ) # check if attempt succeeded, or needs to be retried if self.remaining_indices: # unfinished work; raise exception to trigger retry diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py index 40e19dd85847..ac2c1d9edac3 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py @@ -173,6 +173,13 @@ def _run_attempt(self): for idx in active_request_indices.values(): self._handle_entry_error(idx, exc) raise + for idx in active_request_indices.values(): + self._handle_entry_error( + idx, + bt_exceptions._MutateRowsIncomplete( + "no response entry received for mutation" + ), + ) if self.remaining_indices: raise bt_exceptions._MutateRowsIncomplete diff --git a/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py b/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py index 8ff6e42532b4..5a17b2582e00 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py @@ -53,8 +53,13 @@ def _make_mutation(self, count=1, size=1): return mutation @CrossSync.convert - async def _mock_stream(self, mutation_list, error_dict): + async def _mock_stream(self, mutation_list, error_dict, omit_indices=None): + omit_indices = omit_indices or set() for idx, entry in enumerate(mutation_list): + if idx in omit_indices: + # simulate a server that closes the stream OK without returning + # a response entry for this mutation + continue code = error_dict.get(idx, 0) yield MutateRowsResponse( entries=[ @@ -64,12 +69,12 @@ async def _mock_stream(self, mutation_list, error_dict): ] ) - def _make_mock_gapic(self, mutation_list, error_dict=None): + def _make_mock_gapic(self, mutation_list, error_dict=None, omit_indices=None): mock_fn = CrossSync.Mock() if error_dict is None: error_dict = {} mock_fn.side_effect = lambda *args, **kwargs: self._mock_stream( - mutation_list, error_dict + mutation_list, error_dict, omit_indices ) return mock_fn @@ -374,3 +379,47 @@ async def test_run_attempt_partial_success_non_retryable(self): assert len(instance.errors[1]) == 1 assert instance.errors[1][0].grpc_status_code == 300 assert 2 not in instance.errors + + @CrossSync.pytest + async def test_run_attempt_missing_entry_retryable(self): + """If the server closes the stream successfully but omits a response + entry, the unanswered mutation must not be treated as successful. It + should be recorded as a retryable _MutateRowsIncomplete error so + idempotent entries are retried.""" + from google.cloud.bigtable.data.exceptions import _MutateRowsIncomplete + + mutations = [ + self._make_mutation(), + self._make_mutation(), + self._make_mutation(), + ] + # server omits the response entry for index 1 + mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={1}) + instance = self._make_one(mutation_entries=mutations) + instance.is_retryable = lambda x: True + with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): + with pytest.raises(_MutateRowsIncomplete): + await instance._run_attempt() + assert instance.remaining_indices == [1] + assert 0 not in instance.errors + assert len(instance.errors[1]) == 1 + assert isinstance(instance.errors[1][0], _MutateRowsIncomplete) + assert 2 not in instance.errors + + @CrossSync.pytest + async def test_run_attempt_missing_entry_non_retryable(self): + """A missing response entry for a non-retryable mutation is surfaced as + a failure rather than being silently dropped.""" + from google.cloud.bigtable.data.exceptions import _MutateRowsIncomplete + + mutations = [self._make_mutation(), self._make_mutation()] + # server omits the response entry for index 0 + mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={0}) + instance = self._make_one(mutation_entries=mutations) + instance.is_retryable = lambda x: False + with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): + await instance._run_attempt() + assert instance.remaining_indices == [] + assert len(instance.errors[0]) == 1 + assert isinstance(instance.errors[0][0], _MutateRowsIncomplete) + assert 1 not in instance.errors diff --git a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py index 2fe86a41fef0..e3979f6152fa 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py @@ -52,8 +52,11 @@ def _make_mutation(self, count=1, size=1): mutation.size = lambda: size return mutation - def _mock_stream(self, mutation_list, error_dict): + def _mock_stream(self, mutation_list, error_dict, omit_indices=None): + omit_indices = omit_indices or set() for idx, entry in enumerate(mutation_list): + if idx in omit_indices: + continue code = error_dict.get(idx, 0) yield MutateRowsResponse( entries=[ @@ -63,12 +66,12 @@ def _mock_stream(self, mutation_list, error_dict): ] ) - def _make_mock_gapic(self, mutation_list, error_dict=None): + def _make_mock_gapic(self, mutation_list, error_dict=None, omit_indices=None): mock_fn = CrossSync._Sync_Impl.Mock() if error_dict is None: error_dict = {} mock_fn.side_effect = lambda *args, **kwargs: self._mock_stream( - mutation_list, error_dict + mutation_list, error_dict, omit_indices ) return mock_fn @@ -323,3 +326,43 @@ def test_run_attempt_partial_success_non_retryable(self): assert len(instance.errors[1]) == 1 assert instance.errors[1][0].grpc_status_code == 300 assert 2 not in instance.errors + + def test_run_attempt_missing_entry_retryable(self): + """If the server closes the stream successfully but omits a response + entry, the unanswered mutation must not be treated as successful. It + should be recorded as a retryable _MutateRowsIncomplete error so + idempotent entries are retried.""" + from google.cloud.bigtable.data.exceptions import _MutateRowsIncomplete + + mutations = [ + self._make_mutation(), + self._make_mutation(), + self._make_mutation(), + ] + mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={1}) + instance = self._make_one(mutation_entries=mutations) + instance.is_retryable = lambda x: True + with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): + with pytest.raises(_MutateRowsIncomplete): + instance._run_attempt() + assert instance.remaining_indices == [1] + assert 0 not in instance.errors + assert len(instance.errors[1]) == 1 + assert isinstance(instance.errors[1][0], _MutateRowsIncomplete) + assert 2 not in instance.errors + + def test_run_attempt_missing_entry_non_retryable(self): + """A missing response entry for a non-retryable mutation is surfaced as + a failure rather than being silently dropped.""" + from google.cloud.bigtable.data.exceptions import _MutateRowsIncomplete + + mutations = [self._make_mutation(), self._make_mutation()] + mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={0}) + instance = self._make_one(mutation_entries=mutations) + instance.is_retryable = lambda x: False + with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): + instance._run_attempt() + assert instance.remaining_indices == [] + assert len(instance.errors[0]) == 1 + assert isinstance(instance.errors[0][0], _MutateRowsIncomplete) + assert 1 not in instance.errors diff --git a/packages/google-cloud-bigtable/tests/unit/v2_client/test_batcher.py b/packages/google-cloud-bigtable/tests/unit/v2_client/test_batcher.py index 847e769bf08c..5ea21272a186 100644 --- a/packages/google-cloud-bigtable/tests/unit/v2_client/test_batcher.py +++ b/packages/google-cloud-bigtable/tests/unit/v2_client/test_batcher.py @@ -213,6 +213,36 @@ def test_mutations_batcher_response_with_error_codes(): assert exc.value.exc[1].message == mocked_response[1].message +def test_mutations_batcher_asynchronous_flush_exception_is_surfaced(): + """An exception raised by the underlying ``mutate_rows`` call (e.g. a + non-retryable RPC error or a response-count mismatch) is raised inside the + async flush task. It must be captured and re-raised at ``close()`` rather + than being silently swallowed by the executor -- otherwise the failed + mutations are never reported to the user (silent data loss).""" + from google.api_core.exceptions import PermissionDenied + + with mock.patch("tests.unit.v2_client.test_batcher._Table") as mocked_table: + table = mocked_table.return_value + # flush_count=2 forces the batch to flush asynchronously (through the + # executor) as soon as the second row is added + mutation_batcher = MutationsBatcher(table=table, flush_count=2) + + row1 = DirectRow(row_key=b"row_key") + row1.set_cell("cf1", b"c1", b"1") + row2 = DirectRow(row_key=b"row_key") + row2.set_cell("cf1", b"c1", b"2") + table.mutate_rows.side_effect = PermissionDenied("denied") + + mutation_batcher.mutate_rows([row1, row2]) + with pytest.raises(MutationsBatchError) as exc: + mutation_batcher.close() + assert exc.value.message == "Errors in batch mutations." + # the whole batch (both rows) failed, so both are reported -- the error + # count stays aligned with the number of affected mutations + assert len(exc.value.exc) == 2 + assert all(isinstance(e, PermissionDenied) for e in exc.value.exc) + + def test_flow_control_event_is_set_when_not_blocked(): flow_control = _FlowControl() From 10d42d9af43505ea3640d53190b25fa0b653bd26 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Fri, 14 Aug 2026 18:04:23 +0000 Subject: [PATCH 2/3] fix(bigtable): guard batch completion callback against cancelled futures Change-Id: Ib6727718ec0e39dd7aad1298532b5e2e64439070 --- .../google/cloud/bigtable/batcher.py | 7 ++++++ .../tests/unit/v2_client/test_batcher.py | 24 +++++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py index 32c6820c5978..ece40ac2cec6 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py @@ -359,6 +359,13 @@ def _batch_completed_callback(self, future): # ``_flush_rows``; here the whole batch failed with a single exception, # so record it once per row in the batch to keep the reported error # count aligned with the number of affected mutations. + # + # A cancelled future is "done", so this callback still runs for it, but + # ``future.exception()`` would raise ``CancelledError``. Nothing here + # cancels futures today, but guard against it so the callback stays + # correct if cancellation is ever introduced. + if future.cancelled(): + return exc = future.exception() if exc is not None: for _ in range(processed_rows.rows_count): diff --git a/packages/google-cloud-bigtable/tests/unit/v2_client/test_batcher.py b/packages/google-cloud-bigtable/tests/unit/v2_client/test_batcher.py index 5ea21272a186..0944f4f20f76 100644 --- a/packages/google-cloud-bigtable/tests/unit/v2_client/test_batcher.py +++ b/packages/google-cloud-bigtable/tests/unit/v2_client/test_batcher.py @@ -243,6 +243,30 @@ def test_mutations_batcher_asynchronous_flush_exception_is_surfaced(): assert all(isinstance(e, PermissionDenied) for e in exc.value.exc) +def test_batch_completed_callback_ignores_cancelled_future(): + """A cancelled future is still "done", so the completion callback runs for + it, but ``future.exception()`` would raise ``CancelledError``. The callback + must short-circuit on a cancelled future instead of letting that propagate.""" + from google.cloud.bigtable.batcher import _BatchInfo + + table = _Table(TABLE_NAME) + with MutationsBatcher(table=table) as mutation_batcher: + batch_info = _BatchInfo(rows_count=2, mutations_count=2, mutations_size=0) + + cancelled_future = mock.Mock() + cancelled_future.cancelled.return_value = True + cancelled_future.exception.side_effect = AssertionError( + "exception() must not be called on a cancelled future" + ) + mutation_batcher.futures_mapping[cancelled_future] = batch_info + + # Should not raise, should not record any exceptions + mutation_batcher._batch_completed_callback(cancelled_future) + + assert cancelled_future not in mutation_batcher.futures_mapping + assert mutation_batcher.exceptions.qsize() == 0 + + def test_flow_control_event_is_set_when_not_blocked(): flow_control = _FlowControl() From b976fde96ab64e64c712321e246ba5f0849b4cee Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Fri, 14 Aug 2026 20:56:28 +0000 Subject: [PATCH 3/3] revert(bigtable): drop V3 mutate_rows omitted-entry handling; keep V2 batcher fix Change-Id: If379b98672c53286ab3dcb3d788006fbeb027b9c --- .../bigtable/data/_async/_mutate_rows.py | 12 ---- .../data/_sync_autogen/_mutate_rows.py | 7 --- .../unit/data/_async/test__mutate_rows.py | 55 +------------------ .../data/_sync_autogen/test__mutate_rows.py | 49 +---------------- 4 files changed, 6 insertions(+), 117 deletions(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py index c54f3c97bfb3..974e450d232b 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py @@ -214,18 +214,6 @@ async def _run_attempt(self): self._handle_entry_error(idx, exc) # bubble up exception to be handled by retry wrapper raise - # Any entries that were sent but never received a response entry (a - # successfully-closed but incomplete stream) must not be treated as - # successful. Record a retryable error so idempotent entries are retried - # and non-idempotent entries surface as failures instead of being - # silently dropped. - for idx in active_request_indices.values(): - self._handle_entry_error( - idx, - bt_exceptions._MutateRowsIncomplete( - "no response entry received for mutation" - ), - ) # check if attempt succeeded, or needs to be retried if self.remaining_indices: # unfinished work; raise exception to trigger retry diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py index ac2c1d9edac3..40e19dd85847 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py @@ -173,13 +173,6 @@ def _run_attempt(self): for idx in active_request_indices.values(): self._handle_entry_error(idx, exc) raise - for idx in active_request_indices.values(): - self._handle_entry_error( - idx, - bt_exceptions._MutateRowsIncomplete( - "no response entry received for mutation" - ), - ) if self.remaining_indices: raise bt_exceptions._MutateRowsIncomplete diff --git a/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py b/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py index 5a17b2582e00..8ff6e42532b4 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py @@ -53,13 +53,8 @@ def _make_mutation(self, count=1, size=1): return mutation @CrossSync.convert - async def _mock_stream(self, mutation_list, error_dict, omit_indices=None): - omit_indices = omit_indices or set() + async def _mock_stream(self, mutation_list, error_dict): for idx, entry in enumerate(mutation_list): - if idx in omit_indices: - # simulate a server that closes the stream OK without returning - # a response entry for this mutation - continue code = error_dict.get(idx, 0) yield MutateRowsResponse( entries=[ @@ -69,12 +64,12 @@ async def _mock_stream(self, mutation_list, error_dict, omit_indices=None): ] ) - def _make_mock_gapic(self, mutation_list, error_dict=None, omit_indices=None): + def _make_mock_gapic(self, mutation_list, error_dict=None): mock_fn = CrossSync.Mock() if error_dict is None: error_dict = {} mock_fn.side_effect = lambda *args, **kwargs: self._mock_stream( - mutation_list, error_dict, omit_indices + mutation_list, error_dict ) return mock_fn @@ -379,47 +374,3 @@ async def test_run_attempt_partial_success_non_retryable(self): assert len(instance.errors[1]) == 1 assert instance.errors[1][0].grpc_status_code == 300 assert 2 not in instance.errors - - @CrossSync.pytest - async def test_run_attempt_missing_entry_retryable(self): - """If the server closes the stream successfully but omits a response - entry, the unanswered mutation must not be treated as successful. It - should be recorded as a retryable _MutateRowsIncomplete error so - idempotent entries are retried.""" - from google.cloud.bigtable.data.exceptions import _MutateRowsIncomplete - - mutations = [ - self._make_mutation(), - self._make_mutation(), - self._make_mutation(), - ] - # server omits the response entry for index 1 - mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={1}) - instance = self._make_one(mutation_entries=mutations) - instance.is_retryable = lambda x: True - with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): - with pytest.raises(_MutateRowsIncomplete): - await instance._run_attempt() - assert instance.remaining_indices == [1] - assert 0 not in instance.errors - assert len(instance.errors[1]) == 1 - assert isinstance(instance.errors[1][0], _MutateRowsIncomplete) - assert 2 not in instance.errors - - @CrossSync.pytest - async def test_run_attempt_missing_entry_non_retryable(self): - """A missing response entry for a non-retryable mutation is surfaced as - a failure rather than being silently dropped.""" - from google.cloud.bigtable.data.exceptions import _MutateRowsIncomplete - - mutations = [self._make_mutation(), self._make_mutation()] - # server omits the response entry for index 0 - mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={0}) - instance = self._make_one(mutation_entries=mutations) - instance.is_retryable = lambda x: False - with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): - await instance._run_attempt() - assert instance.remaining_indices == [] - assert len(instance.errors[0]) == 1 - assert isinstance(instance.errors[0][0], _MutateRowsIncomplete) - assert 1 not in instance.errors diff --git a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py index e3979f6152fa..2fe86a41fef0 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py @@ -52,11 +52,8 @@ def _make_mutation(self, count=1, size=1): mutation.size = lambda: size return mutation - def _mock_stream(self, mutation_list, error_dict, omit_indices=None): - omit_indices = omit_indices or set() + def _mock_stream(self, mutation_list, error_dict): for idx, entry in enumerate(mutation_list): - if idx in omit_indices: - continue code = error_dict.get(idx, 0) yield MutateRowsResponse( entries=[ @@ -66,12 +63,12 @@ def _mock_stream(self, mutation_list, error_dict, omit_indices=None): ] ) - def _make_mock_gapic(self, mutation_list, error_dict=None, omit_indices=None): + def _make_mock_gapic(self, mutation_list, error_dict=None): mock_fn = CrossSync._Sync_Impl.Mock() if error_dict is None: error_dict = {} mock_fn.side_effect = lambda *args, **kwargs: self._mock_stream( - mutation_list, error_dict, omit_indices + mutation_list, error_dict ) return mock_fn @@ -326,43 +323,3 @@ def test_run_attempt_partial_success_non_retryable(self): assert len(instance.errors[1]) == 1 assert instance.errors[1][0].grpc_status_code == 300 assert 2 not in instance.errors - - def test_run_attempt_missing_entry_retryable(self): - """If the server closes the stream successfully but omits a response - entry, the unanswered mutation must not be treated as successful. It - should be recorded as a retryable _MutateRowsIncomplete error so - idempotent entries are retried.""" - from google.cloud.bigtable.data.exceptions import _MutateRowsIncomplete - - mutations = [ - self._make_mutation(), - self._make_mutation(), - self._make_mutation(), - ] - mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={1}) - instance = self._make_one(mutation_entries=mutations) - instance.is_retryable = lambda x: True - with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): - with pytest.raises(_MutateRowsIncomplete): - instance._run_attempt() - assert instance.remaining_indices == [1] - assert 0 not in instance.errors - assert len(instance.errors[1]) == 1 - assert isinstance(instance.errors[1][0], _MutateRowsIncomplete) - assert 2 not in instance.errors - - def test_run_attempt_missing_entry_non_retryable(self): - """A missing response entry for a non-retryable mutation is surfaced as - a failure rather than being silently dropped.""" - from google.cloud.bigtable.data.exceptions import _MutateRowsIncomplete - - mutations = [self._make_mutation(), self._make_mutation()] - mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={0}) - instance = self._make_one(mutation_entries=mutations) - instance.is_retryable = lambda x: False - with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): - instance._run_attempt() - assert instance.remaining_indices == [] - assert len(instance.errors[0]) == 1 - assert isinstance(instance.errors[0][0], _MutateRowsIncomplete) - assert 1 not in instance.errors