feat(Pub/Sub): Support WaitForProcessing and NackInmediately shutdown options by amanda-tarafa · Pull Request #15545 · googleapis/google-cloud-dotnet

amanda-tarafa · 2026-03-31T06:25:50Z

Previous shutdown mechanism is now obsolete.

gemini-code-assist

Code Review

This pull request introduces ShutdownOptions to SubscriberClient, enabling NackImmediately and WaitForProcessing shutdown modes. The changes include refactoring internal cancellation logic in SubscriberClientImpl, updating documentation, and expanding test coverage. Review feedback focuses on correcting multiple instances of the typo "Inmediately" to "Immediately" in constants, variables, and documentation.

amanda-tarafa

Typos fixed.

amanda-tarafa · 2026-03-31T06:46:55Z

/gemini review

gemini-code-assist

Code Review

This pull request introduces a more flexible shutdown mechanism for SubscriberClient, allowing users to specify NackImmediately or WaitForProcessing modes through a new StopAsync overload. The internal implementation is refactored to use granular cancellation tokens for managing different stages of the shutdown process. Review feedback suggests removing a side effect from a property getter in the LinkedCancellationTokenSource helper class and ensuring it implements IDisposable to properly release resources. Additionally, a minor correction was recommended for missing spaces in a log message concatenation.

robertvoinescu-work

LGTM: With some minor comments.

amanda-tarafa · 2026-04-16T04:20:06Z

(I'll add the method I accidentally removed, but that's just a wrapper)

… options. Previous shutdown mechanism is now obsolete.

amanda-tarafa · 2026-04-16T16:25:27Z

(I'll add the method I accidentally removed, but that's just a wrapper)

This seems like a bug on the diff checker, I didn't remove the method, I removed the overriding implementation in SubscriberClientImpl, and I changed the inherited implementation in the base class SubscriberClient from throwing to delegating on the newest StopAsync method.

I won't add the allow breaking changes label now, so that we double check before merging that the only "breaking" change is this one, but we can merge as is otherwise.

amanda-tarafa · 2026-04-16T16:25:39Z

/gemini review

gemini-code-assist

Code Review

This pull request introduces a more granular shutdown mechanism for the SubscriberClient, allowing users to choose between graceful processing of received messages or immediate negative acknowledgement via the new ShutdownOptions and ShutdownMode. The implementation involves refactoring internal cancellation logic to manage streaming pull, message handling, and push operations independently. Feedback provided focuses on correcting a typo in a source comment, removing a redundant null-conditional operator, and improving the management of background tasks in the shutdown sequence to prevent potential resource leaks.

gemini-code-assist · 2026-04-16T16:27:41Z

            HandlePush();
            // Start stream-keep-alive ping
            HandleStreamPing();
+            // On NackInmediately we nack everything that we have client side.


Typo: "NackInmediately" should be "NackImmediately".

// On NackImmediately we nack everything that we have client side.

gemini-code-assist · 2026-04-16T16:27:41Z

+                }
+
+                // If we need to nack anything, do so.
+                if (idsToNack?.Count > 0)


The null-conditional operator ?. is redundant here because idsToNack is initialized as a new HashSet<string> at line 1206 and is never null.

if (idsToNack.Count > 0)

gemini-code-assist · 2026-04-16T16:27:41Z

+        void CancelAfterDelay(CancellationTokenSource cts, TimeSpan delay) =>
+            _taskHelper.Run(async () =>
+            {
+                await _taskHelper.ConfigureAwait(_scheduler.Delay(delay, CancellationToken.None));
+                cts.Cancel();
+            });


The background tasks created by CancelAfterDelay use CancellationToken.None for the delay. If the subscriber stops gracefully before the timeout, these tasks will remain active until the delay expires. To ensure consistency and prevent resource leaks, consider using the WithCancellationToken extension method for cancellable waits, or pass a cancellation token to _scheduler.Delay that is triggered when _mainTcs.Task completes.

References

The WithCancellationToken extension method is available and should be used for cancellable waits on tasks.

robertvoinescu-work · 2026-04-16T17:45:32Z

-                CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, _taskHelper.TaskScheduler);
+            // We create the continuation on a new task to avoid the continuation to be executed on this thread
+            // if `task` is already completed. Note that although the continuation "only equeues", if there's
+            // a task (i.e. the main loop) waiting to dequeue, the dequeu would happen synchronously after the


NIT: dequeu -> dequeue

robertvoinescu-work · 2026-04-16T17:47:31Z

-                    Add(_eventReceiptModAckForExactlyOnceDelivery.WaitAsync(_softStopCts.Token)
-                        .ContinueWith(task => ProcessSuccessfulMessages(msgs, msgIds), CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, _taskHelper.TaskScheduler)
+                    Add(_eventReceiptModAckForExactlyOnceDelivery.WaitAsync(_handlerCts.Token)
+                        .ContinueWith(task => ProcessSuccessfulMessages(msgs, msgIds), _handlerCts.Token, TaskContinuationOptions.ExecuteSynchronously, _taskHelper.TaskScheduler)


This continue with is on a task so it'll return a Task<Task> after ContinueWith, where the outer task completes on the first await in ProcessSuccessfulMessages. On shutdown this might cause issues where we shutdown before ProcessSuccessfulMessages is called as we think all subtasks are complete. The simples fix is we unwrap:

_eventReceiptModAckForExactlyOnceDelivery.WaitAsync(_handlerCts.Token) .ContinueWith(task => ProcessSuccessfulMessages(msgs, msgIds).unwrap()

or we switch to using the await keyword instead of .ContinueWith

This was here before so maybe a follow-up is fine.

robertvoinescu-work · 2026-04-16T21:12:04Z

+                }
+
+                HashSet<string> idsToNack = new HashSet<string>();
+                foreach (var leasingSet in messageSetsToNack)


These aren't necessarily leasing sets since they may also include expired sets, maybe just messageSet

robertvoinescu-work · 2026-04-16T21:19:07Z

                Backoff = _disconnectBackoff;
            }
        }
+


NIT: Blank line

robertvoinescu-work · 2026-04-16T21:46:23Z

                    }
-                    // Lock ack/nack-queues, this is accessed concurrently here and in "master" thread.
-                    lock (_lock)
+                    if (wasInLeaseTracking)


Just want to confirm this logic since it departs from the previous for non-shutdown scenarios. We won't Ack/Nack any messages that have hit their max extension period (i.e. 'expired').

robertvoinescu-work · 2026-04-16T22:20:37Z

                    Add(delayTask, new NextAction(false, hasAcksOrNacks
                        ? () => { ackActionToRetry(retryIds); StartPush(); }
-                        : () => { extendActionToRetry(extendIds.Where(j => retryIds.Contains(j.Id))); StartPush(); }));
+                    : () => { extendActionToRetry(extendIds.Where(j => retryIds.Contains(j.Id))); StartPush(); }));


NIT: White space.

robertvoinescu-work · 2026-04-16T22:41:08Z

+                        {
+                            lock (_lock)
+                            {
+                                _messageSetsInLeasing.Remove(leaseTracking);


As a side note for the future: I think for me ideally the message state would be encapsulated within the message object itself, rather than tracking it externally using HashSets in this broader scope. I know changing that would require a significant refactor, so I'm completely fine with the current approach. Just wanted to note it as a potential task to tackle when we have the bandwidth.

robertvoinescu-work · 2026-04-16T22:46:59Z

+            // if `task` is already completed. Note that although the continuation "only equeues", if there's
+            // a task (i.e. the main loop) waiting to dequeue, the dequeu would happen synchronously after the
+            // synchronous enqueue as well as the execution of the next task.
+            _taskHelper.Run(() =>


Just want to confirm, this is the line that fixes the intermittent deadlock we were seeing in the integrations tests prior, is that correct?

robertvoinescu-work · 2026-04-16T22:52:27Z

            }
            if (acks.Count > 0)
            {
                _pushInFlight += acks.Count;


I think we should lock on _pushInFlight as it's accessed here, in HandleAckResponse and in the main StartAsync loop. Maybe it's worth it to create a helper method to reduce the locking duplication, but your call:

IncrementPushInFlightCount(int increment) { lock(_lock) { _pushInFlight += increment } }

amanda-tarafa mentioned this pull request Mar 31, 2026

feat(PubSub): Add NackImmediately and WaitForProcessing shutdown mode #15482

Closed

amanda-tarafa marked this pull request as ready for review March 31, 2026 06:30

amanda-tarafa requested a review from a team as a code owner March 31, 2026 06:30

amanda-tarafa requested a review from robertvoinescu-work March 31, 2026 06:30

gemini-code-assist bot reviewed Mar 31, 2026

View reviewed changes

amanda-tarafa commented Mar 31, 2026

View reviewed changes

Comment thread apis/Google.Cloud.PubSub.V1/docs/index.md

Comment thread apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1/SubscriberClientImpl.cs Outdated

amanda-tarafa force-pushed the pubsub-shutdown branch from 3ed3c8d to f31e1b5 Compare March 31, 2026 06:46

gemini-code-assist bot reviewed Mar 31, 2026

View reviewed changes

robertvoinescu-work approved these changes Mar 31, 2026

View reviewed changes

amanda-tarafa force-pushed the pubsub-shutdown branch from f31e1b5 to c719b95 Compare April 15, 2026 23:34

amanda-tarafa requested a review from robertvoinescu-work April 16, 2026 16:08

feat(Pub/Sub): Support WaitForProcessing and NackInmediately shutdown…

366570e

… options. Previous shutdown mechanism is now obsolete.

amanda-tarafa force-pushed the pubsub-shutdown branch from c719b95 to 366570e Compare April 16, 2026 16:22

gemini-code-assist bot reviewed Apr 16, 2026

View reviewed changes

robertvoinescu-work reviewed Apr 16, 2026

View reviewed changes

Conversation

amanda-tarafa commented Mar 31, 2026

Uh oh!

gemini-code-assist bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

amanda-tarafa left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

amanda-tarafa commented Mar 31, 2026

Uh oh!

gemini-code-assist bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

Uh oh!

Uh oh!

Uh oh!

robertvoinescu-work left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

amanda-tarafa commented Apr 16, 2026

Uh oh!

amanda-tarafa commented Apr 16, 2026

Uh oh!

amanda-tarafa commented Apr 16, 2026

Uh oh!

gemini-code-assist bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

gemini-code-assist bot Apr 16, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist bot Apr 16, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist bot Apr 16, 2026

Choose a reason for hiding this comment

Uh oh!

robertvoinescu-work Apr 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

robertvoinescu-work Apr 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

robertvoinescu-work Apr 16, 2026 •

edited

Loading

robertvoinescu-work Apr 16, 2026 •

edited

Loading