From 6470d7a803c9919e0c1894c562b116f098723073 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Tue, 18 Aug 2026 20:45:11 +0000 Subject: [PATCH 1/2] fix(bigtable): surface swallowed MutationsBatcher flush errors and drop one-shot timer Change-Id: I046bb0ee2c92ef20f842dea8eb5480e85717e87e --- .../google/cloud/bigtable/batcher.py | 37 ++++++++++---- .../tests/unit/v2_client/test_batcher.py | 50 +++++++++++++++---- 2 files changed, 68 insertions(+), 19 deletions(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py index ece40ac2cec6..d6445990ff01 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py +++ b/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,16 @@ def close(self): :raises: * :exc:`.batcherMutationsBatchError` if there's any error in the mutations. """ - self.flush() + try: + self.flush() + 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: diff --git a/packages/google-cloud-bigtable/tests/unit/v2_client/test_batcher.py b/packages/google-cloud-bigtable/tests/unit/v2_client/test_batcher.py index 0944f4f20f76..758ca226a12d 100644 --- a/packages/google-cloud-bigtable/tests/unit/v2_client/test_batcher.py +++ b/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() From 7a43cdd96a3e84f673f1f4c84c4a723a46d9a2fc Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Tue, 18 Aug 2026 17:00:28 -0400 Subject: [PATCH 2/2] Update packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> --- .../google-cloud-bigtable/google/cloud/bigtable/batcher.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py index d6445990ff01..17d4e72e47f1 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py @@ -437,11 +437,14 @@ def close(self): """ 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 + # 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)