From c50d361b20ac807b3c79d6d4ede770f5977ff90c Mon Sep 17 00:00:00 2001
From: Mattie Fu <mattiefu@google.com>
Date: Wed, 19 Aug 2026 16:08:16 +0000
Subject: [PATCH] Require google-cloud-bigtable>=2.42.0 and test write error
 surfacing

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.
---
 .../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 d9ef12a16592..7c371c9b383d 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 @@ def test_write_metrics(self):
           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 61dd55cefa21..5b02dc9ab0ef 100644
--- a/sdks/python/setup.py
+++ b/sdks/python/setup.py
@@ -531,7 +531,9 @@ def get_portability_package_data():
               '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