diff --git a/google/cloud/bigtable/data/_async/_mutate_rows.py b/google/cloud/bigtable/data/_async/_mutate_rows.py index 8e6833bca..4eef80396 100644 --- a/google/cloud/bigtable/data/_async/_mutate_rows.py +++ b/google/cloud/bigtable/data/_async/_mutate_rows.py @@ -204,6 +204,13 @@ async def _run_attempt(self): self._handle_entry_error(idx, exc) # bubble up exception to be handled by retry wrapper raise + for orig_idx in active_request_indices.values(): + self._handle_entry_error( + orig_idx, + core_exceptions.InternalServerError( + "Mutation result missing from server response" + ), + ) # check if attempt succeeded, or needs to be retried if self.remaining_indices: # unfinished work; raise exception to trigger retry diff --git a/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py b/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py index 3bf7b562f..c34270100 100644 --- a/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py +++ b/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py @@ -163,6 +163,13 @@ def _run_attempt(self): for idx in active_request_indices.values(): self._handle_entry_error(idx, exc) raise + for orig_idx in active_request_indices.values(): + self._handle_entry_error( + orig_idx, + core_exceptions.InternalServerError( + "Mutation result missing from server response" + ), + ) if self.remaining_indices: raise bt_exceptions._MutateRowsIncomplete diff --git a/tests/unit/data/_async/test__mutate_rows.py b/tests/unit/data/_async/test__mutate_rows.py index f14fa6dee..036142b7f 100644 --- a/tests/unit/data/_async/test__mutate_rows.py +++ b/tests/unit/data/_async/test__mutate_rows.py @@ -362,3 +362,39 @@ 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_start_missing_response_entry(self): + """When a 3-entry request receives only indices 0 and 2 on normal stream close, + MutationsExceptionGroup is raised containing FailedMutationEntryError with InternalServerError for index 1.""" + from google.cloud.bigtable.data.exceptions import MutationsExceptionGroup + from google.cloud.bigtable.data.exceptions import FailedMutationEntryError + from google.api_core.exceptions import InternalServerError + + mutations = [ + self._make_mutation(), + self._make_mutation(), + self._make_mutation(), + ] + + async def mock_partial_stream(*args, **kwargs): + yield MutateRowsResponse( + entries=[ + MutateRowsResponse.Entry(index=0, status=status_pb2.Status(code=0)), + MutateRowsResponse.Entry(index=2, status=status_pb2.Status(code=0)), + ] + ) + + mock_gapic_fn = CrossSync.Mock() + mock_gapic_fn.side_effect = mock_partial_stream + instance = self._make_one(mutation_entries=mutations) + with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): + with pytest.raises(MutationsExceptionGroup) as exc_info: + await instance.start() + err_group = exc_info.value + assert len(err_group.exceptions) == 1 + entry_err = err_group.exceptions[0] + assert isinstance(entry_err, FailedMutationEntryError) + assert entry_err.index == 1 + assert isinstance(entry_err.__cause__, InternalServerError) + diff --git a/tests/unit/data/_sync_autogen/test__mutate_rows.py b/tests/unit/data/_sync_autogen/test__mutate_rows.py index b198df01b..6abded51e 100644 --- a/tests/unit/data/_sync_autogen/test__mutate_rows.py +++ b/tests/unit/data/_sync_autogen/test__mutate_rows.py @@ -307,3 +307,38 @@ 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_start_missing_response_entry(self): + """When a 3-entry request receives only indices 0 and 2 on normal stream close, + MutationsExceptionGroup is raised containing FailedMutationEntryError with InternalServerError for index 1.""" + from google.cloud.bigtable.data.exceptions import MutationsExceptionGroup + from google.cloud.bigtable.data.exceptions import FailedMutationEntryError + from google.api_core.exceptions import InternalServerError + + mutations = [ + self._make_mutation(), + self._make_mutation(), + self._make_mutation(), + ] + + def mock_partial_stream(*args, **kwargs): + yield MutateRowsResponse( + entries=[ + MutateRowsResponse.Entry(index=0, status=status_pb2.Status(code=0)), + MutateRowsResponse.Entry(index=2, status=status_pb2.Status(code=0)), + ] + ) + + mock_gapic_fn = CrossSync._Sync_Impl.Mock() + mock_gapic_fn.side_effect = mock_partial_stream + instance = self._make_one(mutation_entries=mutations) + with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): + with pytest.raises(MutationsExceptionGroup) as exc_info: + instance.start() + err_group = exc_info.value + assert len(err_group.exceptions) == 1 + entry_err = err_group.exceptions[0] + assert isinstance(entry_err, FailedMutationEntryError) + assert entry_err.index == 1 + assert isinstance(entry_err.__cause__, InternalServerError) +