fix(bigtable): surface batcher flush errors and disable timer (#18145)

Thank you for opening a Pull Request! Before submitting your PR, there
are a few things you can do to make sure it goes smoothly:
- [ ] Make sure to open an issue as a
[bug/issue](https://github.com/googleapis/google-cloud-python/issues)
before writing your code! That way we can discuss the change, evaluate
designs, and agree on the general idea
- [ ] Ensure the tests and linter pass
- [ ] Code coverage does not decrease (if any source code was changed)
- [ ] Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> 🦕

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

Author: mutianf

packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py

@@ -174,10 +174,15 @@ class MutationsBatcher(object):
     request.
 
     This class is not suited for usage in systems where each mutation
-    must be guaranteed to be sent, since calling mutate may only result in an
-    in-memory change. In a case of a system crash, any :class:`DirectRow` remaining in
-    memory will not necessarily be sent to the service, even after the
-    completion of the :func:`mutate()` method.
+    must be guaranteed to be sent, since calling :func:`mutate()` may only
+    result in an in-memory change. Rows are only sent to the service when a size
+    limit is reached, when :func:`flush()` is called explicitly, or when the
+    batcher is closed (:func:`close()` is also registered to run at interpreter
+    exit). There is no time-based background flush. As a result, if the process
+    terminates abruptly -- e.g. a crash, ``SIGKILL``, or ``os._exit`` where the
+    ``atexit`` handler never runs -- any :class:`DirectRow` still buffered in
+    memory is silently dropped and never sent, even after :func:`mutate()`
+    returned.
 
     Note on thread safety: The same :class:`MutationBatcher` cannot be shared by multiple end-user threads.
 
@@ -196,8 +201,9 @@ class MutationsBatcher(object):
         (5 MB).
 
     :type flush_interval: float
-    :param flush_interval: (Optional) The interval (in seconds) between asynchronous flush.
-        Default is 1 second.
+    :param flush_interval: (Deprecated) No longer used. Retained only for
+        backwards compatibility. There is no time-based background flush; see the
+        class docstring for when rows are sent.
 
     :type batch_completed_callback: Callable[list:[`~google.rpc.status_pb2.Status`]] = None
     :param batch_completed_callback: (Optional) A callable for handling responses
@@ -219,8 +225,12 @@ def __init__(
         self.table = table
         self._executor = concurrent.futures.ThreadPoolExecutor()
         atexit.register(self.close)
-        self._timer = threading.Timer(flush_interval, self.flush)
-        self._timer.start()
+        # ``flush_interval`` is retained for backwards compatibility but is no
+        # longer used: the previous background ``threading.Timer`` was one-shot
+        # (never re-armed), so it fired at most once and could silently drop the
+        # rows it dequeued if that single flush raised on the timer thread.
+        # Flushing now happens only on size thresholds, explicit ``flush()``,
+        # or ``close()`` (also registered via ``atexit``).
         self.flow_control = _FlowControl(
             max_mutations=MAX_OUTSTANDING_ELEMENTS,
             max_mutation_bytes=MAX_OUTSTANDING_BYTES,
@@ -425,7 +435,19 @@ def close(self):
         :raises:
             * :exc:`.batcherMutationsBatchError` if there's any error in the mutations.
         """
-        self.flush()
+        try:
+            self.flush()
+        except MutationsBatchError as exc:
+            for e in exc.exc:
+                self.exceptions.put(e)
+        except Exception as exc:
+            # A failure in this final synchronous flush must not abort cleanup.
+            # If it propagated here it would skip the executor shutdown (leaving
+            # in-flight async flushes un-awaited) and skip draining
+            # self.exceptions, masking every error already captured from
+            # earlier async flushes -- silently discarding those failures.
+            # Record it like any other batch failure and continue.
+            self.exceptions.put(exc)
         self._executor.shutdown(wait=True)
         atexit.unregister(self.close)
         if self.exceptions.qsize() > 0:

packages/google-cloud-bigtable/tests/unit/v2_client/test_batcher.py

@@ -175,20 +175,18 @@ def test_mutations_batcher_context_manager_flushed_when_closed():
 
 @mock.patch("google.cloud.bigtable.batcher.threading.Timer")
 @mock.patch("google.cloud.bigtable.batcher.MutationsBatcher.flush")
-def test_mutations_batcher_flush_interval(mocked_flush, mocked_timer):
+def test_mutations_batcher_flush_interval_does_not_start_timer(
+    mocked_flush, mocked_timer
+):
+    # ``flush_interval`` is accepted for backwards compatibility but no longer
+    # starts a background timer. Constructing the batcher must not create a
+    # timer or trigger a flush.
     table = _Table(TABLE_NAME)
-    flush_interval = 0.5
-    mutation_batcher = MutationsBatcher(table=table, flush_interval=flush_interval)
+    MutationsBatcher(table=table, flush_interval=0.5)
 
-    mocked_timer.assert_called_once_with(flush_interval, mutation_batcher.flush)
-    mocked_timer.return_value.start.assert_called_once_with()
+    mocked_timer.assert_not_called()
     mocked_flush.assert_not_called()
 
-    # Manually invoke the timer callback to verify it calls flush
-    timer_callback = mocked_timer.call_args[0][1]
-    timer_callback()
-    mocked_flush.assert_called_once_with()
-
 
 def test_mutations_batcher_response_with_error_codes():
     from google.rpc.status_pb2 import Status
@@ -267,6 +265,38 @@ def test_batch_completed_callback_ignores_cancelled_future():
         assert mutation_batcher.exceptions.qsize() == 0
 
 
+def test_mutations_batcher_close_surfaces_errors_when_final_flush_raises():
+    """If the final flush in ``close()`` raises, ``close()`` must still shut
+    down the executor and surface every accumulated error -- including ones
+    already captured from async flushes -- instead of letting the flush
+    exception mask them and abort cleanup (silent data loss)."""
+    from google.api_core.exceptions import PermissionDenied, ServiceUnavailable
+
+    with mock.patch("tests.unit.v2_client.test_batcher._Table") as mocked_table:
+        table = mocked_table.return_value
+        mutation_batcher = MutationsBatcher(table=table)
+
+        # Simulate an error already captured earlier (e.g. from an async flush).
+        prior_error = ServiceUnavailable("earlier async failure")
+        mutation_batcher.exceptions.put(prior_error)
+
+        # The row stays queued (below flush_count), so it is only flushed by
+        # close(); make that final flush raise.
+        row = DirectRow(row_key=b"row_key")
+        row.set_cell("cf1", b"c1", b"1")
+        mutation_batcher.mutate(row)
+        table.mutate_rows.side_effect = PermissionDenied("denied")
+
+        with pytest.raises(MutationsBatchError) as exc:
+            mutation_batcher.close()
+
+        # both the pre-existing and the flush-time errors are reported
+        assert prior_error in exc.value.exc
+        assert any(isinstance(e, PermissionDenied) for e in exc.value.exc)
+        # cleanup still ran despite the flush raising
+        assert mutation_batcher._executor._shutdown is True
+
+
 def test_flow_control_event_is_set_when_not_blocked():
     flow_control = _FlowControl()