haubur commented on code in PR #3913:
URL: https://github.com/apache/iggy/pull/3913#discussion_r3806116967
##########
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:
True, that makes sense as it is precise.
--
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]
