github-actions[bot] commented on code in PR #4275:
URL: https://github.com/apache/iggy/pull/4275#discussion_r4272064093
##########
core/sdk/src/stream_builder/iggy_stream_consumer.rs:
##########
@@ -21,21 +21,240 @@ use crate::prelude::{IggyError, SystemClient, Validatable};
use crate::stream_builder::{IggyConsumerConfig, build};
use tracing::trace;
+/// Builds a connected [`IggyConsumer`] from one [`IggyConsumerConfig`].
+///
+/// `IggyStreamConsumer` is a namespace of associated functions. It holds no
state, and you never
+/// create a value of it. Use it for a process that only reads from a topic.
To build a producer
+/// and a consumer together, use [`IggyStream`].
+///
+/// # Building the consumer
+///
+/// [`build()`](Self::build) takes an [`IggyClient`] that is connected
already. It sends a ping
+/// first and fails with [`IggyError::NotConnected`] when that ping fails.
+/// [`with_client_from_url()`](Self::with_client_from_url) creates and
connects the client itself,
+/// so it skips the ping and returns the client next to the consumer.
+///
+/// Both functions then create the stream and the topic when
[`create_stream_if_not_exists()`] and
+/// [`create_topic_if_not_exists()`] allow it, build the consumer, and await
+/// [`IggyConsumer::init()`]. The consumer comes back initialized and, for a
group, joined. You can
+/// read messages right away.
+///
+/// # Reading from a stream that does not exist yet
+///
+/// Both switches are off by default. When the stream or the topic is missing
and its switch is
+/// off, the call logs a warning and carries on to [`IggyConsumer::init()`].
That call then retries
+/// [`init_retries()`] times, [`init_interval()`] apart, and fails with
+/// [`IggyError::StreamNameNotFound`] or [`IggyError::TopicNameNotFound`] once
the retries are
+/// spent. The default allows five retries three seconds apart, which covers a
producer that
+/// creates the topic at the same time.
+///
+/// # What the configuration sets
+///
+/// | Field of [`IggyConsumerConfig`] | Effect |
+/// | --- | --- |
+/// | [`stream_name()`], [`topic_name()`] | what is read |
+/// | [`consumer_name()`] | the name of the consumer, or of the group for
[`ConsumerKind::ConsumerGroup`] |
+/// | [`consumer_kind()`] | whether the consumer joins a group or reads one
partition alone |
+/// | [`partitions_count()`] | two things, see below |
+/// | [`batch_length()`] | messages fetched per request |
+/// | [`polling_interval()`], [`polling_strategy()`] | how often reading
happens, and where it starts |
+/// | [`auto_commit()`] | when offsets are committed |
+/// | [`polling_retry_interval()`] | the wait between attempts while polling
is blocked |
+/// | [`init_retries()`], [`init_interval()`] | retries for a missing stream
or topic |
+/// | [`encryptor()`] | decrypting payloads and user headers, and it replaces
the client's own |
+///
+/// [`partitions_count()`] carries two meanings. It is the partition count of
a topic that this
+/// call creates. For [`ConsumerKind::Consumer`] it is also the ID of the
single partition the
Review Comment:
warning: The doc says `partitions_count()` is also the ID of the partition a
standalone consumer reads, but `build_iggy_consumer.rs:61` passes
`config.partition_id()`, which defaults to 0. Fix: name `partition_id()` and
`with_partition_id()` here, and delete the `reads partition 3` example.
##########
core/sdk/src/stream_builder/config/config_iggy_consumer.rs:
##########
@@ -27,6 +27,135 @@ use tracing::error;
const DEFAULT_PARTITION_ID: u32 = 0;
+/// Describes the consumer that [`IggyStreamConsumer`] and [`IggyStream`]
build.
+///
+/// The value is a plain description. It never talks to the server, and
changing it later has no
+/// effect on a consumer that is built already.
+///
+/// # Creating a configuration
+///
+/// | Constructor | Use it for |
+/// | --- | --- |
+/// | [`builder()`](Self::builder) | naming every field you set |
+/// | [`from_stream_topic()`](Self::from_stream_topic) | `ConsumerGroup` for
one stream and one topic, with `batch_length` and `polling_interval`, defaults
elsewhere |
+/// | [`new()`](Self::new) | every field, positional |
+/// | [`default()`](Self::default) | the stream `test_stream` and the topic
`test_topic`, for examples and tests |
+///
+/// The builder requires every field that is not an [`Option`], so
Review Comment:
nit: The rule says the builder requires every field that is not an `Option`,
but `partition_id` has `#[builder(default = DEFAULT_PARTITION_ID)]` at line
183 and line 45 already lists it as optional. Fix: say that the builder
defaults `partition_id` to 0.
##########
core/sdk/src/stream_builder/iggy_stream_consumer.rs:
##########
@@ -21,21 +21,240 @@ use crate::prelude::{IggyError, SystemClient, Validatable};
use crate::stream_builder::{IggyConsumerConfig, build};
use tracing::trace;
+/// Builds a connected [`IggyConsumer`] from one [`IggyConsumerConfig`].
+///
+/// `IggyStreamConsumer` is a namespace of associated functions. It holds no
state, and you never
+/// create a value of it. Use it for a process that only reads from a topic.
To build a producer
+/// and a consumer together, use [`IggyStream`].
+///
+/// # Building the consumer
+///
+/// [`build()`](Self::build) takes an [`IggyClient`] that is connected
already. It sends a ping
+/// first and fails with [`IggyError::NotConnected`] when that ping fails.
+/// [`with_client_from_url()`](Self::with_client_from_url) creates and
connects the client itself,
+/// so it skips the ping and returns the client next to the consumer.
+///
+/// Both functions then create the stream and the topic when
[`create_stream_if_not_exists()`] and
+/// [`create_topic_if_not_exists()`] allow it, build the consumer, and await
+/// [`IggyConsumer::init()`]. The consumer comes back initialized and, for a
group, joined. You can
+/// read messages right away.
+///
+/// # Reading from a stream that does not exist yet
+///
+/// Both switches are off by default. When the stream or the topic is missing
and its switch is
+/// off, the call logs a warning and carries on to [`IggyConsumer::init()`].
That call then retries
+/// [`init_retries()`] times, [`init_interval()`] apart, and fails with
+/// [`IggyError::StreamNameNotFound`] or [`IggyError::TopicNameNotFound`] once
the retries are
+/// spent. The default allows five retries three seconds apart, which covers a
producer that
+/// creates the topic at the same time.
+///
+/// # What the configuration sets
+///
+/// | Field of [`IggyConsumerConfig`] | Effect |
+/// | --- | --- |
+/// | [`stream_name()`], [`topic_name()`] | what is read |
+/// | [`consumer_name()`] | the name of the consumer, or of the group for
[`ConsumerKind::ConsumerGroup`] |
+/// | [`consumer_kind()`] | whether the consumer joins a group or reads one
partition alone |
+/// | [`partitions_count()`] | two things, see below |
+/// | [`batch_length()`] | messages fetched per request |
+/// | [`polling_interval()`], [`polling_strategy()`] | how often reading
happens, and where it starts |
+/// | [`auto_commit()`] | when offsets are committed |
+/// | [`polling_retry_interval()`] | the wait between attempts while polling
is blocked |
+/// | [`init_retries()`], [`init_interval()`] | retries for a missing stream
or topic |
+/// | [`encryptor()`] | decrypting payloads and user headers, and it replaces
the client's own |
+///
+/// [`partitions_count()`] carries two meanings. It is the partition count of
a topic that this
+/// call creates. For [`ConsumerKind::Consumer`] it is also the ID of the
single partition the
+/// consumer reads, so a standalone consumer of a topic with three partitions
reads partition 3 and
+/// nothing else. Run one consumer per partition to cover such a topic, or use
+/// [`ConsumerKind::ConsumerGroup`] and let the server divide the partitions.
+///
+/// A consumer built here always creates its group when the group is missing,
and always joins it.
+/// Build through [`IggyClient::consumer()`] or
[`IggyClient::consumer_group()`] when you need to
+/// pick the partition and the group name apart, to replay messages, or to
join the group yourself.
+///
+/// # Defaults worth knowing
+///
+/// [`IggyConsumerConfig`] does not repeat the defaults of
[`IggyConsumerBuilder`]:
+///
+/// | Option | [`IggyConsumerConfig`] | [`IggyConsumerBuilder`] |
+/// | --- | --- | --- |
+/// | [`polling_strategy()`] | [`PollingStrategy::last()`], so every poll
starts [`batch_length()`] messages back from the end of the partition |
[`PollingStrategy::next()`] |
Review Comment:
nit: The cell says every poll starts `batch_length()` messages back from the
end, but only the first poll of a partition does, and later polls continue from
the consumer's tracked offset. Fix: say `the first poll of a partition`, here
and at config_iggy_consumer.rs:304.
##########
core/sdk/src/stream_builder/config/config_iggy_producer.rs:
##########
@@ -23,6 +23,95 @@ use bon::Builder;
use std::str::FromStr;
use std::sync::Arc;
+/// Describes the producer that [`IggyStreamProducer`] and [`IggyStream`]
build.
+///
+/// The value is a plain description. It never talks to the server, and
changing it later has no
+/// effect on a producer that is built already.
+///
+/// # Creating a configuration
+///
+/// | Constructor | Use it for |
+/// | --- | --- |
+/// | [`builder()`](Self::builder) | naming every field you set |
+/// | [`from_stream_topic()`](Self::from_stream_topic) | one stream and one
topic, defaults elsewhere |
+/// | [`new()`](Self::new) | every field, positional |
+/// | [`default()`](Self::default) | the stream `test_stream` and the topic
`test_topic`, for examples and tests |
+///
+/// The builder requires every field that is not an [`Option`], so it requires
+/// [`stream_name()`](Self::stream_name), [`topic_name()`](Self::topic_name),
+/// [`topic_partitions_count()`](Self::topic_partitions_count),
+/// [`batch_length()`](Self::batch_length),
[`linger_time()`](Self::linger_time) and
+/// [`partitioning()`](Self::partitioning).
+///
+/// # Fields and defaults
+///
+/// The defaults below are what [`default()`](Self::default) and
+/// [`from_stream_topic()`](Self::from_stream_topic) set. The builder has no
defaults.
Review Comment:
nit: `topic_durability` carries `#[builder(default)]` at line 126, so the
builder does have a default, and the fields table below omits that field. Fix:
add the `topic_durability()` row with its `Durability::Replicated` default, and
say which fields the builder defaults.
##########
core/sdk/src/stream_builder/config/config_iggy_producer.rs:
##########
@@ -69,23 +157,32 @@ impl Default for IggyProducerConfig {
}
impl IggyProducerConfig {
- /// Creates a new `IggyProducerConfig` with all fields defined.
+ /// Return a new instance of a `IggyConsumerConfig`.
Review Comment:
nit: `IggyProducerConfig::new()` returns an `IggyProducerConfig`, but its
new doc says `IggyConsumerConfig`. Fix: write the producer type in that
sentence.
##########
core/sdk/src/stream_builder/config/config_iggy_stream.rs:
##########
@@ -76,13 +181,31 @@ impl IggyStreamConfig {
})
}
+ /// Returns the half that describes the consumer.
pub fn consumer_config(&self) -> &IggyConsumerConfig {
&self.consumer_config
}
+ /// Returns the half that describes the producer.
pub fn producer_config(&self) -> &IggyProducerConfig {
&self.producer_config
}
+
+ /// Returns the stream name of the producer half.
+ ///
+ /// The consumer half carries its own, which can differ. Read
+ /// [`consumer_config()`](Self::consumer_config) for it.
+ pub fn stream_name(&self) -> &str {
Review Comment:
simplification: `stream_name()` and `topic_name()` only forward to the
producer half, and the plain names hide which half answers when the two halves
differ. Fix: drop both and read `producer_config().stream_name()`, as
examples/rust/src/stream-builder/stream-basic/main.rs:64 does.
##########
core/sdk/src/stream_builder/config/config_iggy_consumer.rs:
##########
@@ -206,65 +368,87 @@ impl IggyConsumerConfig {
self
}
+ /// Returns the name of the stream that is read.
pub fn stream_name(&self) -> &str {
&self.stream_name
}
+ /// Returns the name of the topic that is read.
pub fn topic_name(&self) -> &str {
&self.topic_name
}
+ /// Returns when the consumer commits an offset by itself.
pub fn auto_commit(&self) -> AutoCommit {
self.auto_commit
}
+ /// Returns how many messages one request fetches at most.
pub fn batch_length(&self) -> u32 {
self.batch_length
}
+
+ /// Returns whether the build creates the stream when it is missing.
pub fn create_stream_if_not_exists(&self) -> bool {
self.create_stream_if_not_exists
}
+ /// Returns whether the build creates the topic when it is missing.
pub fn create_topic_if_not_exists(&self) -> bool {
self.create_topic_if_not_exists
}
+ /// Returns the name of the consumer, or of its group.
pub fn consumer_name(&self) -> &str {
&self.consumer_name
}
+ /// Returns whether the consumer joins a group or reads one partition
alone.
pub fn consumer_kind(&self) -> ConsumerKind {
self.consumer_kind
}
+ /// Returns the smallest gap the consumer keeps between two requests.
pub fn polling_interval(&self) -> IggyDuration {
self.polling_interval
}
+ /// Returns where reading each partition starts.
pub fn polling_strategy(&self) -> PollingStrategy {
self.polling_strategy
}
+ /// Returns how many partitions a topic created by the build gets.
+ ///
+ /// For [`ConsumerKind::Consumer`] this is also the ID of the single
partition that the
Review Comment:
warning: The getter repeats the false claim that `partitions_count()` is the
read partition, while the build passes `config.partition_id()` and the callout
above at line 72 says so. Fix: delete the second paragraph and keep the callout.
--
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]
