From cc1eb5d485241cb7b1a646d45f719bf4b790b746 Mon Sep 17 00:00:00 2001
From: Mattie Fu <mattiefu@google.com>
Date: Thu, 20 Aug 2026 09:45:12 -0400
Subject: [PATCH] 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.
---
 .../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 8af7c7705f08..44f6f2b86e48 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