From ebcc85ef6f56131b2ff621ab7c1fcdc1732ef082 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Mon, 19 Oct 2026 18:25:52 +0000 Subject: [PATCH 1/9] fix(bigtable): pack unpacked error details in _get_status and harden batcher error handling --- .../bigtable/data/_async/mutations_batcher.py | 21 +++++--- .../google/cloud/bigtable/data/_helpers.py | 28 ++++++++-- .../data/_sync_autogen/mutations_batcher.py | 15 ++++-- .../data/_async/test_mutations_batcher.py | 49 +++++++++++++++++ .../_sync_autogen/test_mutations_batcher.py | 48 +++++++++++++++++ .../tests/unit/data/test__helpers.py | 54 ++++++++++++++++++- 6 files changed, 201 insertions(+), 14 deletions(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py index 2ee11ad9834b..272e564e9df5 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py @@ -451,7 +451,7 @@ async def _execute_mutate_rows( list of FailedMutationEntryError objects for mutations that failed. FailedMutationEntryError objects will not contain index information """ - statuses = [status_pb2.Status(code=code_pb2.UNKNOWN) for _ in range(len(batch))] + statuses: list[status_pb2.Status] | None = None try: operation = CrossSync._MutateRowsOperation( self._target.client._gapic_client, @@ -464,14 +464,18 @@ async def _execute_mutate_rows( ) await operation.start() except MutationsExceptionGroup as e: - statuses = _get_statuses_from_mutations_exception_group(e, len(batch)) + if self._user_batch_completed_callback is not None: + statuses = _get_statuses_from_mutations_exception_group(e, len(batch)) # strip index information from exceptions, since it is not useful in a batch context for subexc in e.exceptions: subexc.index = None return list(e.exceptions) else: - statuses = [status_pb2.Status(code=code_pb2.OK) for _ in range(len(batch))] + if self._user_batch_completed_callback is not None: + statuses = [ + status_pb2.Status(code=code_pb2.OK) for _ in range(len(batch)) + ] finally: # mark batch as complete in flow control await self._flow_control.remove_from_flow(batch) @@ -480,6 +484,11 @@ async def _execute_mutate_rows( # This is an internal callback used by the legacy synchronous shim, # though async callbacks are awaited when running in async mode. if self._user_batch_completed_callback: + if statuses is None: + statuses = [ + status_pb2.Status(code=code_pb2.UNKNOWN) + for _ in range(len(batch)) + ] try: result = self._user_batch_completed_callback(statuses) if CrossSync.is_async: @@ -631,10 +640,10 @@ async def _wait_for_batch_results( return [] exceptions: list[FailedMutationEntryError] = [] for task, batch in tasks: - if CrossSync.is_async: - # futures don't need to be awaited in sync mode - await task try: + if CrossSync.is_async: + # futures don't need to be awaited in sync mode + await task exc_list = task.result() if exc_list: # expect a list of FailedMutationEntryError objects diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py index 2bdce069e8e0..19402e5b311e 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py @@ -33,6 +33,7 @@ from google.api_core import exceptions as core_exceptions from google.api_core import retry as retries from google.api_core.retry import RetryFailureReason, exponential_sleep_generator +from google.protobuf import any_pb2, message from google.rpc import code_pb2, status_pb2 from google.cloud.bigtable.data.exceptions import ( @@ -291,10 +292,22 @@ def _get_status(exc: Optional[Exception]) -> status_pb2.Status: isinstance(exc, core_exceptions.GoogleAPICallError) and exc.grpc_status_code is not None ): - return status_pb2.Status( # type: ignore[unreachable] + details: list[any_pb2.Any] = [] # type: ignore[unreachable] + for detail in exc.details or (): + if isinstance(detail, any_pb2.Any): + details.append(detail) + elif isinstance(detail, message.Message): + packed = any_pb2.Any() + packed.Pack(detail) + details.append(packed) + elif isinstance(getattr(detail, "_pb", None), message.Message): + packed = any_pb2.Any() + packed.Pack(detail._pb) + details.append(packed) + return status_pb2.Status( code=exc.grpc_status_code.value[0], message=exc.message, - details=exc.details, + details=details, ) return status_pb2.Status( @@ -337,11 +350,18 @@ def _get_error_type( Args: - call_code: Exception type or gRPC status code. + Raises: + - ValueError: if call_code is not a valid retryable gRPC error status code + or exception type. """ if isinstance(call_code, type): return call_code - else: - return type(core_exceptions.from_grpc_status(call_code, "")) + if isinstance(call_code, bool): + raise ValueError(f"Invalid gRPC error status code: {call_code!r}") + error_type = type(core_exceptions.from_grpc_status(call_code, "")) + if error_type is core_exceptions.GoogleAPICallError: + raise ValueError(f"Invalid gRPC error status code: {call_code!r}") + return error_type def _get_retryable_errors( diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py index 83ffcdecb33a..e94d8c59c802 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py @@ -391,7 +391,7 @@ def _execute_mutate_rows( list[FailedMutationEntryError]: list of FailedMutationEntryError objects for mutations that failed. FailedMutationEntryError objects will not contain index information""" - statuses = [status_pb2.Status(code=code_pb2.UNKNOWN) for _ in range(len(batch))] + statuses: list[status_pb2.Status] | None = None try: operation = CrossSync._Sync_Impl._MutateRowsOperation( self._target.client._gapic_client, @@ -404,15 +404,24 @@ def _execute_mutate_rows( ) operation.start() except MutationsExceptionGroup as e: - statuses = _get_statuses_from_mutations_exception_group(e, len(batch)) + if self._user_batch_completed_callback is not None: + statuses = _get_statuses_from_mutations_exception_group(e, len(batch)) for subexc in e.exceptions: subexc.index = None return list(e.exceptions) else: - statuses = [status_pb2.Status(code=code_pb2.OK) for _ in range(len(batch))] + if self._user_batch_completed_callback is not None: + statuses = [ + status_pb2.Status(code=code_pb2.OK) for _ in range(len(batch)) + ] finally: self._flow_control.remove_from_flow(batch) if self._user_batch_completed_callback: + if statuses is None: + statuses = [ + status_pb2.Status(code=code_pb2.UNKNOWN) + for _ in range(len(batch)) + ] try: result = self._user_batch_completed_callback(statuses) if inspect.isawaitable(result): diff --git a/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py b/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py index 7c9bb888e34c..b144a1bbeb0c 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py @@ -1183,6 +1183,55 @@ async def test__execute_mutate_rows_batch_completed_callback_coroutine(self): else: assert called_with == [] + @CrossSync.pytest + async def test__execute_mutate_rows_batch_completed_callback_unexpected_error(self): + from google.rpc import code_pb2, status_pb2 + + with mock.patch.object(CrossSync._MutateRowsOperation, "start") as mutate_rows: + mutate_rows.side_effect = RuntimeError("unexpected error") + table = mock.Mock() + table.default_mutate_rows_operation_timeout = 17 + table.default_mutate_rows_attempt_timeout = 13 + table.default_mutate_rows_retryable_errors = () + callback = mock.Mock() + async with self._make_one(table) as instance: + instance._user_batch_completed_callback = callback + batch = [self._make_mutation(), self._make_mutation()] + with pytest.raises(RuntimeError, match="unexpected error"): + await instance._execute_mutate_rows(batch, mock.Mock()) + callback.assert_called_once_with( + [ + status_pb2.Status(code=code_pb2.UNKNOWN), + status_pb2.Status(code=code_pb2.UNKNOWN), + ] + ) + + @CrossSync.pytest + async def test__wait_for_batch_results_unexpected_task_exception(self): + """Unexpected exceptions raised by batch tasks should be wrapped in FailedMutationEntryError.""" + batch = [self._make_mutation(), self._make_mutation()] + expected_err = RuntimeError("task failed") + + async with self._make_one() as instance: + with mock.patch.object( + instance, "_execute_mutate_rows", side_effect=expected_err + ): + with mock.patch.object( + instance._flow_control, "add_to_flow" + ) as flow_mock: + + async def gen(x): + yield x + + flow_mock.side_effect = lambda x: gen(x) + await instance._flush_internal(batch) + assert len(instance._oldest_exceptions) == 2 + for failed_entry in instance._oldest_exceptions: + assert failed_entry.index is None + assert failed_entry.__cause__ is expected_err + instance._oldest_exceptions.clear() + instance._newest_exceptions.clear() + @CrossSync.pytest async def test__raise_exceptions(self): """Raise exceptions and reset error state""" diff --git a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py index 51c28e95854e..d67a7623789e 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py @@ -1056,6 +1056,54 @@ def test__execute_mutate_rows_batch_completed_callback_coroutine(self): assert result == [] assert called_with == [] + def test__execute_mutate_rows_batch_completed_callback_unexpected_error(self): + from google.rpc import code_pb2, status_pb2 + + with mock.patch.object( + CrossSync._Sync_Impl._MutateRowsOperation, "start" + ) as mutate_rows: + mutate_rows.side_effect = RuntimeError("unexpected error") + table = mock.Mock() + table.default_mutate_rows_operation_timeout = 17 + table.default_mutate_rows_attempt_timeout = 13 + table.default_mutate_rows_retryable_errors = () + callback = mock.Mock() + with self._make_one(table) as instance: + instance._user_batch_completed_callback = callback + batch = [self._make_mutation(), self._make_mutation()] + with pytest.raises(RuntimeError, match="unexpected error"): + instance._execute_mutate_rows(batch, mock.Mock()) + callback.assert_called_once_with( + [ + status_pb2.Status(code=code_pb2.UNKNOWN), + status_pb2.Status(code=code_pb2.UNKNOWN), + ] + ) + + def test__wait_for_batch_results_unexpected_task_exception(self): + """Unexpected exceptions raised by batch tasks should be wrapped in FailedMutationEntryError.""" + batch = [self._make_mutation(), self._make_mutation()] + expected_err = RuntimeError("task failed") + with self._make_one() as instance: + with mock.patch.object( + instance, "_execute_mutate_rows", side_effect=expected_err + ): + with mock.patch.object( + instance._flow_control, "add_to_flow" + ) as flow_mock: + + def gen(x): + yield x + + flow_mock.side_effect = lambda x: gen(x) + instance._flush_internal(batch) + assert len(instance._oldest_exceptions) == 2 + for failed_entry in instance._oldest_exceptions: + assert failed_entry.index is None + assert failed_entry.__cause__ is expected_err + instance._oldest_exceptions.clear() + instance._newest_exceptions.clear() + def test__raise_exceptions(self): """Raise exceptions and reset error state""" from google.cloud.bigtable.data import exceptions diff --git a/packages/google-cloud-bigtable/tests/unit/data/test__helpers.py b/packages/google-cloud-bigtable/tests/unit/data/test__helpers.py index 3e44ec7c3d04..09389f6bb664 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/test__helpers.py +++ b/packages/google-cloud-bigtable/tests/unit/data/test__helpers.py @@ -16,7 +16,8 @@ import mock import pytest from google.api_core import exceptions as core_exceptions -from google.rpc import code_pb2, status_pb2 +from google.protobuf import any_pb2 +from google.rpc import code_pb2, error_details_pb2, status_pb2 import google.cloud.bigtable.data._helpers as _helpers import google.cloud.bigtable.data.exceptions as bt_exceptions @@ -351,6 +352,39 @@ def test_get_statuses_from_mutations_exception_group( ) assert statuses[0] == expected_status + def test_get_status_packs_unpacked_error_details(self): + err_info = error_details_pb2.ErrorInfo( + reason="IAM_PERMISSION_DENIED", + domain="bigtable.googleapis.com", + metadata={"resource": "projects/p/instances/i/tables/t"}, + ) + retry_info = error_details_pb2.RetryInfo() + already_packed = any_pb2.Any() + already_packed.Pack(retry_info) + + proto_plus_wrapper = mock.Mock() + proto_plus_wrapper._pb = error_details_pb2.DebugInfo(detail="stack") + + exc = core_exceptions.PermissionDenied( + "Permission denied", + details=[err_info, already_packed, proto_plus_wrapper, "ignored"], + ) + + status = _helpers._get_status(exc) + assert status.code == code_pb2.PERMISSION_DENIED + assert status.message == "Permission denied" + assert len(status.details) == 3 + + unpacked_err_info = error_details_pb2.ErrorInfo() + assert status.details[0].Unpack(unpacked_err_info) + assert unpacked_err_info == err_info + + assert status.details[1] == already_packed + + unpacked_debug_info = error_details_pb2.DebugInfo() + assert status.details[2].Unpack(unpacked_debug_info) + assert unpacked_debug_info == proto_plus_wrapper._pb + class TestGetRetryableErrors: @pytest.mark.parametrize( @@ -397,6 +431,24 @@ def test_get_retryable_errors(self, input_codes, input_table, expected): result = _helpers._get_retryable_errors(input_codes, fake_table) assert result == expected + @pytest.mark.parametrize( + "invalid_code", + [ + 0, + grpc.StatusCode.OK, + -1, + 999, + True, + False, + "INVALID", + ], + ) + def test_get_error_type_invalid_status_code(self, invalid_code): + with pytest.raises(ValueError, match="Invalid gRPC error status code"): + _helpers._get_error_type(invalid_code) + with pytest.raises(ValueError, match="Invalid gRPC error status code"): + _helpers._get_retryable_errors([invalid_code], mock.Mock()) + class TestTrackedBackoffGenerator: def test_tracked_backoff_generator_history(self): From 4c19e5747691f284ed2982ce2a92ae53c71219fe Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Mon, 19 Oct 2026 19:16:53 +0000 Subject: [PATCH 2/9] fix(bigtable): fix TypeError swallowing in batch callback and unwrap RetryExceptionGroup in legacy batcher Change-Id: Id9737b9695e281f340fce06712362ee296ae71af --- .../google/cloud/bigtable/batcher.py | 10 +++++++--- .../bigtable/data/_async/mutations_batcher.py | 19 +++++++++---------- .../data/_sync_autogen/mutations_batcher.py | 13 +++++++------ 3 files changed, 23 insertions(+), 19 deletions(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py index 6f5e48244d25..b49d0cee1c75 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py @@ -20,7 +20,7 @@ from google.api_core import exceptions as core_exceptions -from google.cloud.bigtable.data.exceptions import MutationsExceptionGroup +from google.cloud.bigtable.data.exceptions import MutationsExceptionGroup, RetryExceptionGroup from google.cloud.bigtable.data.mutations import RowMutationEntry FLUSH_COUNT = 100 # after this many elements, send out the batch @@ -140,8 +140,12 @@ def _close_batcher(self): # return that error to the user. In standard execution paths, # FailedMutationEntryError always has an Exception cause; # defensively wrap in Unknown if __cause__ is None. - if error.__cause__ is not None: - self._exceptions.put(error.__cause__) + cause = error.__cause__ + if isinstance(cause, RetryExceptionGroup) and cause.exceptions: + # Use the terminal exception from the retry chain. + cause = cause.exceptions[-1] + if cause is not None: + self._exceptions.put(cause) else: unknown_err = core_exceptions.Unknown(str(error)) unknown_err.__cause__ = error diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py index 272e564e9df5..e0279b402cd4 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py @@ -489,22 +489,21 @@ async def _execute_mutate_rows( status_pb2.Status(code=code_pb2.UNKNOWN) for _ in range(len(batch)) ] + result = None try: result = self._user_batch_completed_callback(statuses) - if CrossSync.is_async: - if inspect.isawaitable(result): - await result - else: - if inspect.isawaitable(result): - if inspect.iscoroutine(result): - result.close() - raise TypeError( - "_user_batch_completed_callback must be a synchronous callable" - ) + if CrossSync.is_async and inspect.isawaitable(result): + await result except Exception as exc: _LOGGER.warning( f"Exception raised in user batch completion callback: {exc}" ) + if not CrossSync.is_async and inspect.isawaitable(result): + if inspect.iscoroutine(result): + result.close() + raise TypeError( + "_user_batch_completed_callback must be a synchronous callable" + ) return [] def _add_exceptions(self, excs: list[FailedMutationEntryError]): diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py index e94d8c59c802..4c1063dd17a9 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py @@ -422,18 +422,19 @@ def _execute_mutate_rows( status_pb2.Status(code=code_pb2.UNKNOWN) for _ in range(len(batch)) ] + result = None try: result = self._user_batch_completed_callback(statuses) - if inspect.isawaitable(result): - if inspect.iscoroutine(result): - result.close() - raise TypeError( - "_user_batch_completed_callback must be a synchronous callable" - ) except Exception as exc: _LOGGER.warning( f"Exception raised in user batch completion callback: {exc}" ) + if inspect.isawaitable(result): + if inspect.iscoroutine(result): + result.close() + raise TypeError( + "_user_batch_completed_callback must be a synchronous callable" + ) return [] def _add_exceptions(self, excs: list[FailedMutationEntryError]): From 1c0aa6ec4375fd47e98238069ad3892f977d34db Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Mon, 19 Oct 2026 19:27:02 +0000 Subject: [PATCH 3/9] refactor(bigtable): revert statuses lazy-init optimization, keep only TypeError and batcher fixes Change-Id: Iba6a2a1085355d70b436563c48b66cc7fd9082a7 --- .../bigtable/data/_async/mutations_batcher.py | 15 +++------------ .../data/_sync_autogen/mutations_batcher.py | 15 +++------------ .../unit/data/_async/test_mutations_batcher.py | 6 ++++-- .../data/_sync_autogen/test_mutations_batcher.py | 4 ++-- 4 files changed, 12 insertions(+), 28 deletions(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py index e0279b402cd4..7bd07d300856 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py @@ -451,7 +451,7 @@ async def _execute_mutate_rows( list of FailedMutationEntryError objects for mutations that failed. FailedMutationEntryError objects will not contain index information """ - statuses: list[status_pb2.Status] | None = None + statuses = [status_pb2.Status(code=code_pb2.UNKNOWN) for _ in range(len(batch))] try: operation = CrossSync._MutateRowsOperation( self._target.client._gapic_client, @@ -464,18 +464,14 @@ async def _execute_mutate_rows( ) await operation.start() except MutationsExceptionGroup as e: - if self._user_batch_completed_callback is not None: - statuses = _get_statuses_from_mutations_exception_group(e, len(batch)) + statuses = _get_statuses_from_mutations_exception_group(e, len(batch)) # strip index information from exceptions, since it is not useful in a batch context for subexc in e.exceptions: subexc.index = None return list(e.exceptions) else: - if self._user_batch_completed_callback is not None: - statuses = [ - status_pb2.Status(code=code_pb2.OK) for _ in range(len(batch)) - ] + statuses = [status_pb2.Status(code=code_pb2.OK) for _ in range(len(batch))] finally: # mark batch as complete in flow control await self._flow_control.remove_from_flow(batch) @@ -484,11 +480,6 @@ async def _execute_mutate_rows( # This is an internal callback used by the legacy synchronous shim, # though async callbacks are awaited when running in async mode. if self._user_batch_completed_callback: - if statuses is None: - statuses = [ - status_pb2.Status(code=code_pb2.UNKNOWN) - for _ in range(len(batch)) - ] result = None try: result = self._user_batch_completed_callback(statuses) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py index 4c1063dd17a9..b94013df0f71 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py @@ -391,7 +391,7 @@ def _execute_mutate_rows( list[FailedMutationEntryError]: list of FailedMutationEntryError objects for mutations that failed. FailedMutationEntryError objects will not contain index information""" - statuses: list[status_pb2.Status] | None = None + statuses = [status_pb2.Status(code=code_pb2.UNKNOWN) for _ in range(len(batch))] try: operation = CrossSync._Sync_Impl._MutateRowsOperation( self._target.client._gapic_client, @@ -404,24 +404,15 @@ def _execute_mutate_rows( ) operation.start() except MutationsExceptionGroup as e: - if self._user_batch_completed_callback is not None: - statuses = _get_statuses_from_mutations_exception_group(e, len(batch)) + statuses = _get_statuses_from_mutations_exception_group(e, len(batch)) for subexc in e.exceptions: subexc.index = None return list(e.exceptions) else: - if self._user_batch_completed_callback is not None: - statuses = [ - status_pb2.Status(code=code_pb2.OK) for _ in range(len(batch)) - ] + statuses = [status_pb2.Status(code=code_pb2.OK) for _ in range(len(batch))] finally: self._flow_control.remove_from_flow(batch) if self._user_batch_completed_callback: - if statuses is None: - statuses = [ - status_pb2.Status(code=code_pb2.UNKNOWN) - for _ in range(len(batch)) - ] result = None try: result = self._user_batch_completed_callback(statuses) diff --git a/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py b/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py index b144a1bbeb0c..d8aba825a927 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py @@ -1176,11 +1176,13 @@ async def test__execute_mutate_rows_batch_completed_callback_coroutine(self): async with self._make_one(table) as instance: instance._user_batch_completed_callback = async_callback batch = [self._make_mutation()] - result = await instance._execute_mutate_rows(batch, mock.Mock()) - assert result == [] if CrossSync.is_async: + result = await instance._execute_mutate_rows(batch, mock.Mock()) + assert result == [] assert called_with == [[status_pb2.Status(code=code_pb2.OK)]] else: + with pytest.raises(TypeError, match="_user_batch_completed_callback must be a synchronous callable"): + instance._execute_mutate_rows(batch, mock.Mock()) assert called_with == [] @CrossSync.pytest diff --git a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py index d67a7623789e..af129bf80c14 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py @@ -1052,8 +1052,8 @@ def test__execute_mutate_rows_batch_completed_callback_coroutine(self): with self._make_one(table) as instance: instance._user_batch_completed_callback = async_callback batch = [self._make_mutation()] - result = instance._execute_mutate_rows(batch, mock.Mock()) - assert result == [] + with pytest.raises(TypeError, match="_user_batch_completed_callback must be a synchronous callable"): + instance._execute_mutate_rows(batch, mock.Mock()) assert called_with == [] def test__execute_mutate_rows_batch_completed_callback_unexpected_error(self): From 6baa2a7faa17f51f327d2086782e9e9fbec49f19 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Mon, 19 Oct 2026 19:31:39 +0000 Subject: [PATCH 4/9] fix(bigtable): remove type: ignore[unreachable] in _get_status by splitting isinstance check Change-Id: I6cadc96645514c90bed567ce4bddd9e39c1a8cc8 --- .../google/cloud/bigtable/data/_helpers.py | 41 +++++++++---------- 1 file changed, 20 insertions(+), 21 deletions(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py index 19402e5b311e..fe7d2dce706a 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py @@ -288,27 +288,26 @@ def _get_status(exc: Optional[Exception]) -> status_pb2.Status: Returns: status_pb2.Status: A Status proto object. """ - if ( - isinstance(exc, core_exceptions.GoogleAPICallError) - and exc.grpc_status_code is not None - ): - details: list[any_pb2.Any] = [] # type: ignore[unreachable] - for detail in exc.details or (): - if isinstance(detail, any_pb2.Any): - details.append(detail) - elif isinstance(detail, message.Message): - packed = any_pb2.Any() - packed.Pack(detail) - details.append(packed) - elif isinstance(getattr(detail, "_pb", None), message.Message): - packed = any_pb2.Any() - packed.Pack(detail._pb) - details.append(packed) - return status_pb2.Status( - code=exc.grpc_status_code.value[0], - message=exc.message, - details=details, - ) + if isinstance(exc, core_exceptions.GoogleAPICallError): + grpc_code = exc.grpc_status_code + if grpc_code is not None: + details: list[any_pb2.Any] = [] + for detail in exc.details or (): + if isinstance(detail, any_pb2.Any): + details.append(detail) + elif isinstance(detail, message.Message): + packed = any_pb2.Any() + packed.Pack(detail) + details.append(packed) + elif isinstance(getattr(detail, "_pb", None), message.Message): + packed = any_pb2.Any() + packed.Pack(detail._pb) + details.append(packed) + return status_pb2.Status( + code=grpc_code.value[0], + message=exc.message, + details=details, + ) return status_pb2.Status( code=code_pb2.UNKNOWN, From 802169d3e3375bc6c96db7a09c176acaf65ce9a8 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Mon, 19 Oct 2026 19:40:29 +0000 Subject: [PATCH 5/9] fix(bigtable): default exc.message to empty string and simplify task result in async mode Change-Id: I680fe5cb809dcbd01e293d9318d576e0df49d189 --- .../google/cloud/bigtable/data/_async/mutations_batcher.py | 6 +++--- .../google/cloud/bigtable/data/_helpers.py | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py index 7bd07d300856..acbbef4a15cf 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py @@ -632,9 +632,9 @@ async def _wait_for_batch_results( for task, batch in tasks: try: if CrossSync.is_async: - # futures don't need to be awaited in sync mode - await task - exc_list = task.result() + exc_list = await task + else: + exc_list = task.result() if exc_list: # expect a list of FailedMutationEntryError objects for exc in exc_list: diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py index fe7d2dce706a..e3c87b435f05 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py @@ -305,7 +305,7 @@ def _get_status(exc: Optional[Exception]) -> status_pb2.Status: details.append(packed) return status_pb2.Status( code=grpc_code.value[0], - message=exc.message, + message=exc.message or "", details=details, ) From 8d008d0d0970e81ba71a69f664b1afbd3379db5b Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Thu, 22 Oct 2026 20:31:52 +0000 Subject: [PATCH 6/9] fix(bigtable): guard _get_status against malformed details and format code --- .../google/cloud/bigtable/batcher.py | 5 ++- .../bigtable/data/_async/mutations_batcher.py | 18 +++++---- .../google/cloud/bigtable/data/_helpers.py | 40 +++++++++++-------- .../data/_async/test_mutations_batcher.py | 5 ++- .../_sync_autogen/test_mutations_batcher.py | 5 ++- .../tests/unit/data/test__helpers.py | 33 +++++++++++++++ 6 files changed, 78 insertions(+), 28 deletions(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py index b49d0cee1c75..4f99aaf887ef 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py @@ -20,7 +20,10 @@ from google.api_core import exceptions as core_exceptions -from google.cloud.bigtable.data.exceptions import MutationsExceptionGroup, RetryExceptionGroup +from google.cloud.bigtable.data.exceptions import ( + MutationsExceptionGroup, + RetryExceptionGroup, +) from google.cloud.bigtable.data.mutations import RowMutationEntry FLUSH_COUNT = 100 # after this many elements, send out the batch diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py index acbbef4a15cf..2722bf77c5e6 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py @@ -483,18 +483,20 @@ async def _execute_mutate_rows( result = None try: result = self._user_batch_completed_callback(statuses) - if CrossSync.is_async and inspect.isawaitable(result): - await result + if CrossSync.is_async: + if inspect.isawaitable(result): + await result except Exception as exc: _LOGGER.warning( f"Exception raised in user batch completion callback: {exc}" ) - if not CrossSync.is_async and inspect.isawaitable(result): - if inspect.iscoroutine(result): - result.close() - raise TypeError( - "_user_batch_completed_callback must be a synchronous callable" - ) + if not CrossSync.is_async: + if inspect.isawaitable(result): + if inspect.iscoroutine(result): + result.close() + raise TypeError( + "_user_batch_completed_callback must be a synchronous callable" + ) return [] def _add_exceptions(self, excs: list[FailedMutationEntryError]): diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py index e3c87b435f05..dbca85cb1518 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py @@ -291,23 +291,29 @@ def _get_status(exc: Optional[Exception]) -> status_pb2.Status: if isinstance(exc, core_exceptions.GoogleAPICallError): grpc_code = exc.grpc_status_code if grpc_code is not None: - details: list[any_pb2.Any] = [] - for detail in exc.details or (): - if isinstance(detail, any_pb2.Any): - details.append(detail) - elif isinstance(detail, message.Message): - packed = any_pb2.Any() - packed.Pack(detail) - details.append(packed) - elif isinstance(getattr(detail, "_pb", None), message.Message): - packed = any_pb2.Any() - packed.Pack(detail._pb) - details.append(packed) - return status_pb2.Status( - code=grpc_code.value[0], - message=exc.message or "", - details=details, - ) + try: + details: list[any_pb2.Any] = [] + for detail in exc.details or (): + try: + if isinstance(detail, any_pb2.Any): + details.append(detail) + elif isinstance(detail, message.Message): + packed = any_pb2.Any() + packed.Pack(detail) + details.append(packed) + elif isinstance(getattr(detail, "_pb", None), message.Message): + packed = any_pb2.Any() + packed.Pack(detail._pb) + details.append(packed) + except Exception: + continue + return status_pb2.Status( + code=grpc_code.value[0], + message=exc.message or "", + details=details, + ) + except Exception: + pass return status_pb2.Status( code=code_pb2.UNKNOWN, diff --git a/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py b/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py index d8aba825a927..8ac993a0f7c6 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py @@ -1181,7 +1181,10 @@ async def test__execute_mutate_rows_batch_completed_callback_coroutine(self): assert result == [] assert called_with == [[status_pb2.Status(code=code_pb2.OK)]] else: - with pytest.raises(TypeError, match="_user_batch_completed_callback must be a synchronous callable"): + with pytest.raises( + TypeError, + match="_user_batch_completed_callback must be a synchronous callable", + ): instance._execute_mutate_rows(batch, mock.Mock()) assert called_with == [] diff --git a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py index af129bf80c14..14d5f9645441 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py @@ -1052,7 +1052,10 @@ def test__execute_mutate_rows_batch_completed_callback_coroutine(self): with self._make_one(table) as instance: instance._user_batch_completed_callback = async_callback batch = [self._make_mutation()] - with pytest.raises(TypeError, match="_user_batch_completed_callback must be a synchronous callable"): + with pytest.raises( + TypeError, + match="_user_batch_completed_callback must be a synchronous callable", + ): instance._execute_mutate_rows(batch, mock.Mock()) assert called_with == [] diff --git a/packages/google-cloud-bigtable/tests/unit/data/test__helpers.py b/packages/google-cloud-bigtable/tests/unit/data/test__helpers.py index 09389f6bb664..418fe4e28c16 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/test__helpers.py +++ b/packages/google-cloud-bigtable/tests/unit/data/test__helpers.py @@ -385,6 +385,39 @@ def test_get_status_packs_unpacked_error_details(self): assert status.details[2].Unpack(unpacked_debug_info) assert unpacked_debug_info == proto_plus_wrapper._pb + def test_get_status_handles_malformed_details_and_exceptions(self): + err_info = error_details_pb2.ErrorInfo( + reason="IAM_PERMISSION_DENIED", + domain="bigtable.googleapis.com", + ) + + class BrokenWrapper: + @property + def _pb(self): + raise RuntimeError("broken _pb property") + + exc = core_exceptions.PermissionDenied( + None, + details=[BrokenWrapper(), err_info], + ) + status = _helpers._get_status(exc) + assert status.code == code_pb2.PERMISSION_DENIED + assert status.message == "" + assert len(status.details) == 1 + + unpacked_err_info = error_details_pb2.ErrorInfo() + assert status.details[0].Unpack(unpacked_err_info) + assert unpacked_err_info == err_info + + # Non-iterable details falls back to UNKNOWN without raising + malformed_exc = core_exceptions.PermissionDenied( + "Permission denied", + details=123, # type: ignore[arg-type] + ) + fallback_status = _helpers._get_status(malformed_exc) + assert fallback_status.code == code_pb2.UNKNOWN + assert "Permission denied" in fallback_status.message + class TestGetRetryableErrors: @pytest.mark.parametrize( From 1d9faa73113982acc62df4522474f0082e3be8c6 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Thu, 22 Oct 2026 21:24:19 +0000 Subject: [PATCH 7/9] fix(bigtable): render exceptions safely in the _get_status UNKNOWN fallback Change-Id: Ic8922566c1e440681c8632e955b2cd50ba1e0eaf --- .../google/cloud/bigtable/data/_helpers.py | 24 ++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py index dbca85cb1518..1cb4a3b88fee 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py @@ -317,10 +317,32 @@ def _get_status(exc: Optional[Exception]) -> status_pb2.Status: return status_pb2.Status( code=code_pb2.UNKNOWN, - message=str(exc) if exc else "An unknown error has occurred", + message=_exc_message(exc), ) +def _exc_message(exc: Optional[Exception]) -> str: + """ + Helper function that renders an exception as a string, without raising. + + ``GoogleAPICallError.__str__`` reads ``exc.details``, which raises if the + exception was built with a malformed ``details`` value, so fall back to the + raw message when that happens. + + Args: + exc: The exception to render. + Returns: + str: A description of the exception. + """ + if exc is None: + return "An unknown error has occurred" + try: + return str(exc) + except Exception: + raw_message = getattr(exc, "message", None) + return raw_message if isinstance(raw_message, str) else type(exc).__name__ + + def _validate_timeouts( operation_timeout: float, attempt_timeout: float | None, allow_none: bool = False ): From 8ca6556847c62c320366f6ea8575cc094415b7e8 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Mon, 26 Oct 2026 14:46:58 +0000 Subject: [PATCH 8/9] revert(bigtable): keep sync callback TypeError inside the logged try block Change-Id: I9dc3a3186216d6d5252246d9488b883753e0695b --- .../bigtable/data/_async/mutations_batcher.py | 15 +++++++-------- .../data/_sync_autogen/mutations_batcher.py | 13 ++++++------- .../unit/data/_async/test_mutations_batcher.py | 9 ++------- .../data/_sync_autogen/test_mutations_batcher.py | 7 ++----- 4 files changed, 17 insertions(+), 27 deletions(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py index 2722bf77c5e6..a80b1a3fb049 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py @@ -480,23 +480,22 @@ async def _execute_mutate_rows( # This is an internal callback used by the legacy synchronous shim, # though async callbacks are awaited when running in async mode. if self._user_batch_completed_callback: - result = None try: result = self._user_batch_completed_callback(statuses) if CrossSync.is_async: if inspect.isawaitable(result): await result + else: + if inspect.isawaitable(result): + if inspect.iscoroutine(result): + result.close() + raise TypeError( + "_user_batch_completed_callback must be a synchronous callable" + ) except Exception as exc: _LOGGER.warning( f"Exception raised in user batch completion callback: {exc}" ) - if not CrossSync.is_async: - if inspect.isawaitable(result): - if inspect.iscoroutine(result): - result.close() - raise TypeError( - "_user_batch_completed_callback must be a synchronous callable" - ) return [] def _add_exceptions(self, excs: list[FailedMutationEntryError]): diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py index b94013df0f71..83ffcdecb33a 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py @@ -413,19 +413,18 @@ def _execute_mutate_rows( finally: self._flow_control.remove_from_flow(batch) if self._user_batch_completed_callback: - result = None try: result = self._user_batch_completed_callback(statuses) + if inspect.isawaitable(result): + if inspect.iscoroutine(result): + result.close() + raise TypeError( + "_user_batch_completed_callback must be a synchronous callable" + ) except Exception as exc: _LOGGER.warning( f"Exception raised in user batch completion callback: {exc}" ) - if inspect.isawaitable(result): - if inspect.iscoroutine(result): - result.close() - raise TypeError( - "_user_batch_completed_callback must be a synchronous callable" - ) return [] def _add_exceptions(self, excs: list[FailedMutationEntryError]): diff --git a/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py b/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py index 8ac993a0f7c6..b144a1bbeb0c 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py @@ -1176,16 +1176,11 @@ async def test__execute_mutate_rows_batch_completed_callback_coroutine(self): async with self._make_one(table) as instance: instance._user_batch_completed_callback = async_callback batch = [self._make_mutation()] + result = await instance._execute_mutate_rows(batch, mock.Mock()) + assert result == [] if CrossSync.is_async: - result = await instance._execute_mutate_rows(batch, mock.Mock()) - assert result == [] assert called_with == [[status_pb2.Status(code=code_pb2.OK)]] else: - with pytest.raises( - TypeError, - match="_user_batch_completed_callback must be a synchronous callable", - ): - instance._execute_mutate_rows(batch, mock.Mock()) assert called_with == [] @CrossSync.pytest diff --git a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py index 14d5f9645441..d67a7623789e 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py @@ -1052,11 +1052,8 @@ def test__execute_mutate_rows_batch_completed_callback_coroutine(self): with self._make_one(table) as instance: instance._user_batch_completed_callback = async_callback batch = [self._make_mutation()] - with pytest.raises( - TypeError, - match="_user_batch_completed_callback must be a synchronous callable", - ): - instance._execute_mutate_rows(batch, mock.Mock()) + result = instance._execute_mutate_rows(batch, mock.Mock()) + assert result == [] assert called_with == [] def test__execute_mutate_rows_batch_completed_callback_unexpected_error(self): From ece90f38263b3a6cc64254a9dced552ae35d9d25 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Mon, 26 Oct 2026 19:09:55 +0000 Subject: [PATCH 9/9] docs(bigtable): explain the base-class check in _get_error_type Change-Id: I2bfee48b1bc534f1b6f088f5af05e4f2ffa446ce --- .../google/cloud/bigtable/data/_helpers.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py index 1cb4a3b88fee..7550862dbd7a 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py @@ -386,6 +386,10 @@ def _get_error_type( if isinstance(call_code, bool): raise ValueError(f"Invalid gRPC error status code: {call_code!r}") error_type = type(core_exceptions.from_grpc_status(call_code, "")) + # from_grpc_status falls back to the base GoogleAPICallError for codes it + # doesn't recognize (including OK / 0). Returning the base class here would + # make retries.if_exception_type match every GoogleAPICallError subclass, + # silently turning all errors into retryable ones, so reject it instead. if error_type is core_exceptions.GoogleAPICallError: raise ValueError(f"Invalid gRPC error status code: {call_code!r}") return error_type