This is an automated email from the ASF dual-hosted git repository. damccorm pushed a commit to branch cp-39820 in repository https://gitbox.apache.org/repos/asf/beam.git
commit cc1eb5d485241cb7b1a646d45f719bf4b790b746 Author: Mattie Fu <[email protected]> AuthorDate: Thu Aug 20 09:45:12 2026 -0400 Require google-cloud-bigtable>=2.42.0 and test write error surfacing (#39820) google-cloud-bigtable 2.42.0 improves MutationsBatcher error handling so that errors raised during async flushes are surfaced instead of being swallowed. Bump the minimum version and add BigtableIO write tests that assert batch errors are surfaced (and fail the bundle) from the async flush path, the buffered-row flush at finish_bundle, and close() itself. --- sdks/python/apache_beam/io/gcp/bigtableio_test.py | 55 +++++++++++++++++++++++ sdks/python/setup.py | 4 +- 2 files changed, 58 insertions(+), 1 deletion(-) diff --git a/sdks/python/apache_beam/io/gcp/bigtableio_test.py b/sdks/python/apache_beam/io/gcp/bigtableio_test.py index d9ef12a1659..7c371c9b383 100644 --- a/sdks/python/apache_beam/io/gcp/bigtableio_test.py +++ b/sdks/python/apache_beam/io/gcp/bigtableio_test.py @@ -331,6 +331,61 @@ class TestWriteBigTable(unittest.TestCase): ServiceCallMetric.bigtable_error_code_to_grpc_status_string(OK), 2) + def test_write_batch_error_surfaces_from_async_flush(self): + write_fn = bigtableio._BigTableWriteFn( + self._PROJECT_ID, + self._INSTANCE_ID, + self._TABLE_ID, + flush_count=1, + max_row_bytes=5242880) + write_fn.table = self.table + write_fn.start_bundle() + + direct_rows = [self.generate_row(i) for i in range(5)] + with patch.object(Table, + 'mutate_rows', + side_effect=Exception('batch RPC failed')): + for direct_row in direct_rows: + write_fn.process(direct_row) + with self.assertRaises(Exception): + write_fn.finish_bundle() + + def test_write_batch_error_surfaces_from_buffered_rows(self): + write_fn = bigtableio._BigTableWriteFn( + self._PROJECT_ID, + self._INSTANCE_ID, + self._TABLE_ID, + flush_count=1000, + max_row_bytes=5242880) + write_fn.table = self.table + write_fn.start_bundle() + + mock_mutate = MagicMock(side_effect=Exception('batch RPC failed')) + with patch.object(Table, 'mutate_rows', mock_mutate): + write_fn.process(self.generate_row(0)) + with self.assertRaises(Exception): + write_fn.finish_bundle() + self.assertGreater( + mock_mutate.call_count, 0, 'buffered row was never flushed') + + def test_write_close_error_is_surfaced(self): + write_fn = bigtableio._BigTableWriteFn( + self._PROJECT_ID, + self._INSTANCE_ID, + self._TABLE_ID, + flush_count=1000, + max_row_bytes=5242880) + write_fn.table = self.table + write_fn.start_bundle() + + with patch.object(MutationsBatcher, + 'close', + side_effect=Exception('error on close')) as mock_close: + write_fn.process(self.generate_row(0)) + with self.assertRaises(Exception): + write_fn.finish_bundle() + mock_close.assert_called_once() + def generate_row(self, index=0): rand = choice(string.ascii_letters + string.digits) value = ''.join(rand for i in range(100)) diff --git a/sdks/python/setup.py b/sdks/python/setup.py index 8af7c7705f0..44f6f2b86e4 100644 --- a/sdks/python/setup.py +++ b/sdks/python/setup.py @@ -531,7 +531,9 @@ if __name__ == '__main__': 'google-cloud-bigquery>=2.0.0,<4', 'google-cloud-bigquery-storage>=2.6.3,<3', 'google-cloud-core>=2.0.0,<3', - 'google-cloud-bigtable>=2.19.0,<3', + # 2.42.0 improves MutationsBatcher error handling: it surfaces + # errors raised during async flushes instead of swallowing them. + 'google-cloud-bigtable>=2.42.0,<3', 'google-cloud-build>=3.35.0,<4', 'google-cloud-spanner>=3.0.0,<4', # GCP Packages required by ML functionality
