hubcio commented on code in PR #3913:
URL: https://github.com/apache/iggy/pull/3913#discussion_r3804116395
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -100,6 +100,316 @@ pub enum AutoCommitAfter {
// 4. All `&self` methods only access Sync-safe fields
unsafe impl Sync for IggyConsumer {}
+/// Reads messages from the partitions of one topic and yields them one at a
time.
+///
+/// A topic is split into partitions, and a partition is an ordered log that
producers append to.
+/// Every message sits at an *offset*, its position in that log. Reading is
therefore always the
+/// same three decisions: which partition to read, where in it to start, and
how to keep track of
+/// how far you got so the next run can continue there.
+///
+/// `IggyConsumer` handles all three. It fetches batches of messages from the
server, keeps them in
+/// an in-memory buffer, decrypts them when the client is configured with an
encryptor, and records
+/// how far it has read. **It implements [`Stream`], so consuming is a loop
over [`StreamExt::next`].**
+///
+/// You can use a consumer as a worker draining a topic, a reader that replays
a
+/// partition from a chosen point, and a pool of consumers sharing a workload
through a consumer
+/// group.
+///
+/// # Creating a consumer
+///
+/// Easiest way is to use the [`IggyClient`] with a configured connection.
Then:
+/// - [`IggyClient::consumer()`] builds a standalone consumer, bound to the
one partition passed
+/// in.
+/// - [`IggyClient::consumer_group()`] builds a member of a consumer group.
The server gives every
+/// partition to exactly one member, so several consumers using the same
group name split the
+/// topic between them and share one set of offsets.
+///
+/// Note, building never talks to the server. [`init()`](Self::init) must be
awaited once before the
+/// first message is read.
+///
+/// # Examples
+///
+/// A standalone consumer reading partition 1 with the defaults:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer("my-consumer", "my-stream", "my-topic", 1)?
+/// .batch_length(100)
+/// .poll_interval(IggyDuration::new_from_secs(1))
+/// .build();
+/// consumer.init().await?;
+///
+/// while let Some(received) = consumer.next().await {
+/// match received {
+/// Ok(received) => println!("Offset: {}",
received.message.header.offset),
+/// Err(error) => eprintln!("Failed to read a message: {error}"),
+/// }
+/// }
+/// # Ok(())
+/// # }
+/// ```
+///
+/// A group member that commits every message right after it is handed over,
and shuts down
+/// cleanly:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn handle(message: &IggyMessage) {}
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer_group("order-workers", "my-stream", "my-topic")?
+/// .auto_commit(AutoCommit::When(AutoCommitWhen::ConsumingEachMessage))
+/// .polling_strategy(PollingStrategy::next())
+/// .build();
+/// consumer.init().await?;
+///
+/// let mut consumed = 0;
+/// while let Some(received) = consumer.next().await {
+/// handle(&received?.message).await;
+/// consumed += 1;
+/// if consumed == 100 {
+/// break;
+/// }
+/// }
+///
+/// consumer.shutdown().await?;
+/// # Ok(())
+/// # }
+/// ```
+///
+/// Committing by hand, so that a message the handler could not process comes
back:
Review Comment:
the example doesn't deliver this promise - if message N fails but N+1
succeeds, `store_offset(N+1)` marks N consumed (it's a cumulative watermark),
so N never comes back, in-process or after restart. it only works if you stop
on failure. either break on error or call out the cumulative-commit caveat.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -100,6 +100,316 @@ pub enum AutoCommitAfter {
// 4. All `&self` methods only access Sync-safe fields
unsafe impl Sync for IggyConsumer {}
+/// Reads messages from the partitions of one topic and yields them one at a
time.
+///
+/// A topic is split into partitions, and a partition is an ordered log that
producers append to.
+/// Every message sits at an *offset*, its position in that log. Reading is
therefore always the
+/// same three decisions: which partition to read, where in it to start, and
how to keep track of
+/// how far you got so the next run can continue there.
+///
+/// `IggyConsumer` handles all three. It fetches batches of messages from the
server, keeps them in
+/// an in-memory buffer, decrypts them when the client is configured with an
encryptor, and records
+/// how far it has read. **It implements [`Stream`], so consuming is a loop
over [`StreamExt::next`].**
+///
+/// You can use a consumer as a worker draining a topic, a reader that replays
a
+/// partition from a chosen point, and a pool of consumers sharing a workload
through a consumer
+/// group.
+///
+/// # Creating a consumer
+///
+/// Easiest way is to use the [`IggyClient`] with a configured connection.
Then:
+/// - [`IggyClient::consumer()`] builds a standalone consumer, bound to the
one partition passed
+/// in.
+/// - [`IggyClient::consumer_group()`] builds a member of a consumer group.
The server gives every
+/// partition to exactly one member, so several consumers using the same
group name split the
+/// topic between them and share one set of offsets.
+///
+/// Note, building never talks to the server. [`init()`](Self::init) must be
awaited once before the
+/// first message is read.
+///
+/// # Examples
+///
+/// A standalone consumer reading partition 1 with the defaults:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer("my-consumer", "my-stream", "my-topic", 1)?
+/// .batch_length(100)
+/// .poll_interval(IggyDuration::new_from_secs(1))
+/// .build();
+/// consumer.init().await?;
+///
+/// while let Some(received) = consumer.next().await {
+/// match received {
+/// Ok(received) => println!("Offset: {}",
received.message.header.offset),
+/// Err(error) => eprintln!("Failed to read a message: {error}"),
+/// }
+/// }
+/// # Ok(())
+/// # }
+/// ```
+///
+/// A group member that commits every message right after it is handed over,
and shuts down
+/// cleanly:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn handle(message: &IggyMessage) {}
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer_group("order-workers", "my-stream", "my-topic")?
+/// .auto_commit(AutoCommit::When(AutoCommitWhen::ConsumingEachMessage))
+/// .polling_strategy(PollingStrategy::next())
+/// .build();
+/// consumer.init().await?;
+///
+/// let mut consumed = 0;
+/// while let Some(received) = consumer.next().await {
+/// handle(&received?.message).await;
+/// consumed += 1;
+/// if consumed == 100 {
+/// break;
+/// }
+/// }
+///
+/// consumer.shutdown().await?;
+/// # Ok(())
+/// # }
+/// ```
+///
+/// Committing by hand, so that a message the handler could not process comes
back:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn handle(message: &IggyMessage) -> Result<(), IggyError> { Ok(())
}
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer("my-consumer", "my-stream", "my-topic", 1)?
+/// .auto_commit(AutoCommit::Disabled)
+/// .polling_strategy(PollingStrategy::next())
+/// // Without this, messages already handed over once are never handed
over again.
+/// .allow_replay()
+/// .build();
+/// consumer.init().await?;
+///
+/// while let Some(received) = consumer.next().await {
+/// let received = received?;
+/// if handle(&received.message).await.is_ok() {
+/// consumer
+/// .store_offset(received.message.header.offset,
Some(received.partition_id))
+/// .await?;
+/// }
+/// }
+/// # Ok(())
+/// # }
+/// ```
+///
+/// # Which partitions are read
+///
+/// A **standalone consumer** reads exactly one partition, the one passed to
+/// [`IggyClient::consumer()`]. Covering a whole topic with several partitions
+/// means running one consumer per partition and dividing the work yourself.
+///
+/// A **consumer group member** does not choose. The server hands every
partition of the topic to
+/// exactly one member, so consumers sharing a group name split the topic
between them without
+/// coordinating. [`partition_id()`](Self::partition_id) reports where the
last message came from.
+///
+/// What to know when working with consumer groups:
+/// - A member joins during [`init()`](Self::init), creating the group first if
+/// [`create_consumer_group_if_not_exists()`] is set (the default). It
rejoins on its own after a
+/// reconnect and whenever the server reports that its membership is gone.
+/// - Until the join has succeeded the consumer does not poll. It waits for
+/// [`polling_retry_interval()`] and tries again.
+/// - Partitions are redistributed whenever members join or leave, so a member
reads different
+/// partitions over time and messages from several partitions interleave in
its stream.
+/// - More members than partitions leaves the surplus members idle. The
partition count of the
+/// topic is the ceiling on how far one group can be scaled out.
+/// - The group shares one set of stored offsets, kept under the group name.
Thus,
+/// a partition taken over by another member continues where the previous one
+/// committed.
+///
+/// # How messages are read
+///
+/// Reading is done by polling. One request fetches up to [`batch_length()`]
messages. The consumer
+/// passes the first one to the caller and buffers the rest. The next request
is sent once that buffer
+/// is empty.
+///
+/// [`poll_interval()`] sets a timeout between two requests should the buffer
be empty.
+/// Without it the next request goes out as soon as the previous one is
+/// answered, which is the fastest option but keeps a busy loop running
against an idle topic.
+///
+/// [`polling_strategy()`] decides **where** in the partition reading begins:
+///
+/// | Strategy | Starts at |
+/// | --- | --- |
+/// | [`PollingStrategy::next()`] (default) | the message after the offset
that is stored on the server |
+/// | [`PollingStrategy::first()`] | the oldest message in the partition |
+/// | [`PollingStrategy::last()`] | the end of the partition (returns up to
[`batch_length()`] of the most recent messages) |
+/// | [`PollingStrategy::offset()`] | a custom offset |
+/// | [`PollingStrategy::timestamp()`] | the first message at or after a given
point in time |
+///
+/// Only [`PollingStrategy::next()`] consults the offset stored on the server.
+/// Use this if you want to resume where a previous run stopped. The other
four are starting points for the first request only.
+/// From the second request onwards, the consumer asks for whatever follows it.
+///
+/// Note, when polling, [`StreamExt::next`] never returns `None`, not when the
topic is empty
+/// and not while the client is disconnected. A `while let Some(..)` loop only
ends when the
+/// loop body breaks out of it. A request that comes back empty is not an
error and not the end of
+/// the stream, it just means nothing new has arrived yet.
+///
+/// A failed request is yielded as `Some(Err(..))` and leaves the consumer
usable, while the next call
+/// retries. Connection and authentication failures pause polling until the
client has reconnected
+/// and signed in again, which the consumer handles automatically. Hence,
deciding when to give up on
+/// repeated errors is up to you.
+///
+/// For a boilerplate implementation of such a loop Iggy provides
[`IggyConsumerMessageExt::consume_messages`].
+///
+/// # Tracking what has been read
+///
+/// An offset is the index tracking what has been already read from a
partition by the consumer.
+/// Managing the offset has implications on where consumers resume reading
messages.
+///
+/// There are two positions (offsets) tracked in two different places:
+/// - The **reading position** is held by the consumer, one per partition, and
is the offset of the
+/// last message handed over
+/// ([`get_last_consumed_offset()`](Self::get_last_consumed_offset)). It
dies with the process.
+/// - The **stored offset** lives on the server under the consumer name, or
the group name for a
+/// group. This offset survives restarts. Writing it is called *storing* or
*committing* an offset.
+///
+/// Committing matters because [`PollingStrategy::next()`] resumes from the
stored offset. A
+/// consumer that never commits keeps starting over from the same place.
+/// Note, *committing* is a request of its own, not a side effect of reading
and therefore controllable.
+///
+/// [`auto_commit()`] decides when the consumer commits by itself:
+///
+/// | Setting | Commits |
+/// | --- | --- |
+/// | [`AutoCommit::Disabled`] | never on its own, decide manually with
[`store_offset()`](Self::store_offset) or on [`shutdown()`](Self::shutdown) |
+/// | [`AutoCommit::Interval`] | on every tick, for every partition read so
far |
+/// | [`AutoCommitWhen::PollingMessages`] | with the poll request itself,
before your code sees the batch |
+/// | [`AutoCommitWhen::ConsumingEachMessage`] | after every message was
handed over to the calling code |
+/// | [`AutoCommitWhen::ConsumingEveryNthMessage`] | when the offset of a
message handed over divides by `n` |
+/// | [`AutoCommitWhen::ConsumingAllMessages`] | when the buffer of the
current batch runs empty |
+/// | [`AutoCommitAfter`] variants | as their [`AutoCommitWhen`] counterparts,
but once the handler returned, and only under
[`IggyConsumerMessageExt::consume_messages`] |
+///
+/// [`AutoCommit::IntervalOrWhen`] and [`AutoCommit::IntervalOrAfter`] combine
an interval with a
+/// message trigger. The default is [`AutoCommit::IntervalOrWhen`] with one
second and
+/// [`AutoCommitWhen::PollingMessages`].
+/// Important implications of these defaults:
+/// - [`AutoCommitWhen::PollingMessages`] marks a batch as consumed while it
is being delivered,
+/// before your code has seen any of it. If a crash must not skip messages,
commit after handling
+/// with [`AutoCommitWhen::ConsumingEachMessage`] instead.
+/// - [`AutoCommitWhen::ConsumingEveryNthMessage`] tests the offset of a
message, not a counter of
+/// messages this process handled, so it commits at every `n`-th offset of
the partition.
+///
+/// ## Guarantees
+///
+/// - **Each message is handed over once per consumer.** Messages whose offset
is not greater than
+/// the reading position of their partition are dropped before they reach
the stream. Re-reading
+/// a partition, or letting a message come back because your handler failed,
needs
+/// [`allow_replay()`], which turns that filter off.
+/// - **Delivery is at-least-once.** A crash between handling a message and
committing its offset
Review Comment:
"at-least-once" isn't true for the default config - `PollingMessages`
commits the whole batch as part of the poll request, before your code sees any
of it, so a crash mid-batch skips the rest. same for `ConsumingEachMessage`
(see above). at-least-once only holds when committing after handling - worth
qualifying, since this is the guarantees section.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -100,6 +100,316 @@ pub enum AutoCommitAfter {
// 4. All `&self` methods only access Sync-safe fields
unsafe impl Sync for IggyConsumer {}
+/// Reads messages from the partitions of one topic and yields them one at a
time.
+///
+/// A topic is split into partitions, and a partition is an ordered log that
producers append to.
+/// Every message sits at an *offset*, its position in that log. Reading is
therefore always the
+/// same three decisions: which partition to read, where in it to start, and
how to keep track of
+/// how far you got so the next run can continue there.
+///
+/// `IggyConsumer` handles all three. It fetches batches of messages from the
server, keeps them in
+/// an in-memory buffer, decrypts them when the client is configured with an
encryptor, and records
+/// how far it has read. **It implements [`Stream`], so consuming is a loop
over [`StreamExt::next`].**
+///
+/// You can use a consumer as a worker draining a topic, a reader that replays
a
+/// partition from a chosen point, and a pool of consumers sharing a workload
through a consumer
+/// group.
+///
+/// # Creating a consumer
+///
+/// Easiest way is to use the [`IggyClient`] with a configured connection.
Then:
+/// - [`IggyClient::consumer()`] builds a standalone consumer, bound to the
one partition passed
+/// in.
+/// - [`IggyClient::consumer_group()`] builds a member of a consumer group.
The server gives every
+/// partition to exactly one member, so several consumers using the same
group name split the
+/// topic between them and share one set of offsets.
+///
+/// Note, building never talks to the server. [`init()`](Self::init) must be
awaited once before the
+/// first message is read.
+///
+/// # Examples
+///
+/// A standalone consumer reading partition 1 with the defaults:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer("my-consumer", "my-stream", "my-topic", 1)?
+/// .batch_length(100)
+/// .poll_interval(IggyDuration::new_from_secs(1))
+/// .build();
+/// consumer.init().await?;
+///
+/// while let Some(received) = consumer.next().await {
+/// match received {
+/// Ok(received) => println!("Offset: {}",
received.message.header.offset),
+/// Err(error) => eprintln!("Failed to read a message: {error}"),
+/// }
+/// }
+/// # Ok(())
+/// # }
+/// ```
+///
+/// A group member that commits every message right after it is handed over,
and shuts down
+/// cleanly:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn handle(message: &IggyMessage) {}
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer_group("order-workers", "my-stream", "my-topic")?
+/// .auto_commit(AutoCommit::When(AutoCommitWhen::ConsumingEachMessage))
+/// .polling_strategy(PollingStrategy::next())
+/// .build();
+/// consumer.init().await?;
+///
+/// let mut consumed = 0;
+/// while let Some(received) = consumer.next().await {
+/// handle(&received?.message).await;
+/// consumed += 1;
+/// if consumed == 100 {
+/// break;
+/// }
+/// }
+///
+/// consumer.shutdown().await?;
+/// # Ok(())
+/// # }
+/// ```
+///
+/// Committing by hand, so that a message the handler could not process comes
back:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn handle(message: &IggyMessage) -> Result<(), IggyError> { Ok(())
}
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer("my-consumer", "my-stream", "my-topic", 1)?
+/// .auto_commit(AutoCommit::Disabled)
+/// .polling_strategy(PollingStrategy::next())
+/// // Without this, messages already handed over once are never handed
over again.
+/// .allow_replay()
+/// .build();
+/// consumer.init().await?;
+///
+/// while let Some(received) = consumer.next().await {
+/// let received = received?;
+/// if handle(&received.message).await.is_ok() {
+/// consumer
+/// .store_offset(received.message.header.offset,
Some(received.partition_id))
+/// .await?;
+/// }
+/// }
+/// # Ok(())
+/// # }
+/// ```
+///
+/// # Which partitions are read
+///
+/// A **standalone consumer** reads exactly one partition, the one passed to
+/// [`IggyClient::consumer()`]. Covering a whole topic with several partitions
+/// means running one consumer per partition and dividing the work yourself.
+///
+/// A **consumer group member** does not choose. The server hands every
partition of the topic to
+/// exactly one member, so consumers sharing a group name split the topic
between them without
+/// coordinating. [`partition_id()`](Self::partition_id) reports where the
last message came from.
+///
+/// What to know when working with consumer groups:
+/// - A member joins during [`init()`](Self::init), creating the group first if
+/// [`create_consumer_group_if_not_exists()`] is set (the default). It
rejoins on its own after a
+/// reconnect and whenever the server reports that its membership is gone.
+/// - Until the join has succeeded the consumer does not poll. It waits for
+/// [`polling_retry_interval()`] and tries again.
+/// - Partitions are redistributed whenever members join or leave, so a member
reads different
+/// partitions over time and messages from several partitions interleave in
its stream.
+/// - More members than partitions leaves the surplus members idle. The
partition count of the
+/// topic is the ceiling on how far one group can be scaled out.
+/// - The group shares one set of stored offsets, kept under the group name.
Thus,
+/// a partition taken over by another member continues where the previous one
+/// committed.
+///
+/// # How messages are read
+///
+/// Reading is done by polling. One request fetches up to [`batch_length()`]
messages. The consumer
+/// passes the first one to the caller and buffers the rest. The next request
is sent once that buffer
+/// is empty.
+///
+/// [`poll_interval()`] sets a timeout between two requests should the buffer
be empty.
+/// Without it the next request goes out as soon as the previous one is
+/// answered, which is the fastest option but keeps a busy loop running
against an idle topic.
+///
+/// [`polling_strategy()`] decides **where** in the partition reading begins:
+///
+/// | Strategy | Starts at |
+/// | --- | --- |
+/// | [`PollingStrategy::next()`] (default) | the message after the offset
that is stored on the server |
+/// | [`PollingStrategy::first()`] | the oldest message in the partition |
+/// | [`PollingStrategy::last()`] | the end of the partition (returns up to
[`batch_length()`] of the most recent messages) |
+/// | [`PollingStrategy::offset()`] | a custom offset |
+/// | [`PollingStrategy::timestamp()`] | the first message at or after a given
point in time |
+///
+/// Only [`PollingStrategy::next()`] consults the offset stored on the server.
+/// Use this if you want to resume where a previous run stopped. The other
four are starting points for the first request only.
+/// From the second request onwards, the consumer asks for whatever follows it.
+///
+/// Note, when polling, [`StreamExt::next`] never returns `None`, not when the
topic is empty
+/// and not while the client is disconnected. A `while let Some(..)` loop only
ends when the
+/// loop body breaks out of it. A request that comes back empty is not an
error and not the end of
+/// the stream, it just means nothing new has arrived yet.
+///
+/// A failed request is yielded as `Some(Err(..))` and leaves the consumer
usable, while the next call
+/// retries. Connection and authentication failures pause polling until the
client has reconnected
+/// and signed in again, which the consumer handles automatically. Hence,
deciding when to give up on
+/// repeated errors is up to you.
+///
+/// For a boilerplate implementation of such a loop Iggy provides
[`IggyConsumerMessageExt::consume_messages`].
+///
+/// # Tracking what has been read
+///
+/// An offset is the index tracking what has been already read from a
partition by the consumer.
+/// Managing the offset has implications on where consumers resume reading
messages.
+///
+/// There are two positions (offsets) tracked in two different places:
+/// - The **reading position** is held by the consumer, one per partition, and
is the offset of the
+/// last message handed over
+/// ([`get_last_consumed_offset()`](Self::get_last_consumed_offset)). It
dies with the process.
+/// - The **stored offset** lives on the server under the consumer name, or
the group name for a
+/// group. This offset survives restarts. Writing it is called *storing* or
*committing* an offset.
+///
+/// Committing matters because [`PollingStrategy::next()`] resumes from the
stored offset. A
+/// consumer that never commits keeps starting over from the same place.
+/// Note, *committing* is a request of its own, not a side effect of reading
and therefore controllable.
+///
+/// [`auto_commit()`] decides when the consumer commits by itself:
+///
+/// | Setting | Commits |
+/// | --- | --- |
+/// | [`AutoCommit::Disabled`] | never on its own, decide manually with
[`store_offset()`](Self::store_offset) or on [`shutdown()`](Self::shutdown) |
+/// | [`AutoCommit::Interval`] | on every tick, for every partition read so
far |
+/// | [`AutoCommitWhen::PollingMessages`] | with the poll request itself,
before your code sees the batch |
+/// | [`AutoCommitWhen::ConsumingEachMessage`] | after every message was
handed over to the calling code |
+/// | [`AutoCommitWhen::ConsumingEveryNthMessage`] | when the offset of a
message handed over divides by `n` |
+/// | [`AutoCommitWhen::ConsumingAllMessages`] | when the buffer of the
current batch runs empty |
+/// | [`AutoCommitAfter`] variants | as their [`AutoCommitWhen`] counterparts,
but once the handler returned, and only under
[`IggyConsumerMessageExt::consume_messages`] |
+///
+/// [`AutoCommit::IntervalOrWhen`] and [`AutoCommit::IntervalOrAfter`] combine
an interval with a
+/// message trigger. The default is [`AutoCommit::IntervalOrWhen`] with one
second and
+/// [`AutoCommitWhen::PollingMessages`].
+/// Important implications of these defaults:
+/// - [`AutoCommitWhen::PollingMessages`] marks a batch as consumed while it
is being delivered,
+/// before your code has seen any of it. If a crash must not skip messages,
commit after handling
Review Comment:
this advice doesn't hold - `ConsumingEachMessage` commits at handover,
inside `poll_next` before your handler runs (the store fires at 1442-1447 /
1557-1567 and lands concurrently with the handler). a crash mid-handling can
still skip the message. crash-safe options are `AutoCommit::Disabled` +
`store_offset()` after handling, or the `AutoCommitAfter` variants under
`consume_messages`.
##########
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:
`allow_replay` doesn't affect the auto-commit paths at all - both background
tasks hardcode it to `false` (lines 840 and 940). it only applies to manual
`store_offset()` and the shutdown flush. also the skip check reads the
in-memory `last_stored_offsets`, not the server.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -404,6 +821,9 @@ impl IggyConsumer {
let (store_offset_sender, store_offset_receiver) = flume::unbounded();
self.store_offset_sender = store_offset_sender;
+ // The IggyClients `poll_next` implementation sends store offset
requests down to this receiver.
Review Comment:
`poll_next` is `IggyConsumer`'s `Stream` impl, not `IggyClient`'s.
`consume_messages` also sends into this channel.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -1229,6 +1724,10 @@ impl IggyConsumer {
}
}
+/// Stops the background tasks, nothing more.
+///
+/// Dropping cannot await, so it neither commits pending offsets nor leaves
the consumer group.
Review Comment:
two edges off here: the diagnostics listener task isn't stopped by drop (it
never checks the shutdown flag, only exits on the `Shutdown` event or channel
close), and the store-offset task may still commit queued offsets after drop
since the channel drains before closing.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -229,26 +539,48 @@ impl IggyConsumer {
}
/// Returns the name of the consumer.
+ ///
+ /// For a consumer group this is also the name of the group.
pub fn name(&self) -> &str {
&self.consumer_name
}
- /// Returns the topic ID of the consumer.
+ /// Returns the identifier of the topic this consumer reads from.
pub fn topic(&self) -> &Identifier {
&self.topic_id
}
- /// Returns the stream ID of the consumer.
+ /// Returns the identifier of the stream this consumer reads from.
pub fn stream(&self) -> &Identifier {
&self.stream_id
}
- /// Returns the current partition ID of the consumer.
+ /// Returns the partition the last message came from.
+ ///
+ /// This is `0` until the first message has been read, because a partition
is only known once
+ /// the server has answered. For a consumer group the value changes over
time, as the server
+ /// can hand different partitions to this member.
pub fn partition_id(&self) -> u32 {
self.current_partition_id.load(ORDERING)
}
- /// Stores the consumer offset on the server either for the current
partition or the provided partition ID.
+ /// Stores an offset on the server, marking every message up to and
including it as consumed.
+ ///
+ /// This is the manual counterpart to [`AutoCommit`] and is meant for
+ /// [`AutoCommit::Disabled`].
+ ///
+ /// Pass `None` as `partition_id` to use the partition of the most recent
batch polled.
+ ///
+ /// An offset that is not ahead of the last one this consumer stored for
that partition is
Review Comment:
small carve-out: the skip check is `offset <= stored_offset && offset >= 1`
(line 876), so offset 0 is always sent. and the server allows an explicit store
to rewind the watermark, so `store_offset(0)` actually rewinds even without
`allow_replay`.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -100,6 +100,316 @@ pub enum AutoCommitAfter {
// 4. All `&self` methods only access Sync-safe fields
unsafe impl Sync for IggyConsumer {}
+/// Reads messages from the partitions of one topic and yields them one at a
time.
+///
+/// A topic is split into partitions, and a partition is an ordered log that
producers append to.
+/// Every message sits at an *offset*, its position in that log. Reading is
therefore always the
+/// same three decisions: which partition to read, where in it to start, and
how to keep track of
+/// how far you got so the next run can continue there.
+///
+/// `IggyConsumer` handles all three. It fetches batches of messages from the
server, keeps them in
+/// an in-memory buffer, decrypts them when the client is configured with an
encryptor, and records
+/// how far it has read. **It implements [`Stream`], so consuming is a loop
over [`StreamExt::next`].**
+///
+/// You can use a consumer as a worker draining a topic, a reader that replays
a
+/// partition from a chosen point, and a pool of consumers sharing a workload
through a consumer
+/// group.
+///
+/// # Creating a consumer
+///
+/// Easiest way is to use the [`IggyClient`] with a configured connection.
Then:
+/// - [`IggyClient::consumer()`] builds a standalone consumer, bound to the
one partition passed
+/// in.
+/// - [`IggyClient::consumer_group()`] builds a member of a consumer group.
The server gives every
+/// partition to exactly one member, so several consumers using the same
group name split the
+/// topic between them and share one set of offsets.
+///
+/// Note, building never talks to the server. [`init()`](Self::init) must be
awaited once before the
+/// first message is read.
+///
+/// # Examples
+///
+/// A standalone consumer reading partition 1 with the defaults:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer("my-consumer", "my-stream", "my-topic", 1)?
+/// .batch_length(100)
+/// .poll_interval(IggyDuration::new_from_secs(1))
+/// .build();
+/// consumer.init().await?;
+///
+/// while let Some(received) = consumer.next().await {
+/// match received {
+/// Ok(received) => println!("Offset: {}",
received.message.header.offset),
+/// Err(error) => eprintln!("Failed to read a message: {error}"),
+/// }
+/// }
+/// # Ok(())
+/// # }
+/// ```
+///
+/// A group member that commits every message right after it is handed over,
and shuts down
+/// cleanly:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn handle(message: &IggyMessage) {}
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer_group("order-workers", "my-stream", "my-topic")?
+/// .auto_commit(AutoCommit::When(AutoCommitWhen::ConsumingEachMessage))
+/// .polling_strategy(PollingStrategy::next())
+/// .build();
+/// consumer.init().await?;
+///
+/// let mut consumed = 0;
+/// while let Some(received) = consumer.next().await {
+/// handle(&received?.message).await;
+/// consumed += 1;
+/// if consumed == 100 {
+/// break;
+/// }
+/// }
+///
+/// consumer.shutdown().await?;
+/// # Ok(())
+/// # }
+/// ```
+///
+/// Committing by hand, so that a message the handler could not process comes
back:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn handle(message: &IggyMessage) -> Result<(), IggyError> { Ok(())
}
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer("my-consumer", "my-stream", "my-topic", 1)?
+/// .auto_commit(AutoCommit::Disabled)
+/// .polling_strategy(PollingStrategy::next())
+/// // Without this, messages already handed over once are never handed
over again.
+/// .allow_replay()
+/// .build();
+/// consumer.init().await?;
+///
+/// while let Some(received) = consumer.next().await {
+/// let received = received?;
+/// if handle(&received.message).await.is_ok() {
+/// consumer
+/// .store_offset(received.message.header.offset,
Some(received.partition_id))
+/// .await?;
+/// }
+/// }
+/// # Ok(())
+/// # }
+/// ```
+///
+/// # Which partitions are read
+///
+/// A **standalone consumer** reads exactly one partition, the one passed to
+/// [`IggyClient::consumer()`]. Covering a whole topic with several partitions
+/// means running one consumer per partition and dividing the work yourself.
+///
+/// A **consumer group member** does not choose. The server hands every
partition of the topic to
+/// exactly one member, so consumers sharing a group name split the topic
between them without
+/// coordinating. [`partition_id()`](Self::partition_id) reports where the
last message came from.
+///
+/// What to know when working with consumer groups:
+/// - A member joins during [`init()`](Self::init), creating the group first if
+/// [`create_consumer_group_if_not_exists()`] is set (the default). It
rejoins on its own after a
+/// reconnect and whenever the server reports that its membership is gone.
+/// - Until the join has succeeded the consumer does not poll. It waits for
+/// [`polling_retry_interval()`] and tries again.
+/// - Partitions are redistributed whenever members join or leave, so a member
reads different
+/// partitions over time and messages from several partitions interleave in
its stream.
+/// - More members than partitions leaves the surplus members idle. The
partition count of the
+/// topic is the ceiling on how far one group can be scaled out.
+/// - The group shares one set of stored offsets, kept under the group name.
Thus,
+/// a partition taken over by another member continues where the previous one
+/// committed.
+///
+/// # How messages are read
+///
+/// Reading is done by polling. One request fetches up to [`batch_length()`]
messages. The consumer
+/// passes the first one to the caller and buffers the rest. The next request
is sent once that buffer
+/// is empty.
+///
+/// [`poll_interval()`] sets a timeout between two requests should the buffer
be empty.
+/// Without it the next request goes out as soon as the previous one is
+/// answered, which is the fastest option but keeps a busy loop running
against an idle topic.
+///
+/// [`polling_strategy()`] decides **where** in the partition reading begins:
+///
+/// | Strategy | Starts at |
+/// | --- | --- |
+/// | [`PollingStrategy::next()`] (default) | the message after the offset
that is stored on the server |
+/// | [`PollingStrategy::first()`] | the oldest message in the partition |
+/// | [`PollingStrategy::last()`] | the end of the partition (returns up to
[`batch_length()`] of the most recent messages) |
+/// | [`PollingStrategy::offset()`] | a custom offset |
+/// | [`PollingStrategy::timestamp()`] | the first message at or after a given
point in time |
+///
+/// Only [`PollingStrategy::next()`] consults the offset stored on the server.
+/// Use this if you want to resume where a previous run stopped. The other
four are starting points for the first request only.
+/// From the second request onwards, the consumer asks for whatever follows it.
+///
+/// Note, when polling, [`StreamExt::next`] never returns `None`, not when the
topic is empty
+/// and not while the client is disconnected. A `while let Some(..)` loop only
ends when the
+/// loop body breaks out of it. A request that comes back empty is not an
error and not the end of
+/// the stream, it just means nothing new has arrived yet.
+///
+/// A failed request is yielded as `Some(Err(..))` and leaves the consumer
usable, while the next call
+/// retries. Connection and authentication failures pause polling until the
client has reconnected
+/// and signed in again, which the consumer handles automatically. Hence,
deciding when to give up on
+/// repeated errors is up to you.
+///
+/// For a boilerplate implementation of such a loop Iggy provides
[`IggyConsumerMessageExt::consume_messages`].
+///
+/// # Tracking what has been read
+///
+/// An offset is the index tracking what has been already read from a
partition by the consumer.
+/// Managing the offset has implications on where consumers resume reading
messages.
+///
+/// There are two positions (offsets) tracked in two different places:
+/// - The **reading position** is held by the consumer, one per partition, and
is the offset of the
+/// last message handed over
+/// ([`get_last_consumed_offset()`](Self::get_last_consumed_offset)). It
dies with the process.
+/// - The **stored offset** lives on the server under the consumer name, or
the group name for a
+/// group. This offset survives restarts. Writing it is called *storing* or
*committing* an offset.
+///
+/// Committing matters because [`PollingStrategy::next()`] resumes from the
stored offset. A
+/// consumer that never commits keeps starting over from the same place.
+/// Note, *committing* is a request of its own, not a side effect of reading
and therefore controllable.
+///
+/// [`auto_commit()`] decides when the consumer commits by itself:
+///
+/// | Setting | Commits |
+/// | --- | --- |
+/// | [`AutoCommit::Disabled`] | never on its own, decide manually with
[`store_offset()`](Self::store_offset) or on [`shutdown()`](Self::shutdown) |
+/// | [`AutoCommit::Interval`] | on every tick, for every partition read so
far |
+/// | [`AutoCommitWhen::PollingMessages`] | with the poll request itself,
before your code sees the batch |
+/// | [`AutoCommitWhen::ConsumingEachMessage`] | after every message was
handed over to the calling code |
+/// | [`AutoCommitWhen::ConsumingEveryNthMessage`] | when the offset of a
message handed over divides by `n` |
+/// | [`AutoCommitWhen::ConsumingAllMessages`] | when the buffer of the
current batch runs empty |
+/// | [`AutoCommitAfter`] variants | as their [`AutoCommitWhen`] counterparts,
but once the handler returned, and only under
[`IggyConsumerMessageExt::consume_messages`] |
+///
+/// [`AutoCommit::IntervalOrWhen`] and [`AutoCommit::IntervalOrAfter`] combine
an interval with a
+/// message trigger. The default is [`AutoCommit::IntervalOrWhen`] with one
second and
+/// [`AutoCommitWhen::PollingMessages`].
+/// Important implications of these defaults:
+/// - [`AutoCommitWhen::PollingMessages`] marks a batch as consumed while it
is being delivered,
+/// before your code has seen any of it. If a crash must not skip messages,
commit after handling
+/// with [`AutoCommitWhen::ConsumingEachMessage`] instead.
+/// - [`AutoCommitWhen::ConsumingEveryNthMessage`] tests the offset of a
message, not a counter of
+/// messages this process handled, so it commits at every `n`-th offset of
the partition.
+///
+/// ## Guarantees
+///
+/// - **Each message is handed over once per consumer.** Messages whose offset
is not greater than
+/// the reading position of their partition are dropped before they reach
the stream. Re-reading
+/// a partition, or letting a message come back because your handler failed,
needs
+/// [`allow_replay()`], which turns that filter off.
+/// - **Delivery is at-least-once.** A crash between handling a message and
committing its offset
+/// replays that message on the next run, so handlers have to tolerate
seeing one twice. No
+/// setting makes this exactly-once.
+///
+/// # Options and defaults
+///
+/// Everything is configured on the [`IggyConsumerBuilder`] before [`build()`]
and is fixed
+/// afterwards.
+///
+/// | Option | Default | Controls |
+/// | --- | --- | --- |
+/// | [`stream()`], [`topic()`], [`partition()`] | the values passed to the
entry point | what is read |
+/// | [`batch_length()`] | 1000 | messages fetched per request |
+/// | [`poll_interval()`] | none | smallest gap between two requests |
+/// | [`polling_strategy()`] | [`PollingStrategy::next()`] | where reading
starts |
+/// | [`auto_commit()`] | [`AutoCommit::IntervalOrWhen`], one second,
[`AutoCommitWhen::PollingMessages`] | when offsets are committed |
+/// | [`allow_replay()`] | off | whether a message can be handed over again |
+/// | [`auto_join_consumer_group()`] | on | joining the group during
[`init()`](Self::init) |
+/// | [`create_consumer_group_if_not_exists()`] | on | creating the group when
it is missing |
+/// | [`polling_retry_interval()`] | one second | wait between attempts while
polling is blocked |
+/// | [`init_retries()`] | none, one second apart | retries when the stream or
topic is missing at [`init()`](Self::init) |
+/// | [`offset_drain_timeout()`] | five seconds | how long
[`shutdown()`](Self::shutdown) waits for pending commits |
+/// | [`encryptor()`] | inherited from the client | decrypting payloads and
user headers |
+///
+/// The switches have inverse setters as well, such as
[`without_poll_interval()`],
+/// [`without_encryptor()`], [`do_not_auto_join_consumer_group()`] and
+/// [`do_not_create_consumer_group_if_not_exists()`].
+///
+/// # Encryption
+///
+/// When the [`IggyClient`] was created with an encryptor, payloads and user
headers are decrypted
+/// before a message is yielded, which only works if the producer encrypted
them with a matching
+/// key. This is guaranteed if you spawned both, the [`IggyProducer`] and the
[`IggyConsumer`] from the same [`IggyClient`].
+/// A message that cannot be decrypted is yielded as an error and the rest of
its batch is
Review Comment:
messages before the failing one are dropped too - decryption runs over the
whole batch before anything is handed over, so the entire batch goes. and since
no offset advanced, the next poll re-fetches the same batch and fails again, so
it loops.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -229,26 +539,48 @@ impl IggyConsumer {
}
/// Returns the name of the consumer.
+ ///
+ /// For a consumer group this is also the name of the group.
pub fn name(&self) -> &str {
&self.consumer_name
}
- /// Returns the topic ID of the consumer.
+ /// Returns the identifier of the topic this consumer reads from.
pub fn topic(&self) -> &Identifier {
&self.topic_id
}
- /// Returns the stream ID of the consumer.
+ /// Returns the identifier of the stream this consumer reads from.
pub fn stream(&self) -> &Identifier {
&self.stream_id
}
- /// Returns the current partition ID of the consumer.
+ /// Returns the partition the last message came from.
+ ///
+ /// This is `0` until the first message has been read, because a partition
is only known once
+ /// the server has answered. For a consumer group the value changes over
time, as the server
+ /// can hand different partitions to this member.
pub fn partition_id(&self) -> u32 {
self.current_partition_id.load(ORDERING)
}
- /// Stores the consumer offset on the server either for the current
partition or the provided partition ID.
+ /// Stores an offset on the server, marking every message up to and
including it as consumed.
+ ///
+ /// This is the manual counterpart to [`AutoCommit`] and is meant for
+ /// [`AutoCommit::Disabled`].
+ ///
+ /// Pass `None` as `partition_id` to use the partition of the most recent
batch polled.
+ ///
+ /// An offset that is not ahead of the last one this consumer stored for
that partition is
+ /// skipped and `Ok(())` is returned without a request.
+ /// If you to re-read messages again, e.g. want to move an offset
backwards configure the consumer
Review Comment:
this sentence got scrambled ("If you to re-read messages again... backwards
configure"). similar small ones: 658 "the consumers", 671 "a background tasks",
742 "init_retires", 805 "If a the ... an time interval", 915 "Wait the task
until", 1482 "A previous used", 1666 "if it the", 371-372 missing word before
"Hence", 385 "a `IggyConsumer`", 360 "both, the".
##########
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
Review Comment:
under the default auto-commit-on-poll this lags the server by up to one
batch - the local mirror gets the pre-batch consumed offset (1173-1175) while
the server committed the batch end.
##########
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())
self.background_commit_notify.notify_one();
+
+ // A background_commit_task exists, if auto_commit is configured with
an interval option.
+ // If it exists, the task may be waiting or currently perform the
interval based store offset operation.
+ // In case it is currently working, wait until drain timeout has
passed and then force
+ // the task to abort.
if let Some(mut task) = self.background_commit_task.take()
&& time::timeout(self.offset_drain_timeout.get_duration(), &mut
task)
.await
.is_err()
{
- // Still running past the bound: abort it rather than leaving it
- // detached, so it can't send a stale store after we leave below.
task.abort();
warn!(
"Timed out waiting for the background offset-commit task to
stop for consumer: {}, aborted",
self.consumer_name
);
}
+ // Drop the sending end of the store offset task to end the
`recv_async()` loop in `send_store_offset().
Review Comment:
the `recv_async()` loop lives in the task spawned in `init()`, not in
`send_store_offset()` - that's the sender side. unclosed backtick too.
##########
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.
Review Comment:
not quite - the final flush below (1668-1695) commits every partition where
consumed > stored, and everything the aborted tasks had queued is covered by
that. offsets are only lost when the flush request itself fails (errors are
swallowed at 1683).
##########
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:
typo: the fn is `store_offsets_in_background()`. also the old comment here
explained why the drain happens before leaving the group (a store sent after
the leave would hit a group we already left) - that ordering constraint, and
the abort-so-no-stale-store rationale below, are worth keeping.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -229,26 +539,48 @@ impl IggyConsumer {
}
/// Returns the name of the consumer.
+ ///
+ /// For a consumer group this is also the name of the group.
pub fn name(&self) -> &str {
&self.consumer_name
}
- /// Returns the topic ID of the consumer.
+ /// Returns the identifier of the topic this consumer reads from.
pub fn topic(&self) -> &Identifier {
&self.topic_id
}
- /// Returns the stream ID of the consumer.
+ /// Returns the identifier of the stream this consumer reads from.
pub fn stream(&self) -> &Identifier {
&self.stream_id
}
- /// Returns the current partition ID of the consumer.
+ /// Returns the partition the last message came from.
+ ///
+ /// This is `0` until the first message has been read, because a partition
is only known once
Review Comment:
it's set on any poll answer, even an empty one (stored at 1492 before the
empty check), so it can be nonzero with zero messages read. the all-duplicates
filter path also resets it to 0 via `PolledMessages::empty()`. "until the
server has answered the first poll" would be closer.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -974,13 +1412,24 @@ impl ReceivedMessage {
}
}
+/// Yields messages one at a time.
+///
+/// Tries the buffer first, before a new batch is fetched from the server and
stored in the buffer.
+///
+/// The stream never yields `None`. So a `while let Some(..)` loop over it runs
+/// until the loop body breaks out. Errors are yielded as items and do not end
the stream, polling
+/// again retries. See the [type documentation](IggyConsumer#polling) for the
details of polling.
Review Comment:
there's no `#polling` anchor - the section is "How messages are read", so
`#how-messages-are-read`. rustdoc doesn't check fragments, so this stays green
but lands at the top of the page.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -100,6 +100,316 @@ pub enum AutoCommitAfter {
// 4. All `&self` methods only access Sync-safe fields
unsafe impl Sync for IggyConsumer {}
+/// Reads messages from the partitions of one topic and yields them one at a
time.
+///
+/// A topic is split into partitions, and a partition is an ordered log that
producers append to.
+/// Every message sits at an *offset*, its position in that log. Reading is
therefore always the
+/// same three decisions: which partition to read, where in it to start, and
how to keep track of
+/// how far you got so the next run can continue there.
+///
+/// `IggyConsumer` handles all three. It fetches batches of messages from the
server, keeps them in
+/// an in-memory buffer, decrypts them when the client is configured with an
encryptor, and records
+/// how far it has read. **It implements [`Stream`], so consuming is a loop
over [`StreamExt::next`].**
+///
+/// You can use a consumer as a worker draining a topic, a reader that replays
a
+/// partition from a chosen point, and a pool of consumers sharing a workload
through a consumer
+/// group.
+///
+/// # Creating a consumer
+///
+/// Easiest way is to use the [`IggyClient`] with a configured connection.
Then:
+/// - [`IggyClient::consumer()`] builds a standalone consumer, bound to the
one partition passed
+/// in.
+/// - [`IggyClient::consumer_group()`] builds a member of a consumer group.
The server gives every
+/// partition to exactly one member, so several consumers using the same
group name split the
+/// topic between them and share one set of offsets.
+///
+/// Note, building never talks to the server. [`init()`](Self::init) must be
awaited once before the
+/// first message is read.
+///
+/// # Examples
+///
+/// A standalone consumer reading partition 1 with the defaults:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer("my-consumer", "my-stream", "my-topic", 1)?
+/// .batch_length(100)
+/// .poll_interval(IggyDuration::new_from_secs(1))
+/// .build();
+/// consumer.init().await?;
+///
+/// while let Some(received) = consumer.next().await {
+/// match received {
+/// Ok(received) => println!("Offset: {}",
received.message.header.offset),
+/// Err(error) => eprintln!("Failed to read a message: {error}"),
+/// }
+/// }
+/// # Ok(())
+/// # }
+/// ```
+///
+/// A group member that commits every message right after it is handed over,
and shuts down
+/// cleanly:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn handle(message: &IggyMessage) {}
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer_group("order-workers", "my-stream", "my-topic")?
+/// .auto_commit(AutoCommit::When(AutoCommitWhen::ConsumingEachMessage))
+/// .polling_strategy(PollingStrategy::next())
+/// .build();
+/// consumer.init().await?;
+///
+/// let mut consumed = 0;
+/// while let Some(received) = consumer.next().await {
+/// handle(&received?.message).await;
+/// consumed += 1;
+/// if consumed == 100 {
+/// break;
+/// }
+/// }
+///
+/// consumer.shutdown().await?;
+/// # Ok(())
+/// # }
+/// ```
+///
+/// Committing by hand, so that a message the handler could not process comes
back:
+///
+/// ```rust,no_run
+/// use futures_util::StreamExt;
+/// use iggy::prelude::*;
+///
+/// # async fn handle(message: &IggyMessage) -> Result<(), IggyError> { Ok(())
}
+/// # async fn example() -> Result<(), IggyError> {
+/// let client =
IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?;
+/// client.connect().await?;
+///
+/// let mut consumer = client
+/// .consumer("my-consumer", "my-stream", "my-topic", 1)?
+/// .auto_commit(AutoCommit::Disabled)
+/// .polling_strategy(PollingStrategy::next())
+/// // Without this, messages already handed over once are never handed
over again.
+/// .allow_replay()
+/// .build();
+/// consumer.init().await?;
+///
+/// while let Some(received) = consumer.next().await {
+/// let received = received?;
+/// if handle(&received.message).await.is_ok() {
+/// consumer
+/// .store_offset(received.message.header.offset,
Some(received.partition_id))
+/// .await?;
+/// }
+/// }
+/// # Ok(())
+/// # }
+/// ```
+///
+/// # Which partitions are read
+///
+/// A **standalone consumer** reads exactly one partition, the one passed to
+/// [`IggyClient::consumer()`]. Covering a whole topic with several partitions
+/// means running one consumer per partition and dividing the work yourself.
+///
+/// A **consumer group member** does not choose. The server hands every
partition of the topic to
+/// exactly one member, so consumers sharing a group name split the topic
between them without
+/// coordinating. [`partition_id()`](Self::partition_id) reports where the
last message came from.
+///
+/// What to know when working with consumer groups:
+/// - A member joins during [`init()`](Self::init), creating the group first if
+/// [`create_consumer_group_if_not_exists()`] is set (the default). It
rejoins on its own after a
+/// reconnect and whenever the server reports that its membership is gone.
+/// - Until the join has succeeded the consumer does not poll. It waits for
+/// [`polling_retry_interval()`] and tries again.
+/// - Partitions are redistributed whenever members join or leave, so a member
reads different
+/// partitions over time and messages from several partitions interleave in
its stream.
+/// - More members than partitions leaves the surplus members idle. The
partition count of the
+/// topic is the ceiling on how far one group can be scaled out.
+/// - The group shares one set of stored offsets, kept under the group name.
Thus,
+/// a partition taken over by another member continues where the previous one
+/// committed.
+///
+/// # How messages are read
+///
+/// Reading is done by polling. One request fetches up to [`batch_length()`]
messages. The consumer
+/// passes the first one to the caller and buffers the rest. The next request
is sent once that buffer
+/// is empty.
+///
+/// [`poll_interval()`] sets a timeout between two requests should the buffer
be empty.
+/// Without it the next request goes out as soon as the previous one is
+/// answered, which is the fastest option but keeps a busy loop running
against an idle topic.
+///
+/// [`polling_strategy()`] decides **where** in the partition reading begins:
+///
+/// | Strategy | Starts at |
+/// | --- | --- |
+/// | [`PollingStrategy::next()`] (default) | the message after the offset
that is stored on the server |
+/// | [`PollingStrategy::first()`] | the oldest message in the partition |
+/// | [`PollingStrategy::last()`] | the end of the partition (returns up to
[`batch_length()`] of the most recent messages) |
+/// | [`PollingStrategy::offset()`] | a custom offset |
+/// | [`PollingStrategy::timestamp()`] | the first message at or after a given
point in time |
+///
+/// Only [`PollingStrategy::next()`] consults the offset stored on the server.
+/// Use this if you want to resume where a previous run stopped. The other
four are starting points for the first request only.
+/// From the second request onwards, the consumer asks for whatever follows it.
+///
+/// Note, when polling, [`StreamExt::next`] never returns `None`, not when the
topic is empty
+/// and not while the client is disconnected. A `while let Some(..)` loop only
ends when the
+/// loop body breaks out of it. A request that comes back empty is not an
error and not the end of
+/// the stream, it just means nothing new has arrived yet.
+///
+/// A failed request is yielded as `Some(Err(..))` and leaves the consumer
usable, while the next call
+/// retries. Connection and authentication failures pause polling until the
client has reconnected
+/// and signed in again, which the consumer handles automatically. Hence,
deciding when to give up on
+/// repeated errors is up to you.
+///
+/// For a boilerplate implementation of such a loop Iggy provides
[`IggyConsumerMessageExt::consume_messages`].
+///
+/// # Tracking what has been read
+///
+/// An offset is the index tracking what has been already read from a
partition by the consumer.
+/// Managing the offset has implications on where consumers resume reading
messages.
+///
+/// There are two positions (offsets) tracked in two different places:
+/// - The **reading position** is held by the consumer, one per partition, and
is the offset of the
+/// last message handed over
+/// ([`get_last_consumed_offset()`](Self::get_last_consumed_offset)). It
dies with the process.
+/// - The **stored offset** lives on the server under the consumer name, or
the group name for a
+/// group. This offset survives restarts. Writing it is called *storing* or
*committing* an offset.
+///
+/// Committing matters because [`PollingStrategy::next()`] resumes from the
stored offset. A
+/// consumer that never commits keeps starting over from the same place.
+/// Note, *committing* is a request of its own, not a side effect of reading
and therefore controllable.
Review Comment:
under the default `PollingMessages` the commit is a flag on the poll request
itself (line 1141), not a request of its own - the table row below says exactly
that. maybe scope this sentence to the other modes.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -1207,9 +1702,9 @@ impl IggyConsumer {
);
let client = self.client.read().await;
- // Cleared either way: this consumer is torn down regardless of
- // whether the broker confirmed the leave.
+ // Update consumer state to not being part of a consumer group.
Review Comment:
the old comment explained why this is cleared before the leave result is
known (the consumer is torn down either way, even if the broker never confirms)
- the replacement lost that.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -332,7 +737,11 @@ impl IggyConsumer {
let mut stream_exists =
client.get_stream(&stream_id).await?.is_some();
let mut topic_exists = client.get_topic(&stream_id,
&topic_id).await?.is_some();
+ // Absent streams or topics are not necessarily permanent failures.
+ // It may happen that get_stream/ get_topic races the initial
setup of the stream/ topic.
+ // Retry for init_retires times, while waiting interval between
retries.
loop {
+ // immediate happy path
Review Comment:
"immediate happy path" and a few others (1539, 1707) just restate the next
line - they can go.
--
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]
