feat(PubSub): Add NackImmediately and WaitForProcessing shutdown mode#15482
feat(PubSub): Add NackImmediately and WaitForProcessing shutdown mode#15482robertvoinescu-work wants to merge 2 commits intogoogleapis:mainfrom
Conversation
Summary of ChangesHello @robertvoinescu-work, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request significantly improves the SubscriberClient shutdown mechanism by introducing explicit shutdown settings: NackImmediately and WaitForProcessing. These settings provide developers with greater control over how messages are handled when a subscriber is stopped, allowing for either immediate negative acknowledgment of unhandled messages or a graceful period to complete processing before Nacking. This change enhances the reliability and predictability of subscriber termination, ensuring messages are managed appropriately based on application requirements. Highlights
Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces two new shutdown modes for SubscriberClient: NackImmediately and WaitForProcessing, providing users more control over handling outstanding messages during shutdown. A security audit found no security vulnerabilities in the changes related to the graceful shutdown process. However, there is a potential memory leak issue related to orphaned delay tasks during shutdown that should be addressed.
| } | ||
| _taskHelper.Run(async () => | ||
| { | ||
| await _taskHelper.ConfigureAwait(_scheduler.Delay(delay, CancellationToken.None)); |
There was a problem hiding this comment.
The _scheduler.Delay call uses CancellationToken.None. If the SubscriberClient is stopped via the cancellationToken parameter (triggering an immediate hard stop), these background delay tasks will continue to run for their full duration (which can be up to 1 hour by default). These tasks hold references to the CancellationTokenSource objects, which can prevent the SubscriberClient and its associated resources from being garbage collected until the delays expire. It is recommended to pass cts.Token or a combined token to the Delay call to ensure these tasks are cancelled as soon as the target CTS is cancelled or a hard stop occurs.
await _taskHelper.ConfigureAwait(_scheduler.Delay(delay, cts.Token));
amanda-tarafa
left a comment
amanda-tarafa
left a comment
There was a problem hiding this comment.
I've reviewed some of this, but I have some questions. And also a lot of my comments are about rewording, etc. Let's chat tomorrow.
| /// <summary> | ||
| /// Settings available for subscriber shutdown. | ||
| /// </summary> | ||
| public enum SubscriberShutdownSetting |
There was a problem hiding this comment.
Consider ShutdownOption, mostly dropping subscription given that this is a type defined on the SubscriptionClient.
| /// Stops streaming new upstream messages and then continues processing all received messages. If there are | ||
| /// still messages that need to be processed 30s before the timeout is reached it will switch to <see cref="NackImmediately"/>. | ||
| /// </summary> |
There was a problem hiding this comment.
upstream is not a familiar term for Pub/Sub users as far as I'm aware. I think it's better to say, something like "stops the subscriber stream so no new messages are received".
Also, there's no timeout here, so referencing the timeout is unclear. The explanation re switching to nack inmediately is likely better suited for the method itself.
| WaitForProcessing = 0, | ||
|
|
||
| /// <summary> | ||
| /// Stops streaming new upstream messages and then aggressively releases unhandled messages by sending Nack responses. |
There was a problem hiding this comment.
Same comment re: upstream. And avoid using aggressively. Just say that it nacks all unhandled messages.
|
|
||
| /// <summary> | ||
| /// Stops streaming new upstream messages and then aggressively releases unhandled messages by sending Nack responses. | ||
| /// Already handled messages will still be acknowledged. |
There was a problem hiding this comment.
The client app response to a handle message may have been nack, so here something like "the ack/nack will be sent".
And are we certain about this?
|
|
||
| /// <summary> | ||
| /// Stop this <see cref="SubscriberClient"/>. | ||
| /// The returned <see cref="Task"/> completes when all handled messages have been acknowledged. |
There was a problem hiding this comment.
Again, or nacked.
But also, this is not true, it depends on what the settings are.
| _globalNackImmediatelyCts = CancellationTokenSource.CreateLinkedTokenSource(_globalHardStopCts.Token); | ||
| _globalWaitForProcessingCts = CancellationTokenSource.CreateLinkedTokenSource(_globalNackImmediatelyCts.Token, _globalSoftStopCts.Token); |
There was a problem hiding this comment.
I think we might indeed need a paragraph or two about the token sources, and specifically why we can't reuse the existing ones and what is the relationship between the existing ones and the new ones.
There was a problem hiding this comment.
And also, can we start using the new ones which are the ones that will remain after the deprecation and eventual deletion of the current method, and make the old ones depend on the new ones?
| { | ||
| // Note: If multiple stop requests are made, only the first cancellation token is observed. | ||
| if (_mainTcs is not null && _globalSoftStopCts.IsCancellationRequested) | ||
| if (_mainTcs is not null && _isStopInitiated) |
There was a problem hiding this comment.
There's a change here, IsStopInitiated is looking at the new globalWaitForProcessingCts
| _globalWaitForProcessingCts.Cancel(); | ||
| } | ||
|
|
||
| TimeSpan shutdownTimeout = timeout ?? DefaultMaxTotalAckExtension; |
There was a problem hiding this comment.
I think this should be _maxExtensionDuration
| // Delay, then start the streaming-pull. | ||
| _logger?.LogDebug("Client {index} delaying for {seconds}s before streaming pull call.", _clientIndex, (int) backoff.TotalSeconds); | ||
| Task delayTask = _scheduler.Delay(backoff, _softStopCts.Token); | ||
| Task delayTask = _scheduler.Delay(backoff, _waitForProcessingCts.Token); |
There was a problem hiding this comment.
For all of these I still don't fully understand the relationship between the new and old tokens. A paragraph on the design should be enough.
| @@ -501,12 +507,15 @@ private async Task ProcessPullMessagesAsync(List<ReceivedMessage> msgs, HashSet< | |||
| // Running async. Common data needs locking | |||
There was a problem hiding this comment.
Should you have changed the token in line 455?
2 participants
b/427314526