hubcio commented on code in PR #3913:
URL: https://github.com/apache/iggy/pull/3913#discussion_r3806083379
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -1128,36 +1588,66 @@ impl Stream for IggyConsumer {
}
impl IggyConsumer {
+ /// Shuts the consumer down.
+ ///
+ /// Specifically, run shutdown and await before dropping the consumer to
+ /// - finish storing the offsets that are currently in-flight.
+ /// There are two background tasks that can have commits in flight. The
interval-based one
+ /// (only spawned for [`AutoCommit`] variants that carry an interval)
and the one driven by
+ /// [`AutoCommitWhen`]/[`AutoCommitAfter`] (always spawned). The
consumer waits for
+ /// `offset_drain_timeout` on each in turn before forcing it to abort.
+ /// Any offset that is not stored until then will be lost.
+ /// - commit every offset from partitions where the consumed offset is
ahead of the stored one.
+ /// Note, this happens even under [`AutoCommit::Disabled`].
+ /// - leave the consumer group, if this consumer is a group member. This
lets the server give its partitions to
+ /// the remaining members immediately instead of waiting for the
connection to time out.
+ ///
+ /// # Errors
+ ///
+ /// Returns `Ok(())` even when the final commits or the group leave
failed, since those
+ /// failures are logged and do not leave anything for the caller to undo.
The
+ /// [`Result`] is part of the signature for forward compatibility.
pub async fn shutdown(&mut self) -> Result<(), IggyError> {
+ // Immediately return, if the consumer is already shut down.
+ // Otherwise, swap so background tasks see that the consumer got shut
down.
if self.shutdown.swap(true, ORDERING) {
return Ok(());
}
info!("Shutting down consumer: {}...", self.consumer_name);
- // Drain the background commit tasks while still a group member,
- // before leaving below — otherwise a store they send afterward hits
- // a group we've already left.
+ // Wake the task responsible for storing the offsets (spawned in
store_offset_in_background())
Review Comment:
fair, order hold today but nothing enforces it. leave it as is
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
