You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/10/26 21:30:39 UTC
(iggy) 14/15: fix(sdk): warn when consume_messages can never store an offset
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/deferred-polling-pr
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit bf45231734a269256c15a679f1823f70eea7c634
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Oct 26 21:34:17 2026 +0100
fix(sdk): warn when consume_messages can never store an offset
The consumer defaults moved to AutoCommit::Disabled and Next, but the
stream-builder path gives no signal. A MessageConsumer gets no handle
to store offsets, and the shutdown warning never fires when the
consumer is dropped inside a task, so an app on the default config
reads from offset 0 on every start without a log line.
consume_messages now warns once at its start for Disabled with Next,
sharing the shutdown check. IggyConsumerConfig::with_auto_commit lets
a default config set a mode, and the stream-consumer and stream-basic
examples keep committing on poll.
---
core/sdk/README.md | 7 +-
core/sdk/src/clients/consumer.rs | 75 +++++++++++++++++++++-
core/sdk/src/consumer_ext/consumer_message_ext.rs | 11 +++-
.../stream_builder/config/config_iggy_consumer.rs | 6 ++
.../rust/src/stream-builder/stream-basic/main.rs | 4 +-
.../src/stream-builder/stream-consumer/main.rs | 4 +-
6 files changed, 99 insertions(+), 8 deletions(-)
diff --git a/core/sdk/README.md b/core/sdk/README.md
index e98159350..637a3281b 100644
--- a/core/sdk/README.md
+++ b/core/sdk/README.md
@@ -290,8 +290,11 @@ This default is incompatible with earlier releases. The builder used to commit
with `AutoCommit::IntervalOrWhen(1s, PollingMessages)`, and `IggyConsumerConfig`
used `AutoCommit::When(PollingMessages)` with `Last`. Code that relied on those
defaults must set `auto_commit` explicitly, or each restart reads the same
-messages again. A consumer that delivered messages but stored no offset logs a
-warning at shutdown.
+messages again. `IggyConsumerConfig::with_auto_commit` sets it on a `default` or
+`from_stream_topic` config. A consumer that delivered messages but stored no
+offset logs a warning at shutdown. With `AutoCommit::Disabled` and `Next`,
+`consume_messages` also warns at its start, because its message consumer cannot
+store offsets.
Under `PollingMessages` the server commits every batch that it reads, so the
consumer does not prefetch. It fetches one batch when the application asks for a
diff --git a/core/sdk/src/clients/consumer.rs b/core/sdk/src/clients/consumer.rs
index 262202422..9aad2501f 100644
--- a/core/sdk/src/clients/consumer.rs
+++ b/core/sdk/src/clients/consumer.rs
@@ -717,6 +717,11 @@ impl IggyConsumer {
self.auto_commit
}
+ /// Whether the consumer resumes after its stored offsets but never stores one itself.
+ pub(crate) fn relies_on_manual_commits(&self) -> bool {
+ self.auto_commit == AutoCommit::Disabled && self.polling_strategy.kind == PollingKind::Next
+ }
+
/// Whether the client is up or comes back without the caller.
pub(crate) async fn client_can_recover(&self) -> bool {
self.client.read().await.can_recover().await
@@ -1618,8 +1623,7 @@ impl IggyConsumer {
);
}
- if self.auto_commit == AutoCommit::Disabled
- && self.polling_strategy.kind == PollingKind::Next
+ if self.relies_on_manual_commits()
&& self.state.last_stored_offsets.is_empty()
&& !self.state.last_consumed_offsets.is_empty()
{
@@ -1764,7 +1768,8 @@ mod tests {
}
}
- /// Records the offset of each message and stops the consumption after the first one.
+ /// Records the offset of each message. With a `shutdown` sender, it stops the consumption
+ /// after the first one.
struct RecordMessages {
offsets: std::sync::Mutex<Vec<u64>>,
shutdown: std::sync::Mutex<Option<tokio::sync::oneshot::Sender<()>>>,
@@ -3840,6 +3845,70 @@ mod tests {
assert_eq!(result, Ok(Err(IggyError::InvalidConfiguration)));
}
+ #[tokio::test]
+ async fn given_disabled_auto_commit_and_next_when_consuming_callbacks_should_warn_once() {
+ let warnings = consume_three_messages(AutoCommit::Disabled, PollingStrategy::next()).await;
+
+ assert_eq!(
+ warnings.len(),
+ 1,
+ "a run warns once, not per message: {warnings:?}"
+ );
+ assert!(warnings[0].contains("never stores offsets"), "{warnings:?}");
+ }
+
+ #[tokio::test]
+ async fn given_auto_commit_or_a_fixed_start_when_consuming_callbacks_should_not_warn() {
+ for (auto_commit, polling_strategy) in [
+ (
+ AutoCommit::When(AutoCommitWhen::PollingMessages),
+ PollingStrategy::next(),
+ ),
+ (AutoCommit::Disabled, PollingStrategy::first()),
+ ] {
+ let warnings = consume_three_messages(auto_commit, polling_strategy).await;
+
+ assert!(
+ warnings.is_empty(),
+ "{auto_commit:?} with {polling_strategy:?}: {warnings:?}"
+ );
+ }
+ }
+
+ /// Runs `consume_messages` over three buffered messages, ends it with a blocked poll, and
+ /// returns the warnings that the run logged.
+ async fn consume_three_messages(
+ auto_commit: AutoCommit,
+ polling_strategy: PollingStrategy,
+ ) -> Vec<String> {
+ const OPERATION: u64 = 42;
+ let mut consumer = builder()
+ .partition(Some(1))
+ .auto_commit(auto_commit)
+ .polling_strategy(polling_strategy)
+ .build();
+ consumer.initialized = true;
+ consumer.buffered_messages =
+ VecDeque::from(vec![message_at(1), message_at(2), message_at(3)]);
+ consumer.deferred_polls = Some(deferred::DeferredPolls::with_error(
+ &consumer,
+ IggyError::LaneBlocked(OPERATION),
+ ));
+ let handler = RecordMessages {
+ offsets: std::sync::Mutex::default(),
+ shutdown: std::sync::Mutex::new(None),
+ };
+ let (_shutdown, receiver) = tokio::sync::oneshot::channel();
+ let recorder = LevelReports::warnings();
+ let _subscriber = tracing::subscriber::set_default(recorder.clone());
+
+ let result = timeout(POLL_TIMEOUT, consumer.consume_messages(&handler, receiver)).await;
+
+ assert_eq!(result, Ok(Err(IggyError::LaneBlocked(OPERATION))));
+ assert_eq!(*handler.offsets.lock().unwrap(), [1, 2, 3]);
+ recorder.events.lock().unwrap().clone()
+ }
+
#[tokio::test]
async fn given_an_uninitialized_consumer_when_polled_should_fail_each_poll_and_warn_once() {
let mut consumer = builder_for(Consumer::default()).build();
diff --git a/core/sdk/src/consumer_ext/consumer_message_ext.rs b/core/sdk/src/consumer_ext/consumer_message_ext.rs
index 47edac85c..39dbea810 100644
--- a/core/sdk/src/consumer_ext/consumer_message_ext.rs
+++ b/core/sdk/src/consumer_ext/consumer_message_ext.rs
@@ -21,7 +21,7 @@ use crate::prelude::IggyError;
use async_trait::async_trait;
use futures_util::StreamExt;
use tokio::sync::oneshot;
-use tracing::{error, info, trace};
+use tracing::{error, info, trace, warn};
#[async_trait]
impl<'a> IggyConsumerMessageExt<'a> for IggyConsumer {
@@ -59,6 +59,15 @@ impl<'a> IggyConsumerMessageExt<'a> for IggyConsumer {
where
P: MessageConsumer + Sync,
{
+ // Unlike a caller of `next()`, a message consumer gets no handle to store offsets with.
+ if self.relies_on_manual_commits() {
+ warn!(
+ "Consumer: {name} on topic: {topic} and stream: {stream} never stores offsets, because it runs with AutoCommit::Disabled. With PollingStrategy::next(), each restart reads the same messages again. Set auto_commit to store them.",
+ name = self.name(),
+ topic = self.topic(),
+ stream = self.stream()
+ );
+ }
let auto_commit = self.auto_commit();
let store_offset_after_each_message = matches!(
auto_commit,
diff --git a/core/sdk/src/stream_builder/config/config_iggy_consumer.rs b/core/sdk/src/stream_builder/config/config_iggy_consumer.rs
index 39df1c23b..293466f04 100644
--- a/core/sdk/src/stream_builder/config/config_iggy_consumer.rs
+++ b/core/sdk/src/stream_builder/config/config_iggy_consumer.rs
@@ -222,6 +222,12 @@ impl IggyConsumerConfig {
self
}
+ /// Sets the auto-commit configuration, which `default` and `from_stream_topic` leave disabled.
+ pub fn with_auto_commit(mut self, auto_commit: AutoCommit) -> Self {
+ self.auto_commit = auto_commit;
+ self
+ }
+
pub fn stream_name(&self) -> &str {
&self.stream_name
}
diff --git a/examples/rust/src/stream-builder/stream-basic/main.rs b/examples/rust/src/stream-builder/stream-basic/main.rs
index 08d77048c..4c382154d 100644
--- a/examples/rust/src/stream-builder/stream-basic/main.rs
+++ b/examples/rust/src/stream-builder/stream-basic/main.rs
@@ -29,7 +29,9 @@ async fn main() -> Result<(), IggyError> {
println!("Build iggy producer & consumer");
// For customization, use the `new` or `from_stream_topic` constructor
- let stream_config = IggyStreamConfig::default();
+ let consumer_config = IggyConsumerConfig::default()
+ .with_auto_commit(AutoCommit::When(AutoCommitWhen::PollingMessages));
+ let stream_config = IggyStreamConfig::new(consumer_config, IggyProducerConfig::default());
let (producer, mut consumer) = IggyStream::build(&client, &stream_config).await?;
println!("Start message stream");
diff --git a/examples/rust/src/stream-builder/stream-consumer/main.rs b/examples/rust/src/stream-builder/stream-consumer/main.rs
index 6d833a98b..dc09a776b 100644
--- a/examples/rust/src/stream-builder/stream-consumer/main.rs
+++ b/examples/rust/src/stream-builder/stream-consumer/main.rs
@@ -25,7 +25,9 @@ const IGGY_URL: &str = "iggy://iggy:iggy@localhost:8090";
async fn main() -> Result<(), IggyError> {
println!("Build iggy client & consumer");
//For customization, use the `new` or `from_stream_topic` constructor
- let config = IggyConsumerConfig::default();
+ // Stores the offset of each polled batch, so a restart continues after it.
+ let config = IggyConsumerConfig::default()
+ .with_auto_commit(AutoCommit::When(AutoCommitWhen::PollingMessages));
let (client, mut consumer) =
IggyStreamConsumer::with_client_from_url(IGGY_URL, &config).await?;