This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 93f3e051c3e Require google-cloud-bigtable>=2.42.0 and test write error
surfacing (#39820)
93f3e051c3e is described below
commit 93f3e051c3ef47238711eb4f313d7e01477c42c3
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 61dd55cefa2..5b02dc9ab0e 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
