haubur commented on code in PR #3913:
URL: https://github.com/apache/iggy/pull/3913#discussion_r3806037498
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -297,16 +639,79 @@ impl IggyConsumer {
.await
}
- /// Retrieves the last stored offset (on the server) for the specified
partition ID.
- /// To get the current partition ID use `partition_id()`
+ /// Returns the offset this consumer last stored on the server for the
given partition, or
+ /// `None` if it has not stored one yet.
+ ///
+ /// The value is this consumer's own record of what it committed, kept in
memory rather than
+ /// read back from the server.
pub fn get_last_stored_offset(&self, partition_id: u32) -> Option<u64> {
let offset = self.last_stored_offsets.get(&partition_id)?;
Some(offset.load(ORDERING))
}
- /// Initializes the consumer by subscribing to diagnostic events,
initializing the consumer group if needed, storing the offsets in the
background etc.
+ /// Initializes the consumer and makes it ready to poll messages.
+ ///
+ /// This must be called before the consumer can start polling messages.
Calling it again on an
+ /// initialized consumer does nothing and returns immediately.
+ ///
+ /// Initialization ensures that:
+ /// - the consumers `stream_id` and `topic_id` exist on the server.
+ /// It retries for a number of `init_retries` (defaults to `None`, which
is treated as no
+ /// retry) with `init_retry_interval` (defaults to one
+ /// second) time in between retries. Both can be set together through
+ ///
[`IggyConsumerBuilder::init_retries`](crate::clients::consumer_builder::IggyConsumerBuilder::init_retries).
+ /// - the consumer subscribes to connection lifecycle events
([`DiagnosticEvent`]) in order to
+ /// update its state, should it receive a shutdown, connected,
disconnected, log in or log out event.
+ /// - if the consumer belongs to a group and `auto_join_consumer_group` is
enabled, the group is
+ /// initialized if it does not exist yet, and the consumer joins that
group.
+ /// - the tasks that store the offset on the server are spawned.
+ ///
+ /// # Lifecycle events
+ ///
+ /// Calling init spawns a background tasks that listens for lifecycle
changes ([`DiagnosticEvent`]s) of the
+ /// client connection.
+ /// - [`DiagnosticEvent::Connected`]: a fresh connection has not joined
anything yet.
+ /// Polling resumes immediately only for a consumer that is not a group
member.
+ /// - [`DiagnosticEvent::SignedIn`]: re-enables polling. A group member
signing in after a
+ /// reconnect rejoins its group first and only polls once that
succeeded. A failed rejoin is
+ /// logged and leaves polling disabled until the next event.
+ /// - [`DiagnosticEvent::Disconnected`] and [`DiagnosticEvent::SignedOut`]
disables polling.
+ /// - [`DiagnosticEvent::Shutdown`] disables polling and terminates the
background task listening
+ /// for lifecycle changes. It does not flush in-flight commits; that
only happens when
+ /// [`shutdown()`](Self::shutdown) itself is called.
+ ///
+ /// # Storing offsets
///
- /// Note: This method must be called before polling messages.
+ /// An offset is the position of a message within a partition, and storing
one tells the server
+ /// how many this consumer (or its consumer group) has consumed already.
+ /// When this offset is stored at the server is configured in
`auto_commit`, which defaults to
+ /// [`AutoCommit::IntervalOrWhen`] equal to 1s and
[`AutoCommitWhen::PollingMessages`].
+ /// - An interval background task is only spawned for the variants that
carry an interval
+ /// ([`AutoCommit::Interval`], [`AutoCommit::IntervalOrWhen`],
[`AutoCommit::IntervalOrAfter`]).
+ /// - The offset store task is spawned in any case. It can be configured
with [`AutoCommitWhen::ConsumingEachMessage`],
+ /// [`AutoCommitWhen::ConsumingEveryNthMessage`],
[`AutoCommitWhen::ConsumingAllMessages`] and
+ /// their [`AutoCommitAfter`] counterparts. Under
[`AutoCommit::Disabled`] nothing is
+ /// ever sent and the task stays idle.
+ ///
+ /// A variant such as [`AutoCommit::IntervalOrWhen`] runs both together.
The message count
+ /// trigger stores as messages are consumed, the interval stores what the
trigger has not
+ /// covered yet. There is no double-work, since an offset that is not
ahead of the one last stored
+ /// for that partition is skipped instead of sent.
+ /// Unless `allow_replay` is enabled an offset that is not past the last
stored offset on the server
Review Comment:
Indeed, missed that it is hard coded for AutoCommit.
--
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]
