[sushanb]

Summary

• Propagate future::cancel() on the future returned by AsyncBulkApplier::Create across rate-limiter acquisition (limiter_->AsyncAcquire()), in-flight MutateRows RPCs (grpc::ClientContext::TryCancel()), OnFinish retry transitions, and retry backoff timers so cancelled AsyncBulkApply calls abort promptly and never dispatch new MutateRows RPCs.
• Record StatusCode::kCancelled in BulkMutatorState::OnCancel for unconfirmed mutations when cancelled without an existing stream error instead of reporting an internal server-bug error.
• Ensure OperationContext::OnDone is invoked consistently whenever PreCall has executed (including when the retry backoff timer fails or is cancelled), and never invoked if cancelled before the first RPC attempt starts.

Testing

• Added unit and in-process gRPC server tests in google/cloud/bigtable/internal/async_bulk_apply_test.cc and google/cloud/bigtable/internal/bulk_mutator_test.cc covering cancellation before first attempt, mid-RPC before first read, before OnFinish, during backoff, racing with timer expiry, during retry limiter acquisition, and backoff timer errors.
• Verified bazel test //google/cloud/bigtable:internal_async_bulk_apply_test //google/cloud/bigtable:internal_bulk_mutator_test.

[gemini-code-assist]

Code Review

This pull request implements cancellation support for the asynchronous bulk mutation applier (AsyncBulkApplier) in the Bigtable client. It introduces thread-safe state management using a mutex to handle cancellation during rate-limiting, active RPCs, and backoff periods. Unit tests have been significantly expanded to cover these cancellation paths. The feedback suggests a minor optimization to move the future result in OnFinish to avoid an unnecessary copy of a StatusOr object.

[codecov]

Codecov Report

❌ Patch coverage is 97.97571% with 10 lines in your changes missing coverage. Please review.
✅ Project coverage is 92.50%. Comparing base (5dcdef5) to head (74e2bd0).
⚠️ Report is 20 commits behind head on main.

Files with missing lines	Patch %	Lines
google/cloud/bigtable/internal/async_bulk_apply.cc	91.07%	10 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #16648      +/-   ##
==========================================
- Coverage   92.53%   92.50%   -0.03%     
==========================================
  Files        2269     2269              
  Lines      222896   223365     +469     
==========================================
+ Hits       206250   206622     +372     
- Misses      16646    16743      +97     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[scotthart]

Review Findings

1. [HIGH] Unconfirmed non-idempotent mutations still fail with kInternal (and OnDone can receive OK) when cancelled before OnFinish with an OK stream status

• File: google/cloud/bigtable/internal/async_bulk_apply.cc (lines 125–144)

  operation_context_->PostCall(*client_context, status);
  state_.OnFinish(status);

  if (!state_.HasPendingMutations()) {
    operation_context_->OnDone(status);
    SetPromise();
    return;
  }
  if (cancelled) {
    Status cancel_status = MakeCancelledStatus();
    state_.OnCancel(cancel_status);
    operation_context_->OnDone(status.ok() ? cancel_status : status);
    SetPromise();
    return;
  }

Problem:
When Cancel() is called while a MutateRows stream is in flight, keep_reading_ is set to false, causing PerformAsyncStreamingRead (AsyncStreamingReadImpl) to stop invoking OnRead and Discard() any remaining buffered responses before calling Finish(). If the server had already finished sending responses and closed the stream with grpc::Status::OK before cancellation took effect on the wire (or in the scenario tested by CancelAfterTransientEntryResponseBeforeOnFinish), OnFinish is invoked with status.ok() == true and cancelled == true while unconfirmed mutations still have annotation.has_mutation_result == false.

Because state_.OnFinish(status) is called with status (OK) before cancelled is handled:

1. In BulkMutatorState::OnFinish, only idempotent unconfirmed mutations are moved to pending_mutations_. Any unconfirmed non-idempotent mutation (annotation.idempotency == Idempotency::kNonIdempotent) immediately takes the else branch in BulkMutatorState::OnFinish, sees last_status_.ok() == true, and is pushed to failures_ with StatusCode::kInternal ("The server never sent a confirmation for this mutation but the stream didn't fail either. This is most likely a bug..."). Calling state_.OnCancel(cancel_status) afterward does not fix entries already added to failures_.
2. If all remaining unconfirmed mutations were non-idempotent (e.g., mutation 0 was confirmed OK in an earlier OnRead chunk and mutation 1 was non-idempotent and discarded after Cancel()), state_.HasPendingMutations() is false after state_.OnFinish(status). OnFinish then returns early at if (!state_.HasPendingMutations()), passing Status{} (OK) to operation_context_->OnDone(status) instead of kCancelled.

Passing effective_status = (cancelled && status.ok()) ? MakeCancelledStatus() : status into state_.OnFinish(...) (while keeping the raw status for operation_context_->PostCall) fixes both issues:

• Mutations that already received a response in OnRead have annotation.has_mutation_result == true and are skipped by BulkMutatorState::OnFinish.
• Unconfirmed non-idempotent mutations record kCancelled in failures_ instead of the kInternal server-bug error.
• Unconfirmed idempotent mutations are moved to pending_mutations_ with last_status_ = kCancelled, which OnRetryDone() reports as kCancelled.

Suggested Fix (google/cloud/bigtable/internal/async_bulk_apply.cc lines 125–144):

  operation_context_->PostCall(*client_context, status);
  Status const effective_status =
      cancelled && status.ok() ? MakeCancelledStatus() : status;
  state_.OnFinish(effective_status);

  if (!state_.HasPendingMutations() || cancelled) {
    operation_context_->OnDone(effective_status);
    SetPromise();
    return;
  }

Suggested Test Update (google/cloud/bigtable/internal/async_bulk_apply_test.cc lines 688–693):
Adding a NonIdempotentMutation to CancelAfterTransientEntryResponseBeforeOnFinish (or a dedicated test where only a NonIdempotentMutation remains unconfirmed when Finish returns Status{}) will exercise this path:

TEST_F(AsyncBulkApplyTest, CancelAfterTransientEntryResponseBeforeOnFinish) {
  std::vector<bigtable::FailedMutation> expected = {
      {TransientError(), 1},
      {Status(StatusCode::kCancelled, "cancelled"), 2},
      {Status(StatusCode::kCancelled, "cancelled"), 3}};
  bigtable::BulkMutation mut(IdempotentMutation("r0"), IdempotentMutation("r1"),
                             IdempotentMutation("r2"),
                             NonIdempotentMutation("r3"));

---

[sushanb]

Thanks for catching that! Fixed in 74e2bd021b:

• Computed effective_status = cancelled && status.ok() ? MakeCancelledStatus() : status and passed effective_status into state_.OnFinish(effective_status) (while keeping the raw stream status for operation_context_->PostCall), then completed immediately when !state_.HasPendingMutations() || cancelled with operation_context_->OnDone(effective_status).
• Updated CancelAfterTransientEntryResponseBeforeOnFinish to include an unconfirmed NonIdempotentMutation and added CancelWithOnlyUnconfirmedNonIdempotentMutationReportsCancelledOnDone in async_bulk_apply_test.cc to cover the !state_.HasPendingMutations() path when only a non-idempotent mutation remains unconfirmed at cancellation.

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces cancellation support to the AsyncBulkApplier class in the Cloud Bigtable C++ client. It adds thread-safe state management using a mutex to track cancellation, active operations, and pending futures. The BulkMutatorState class is updated with an OnCancel method to record cancellation status, and comprehensive unit tests are added to verify cancellation behavior during different stages of the bulk apply operation (such as during limiter acquisition, backoff, and in-flight RPCs). I have no feedback to provide as there are no review comments.