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:38 UTC
(iggy) 13/15: fix(sdk): warn once when a consumer is polled before init
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 0bd3e28391820f3cb070dded08e10b26290327a5
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Oct 26 21:05:23 2026 +0100
fix(sdk): warn once when a consumer is polled before init
IggyConsumer::poll_next fails with InvalidConfiguration until init()
runs. Earlier releases polled without init(), so a loop that
continues on errors now spins without a single log line.
The first poll before init() logs a warning that names the
consumer. Later polls skip the warning, and every poll still fails
with the same error.
---
core/sdk/src/clients/consumer.rs | 76 ++++++++++++++++++++++++++++++++++------
1 file changed, 65 insertions(+), 11 deletions(-)
diff --git a/core/sdk/src/clients/consumer.rs b/core/sdk/src/clients/consumer.rs
index 7877d7fb7..262202422 100644
--- a/core/sdk/src/clients/consumer.rs
+++ b/core/sdk/src/clients/consumer.rs
@@ -556,6 +556,7 @@ pub struct IggyConsumer {
assignment_state: Option<(Arc<ConsumerGroupClientState>, String)>,
buffered_capacity: Option<deferred::FetchCapacity>,
initialized: bool,
+ uninitialized_poll_warned: bool,
shutdown: Arc<AtomicBool>,
can_poll: Arc<AtomicBool>,
client: IggyRwLock<ClientWrapper>,
@@ -654,6 +655,7 @@ impl IggyConsumer {
assignment_state: None,
buffered_capacity: None,
initialized: false,
+ uninitialized_poll_warned: false,
shutdown: Arc::new(AtomicBool::new(false)),
is_consumer_group,
joined_consumer_group: Arc::new(AtomicBool::new(false)),
@@ -1411,6 +1413,14 @@ impl Stream for IggyConsumer {
return Poll::Ready(None);
}
if !self.initialized {
+ // A loop that continues on errors would otherwise spin without a trace.
+ if !self.uninitialized_poll_warned {
+ self.uninitialized_poll_warned = true;
+ warn!(
+ "Consumer: {} was polled before init(). Every poll fails until init() runs.",
+ self.consumer_name
+ );
+ }
return Poll::Ready(Some(Err(IggyError::InvalidConfiguration)));
}
loop {
@@ -2666,14 +2676,17 @@ mod tests {
assert_eq!(consumer.get_last_stored_offset(1), Some(LOWER_OFFSET));
}
- /// Records the error events emitted on the test thread. The current-thread runtime of
- /// `#[tokio::test]` polls the tasks it spawns on that thread, so their events count too.
- #[derive(Clone, Default)]
- struct ErrorReports(Arc<std::sync::Mutex<Vec<String>>>);
+ /// Records the events of one level emitted on the test thread. The current-thread runtime
+ /// of `#[tokio::test]` polls the tasks it spawns on that thread, so their events count too.
+ #[derive(Clone)]
+ struct LevelReports {
+ level: Level,
+ events: Arc<std::sync::Mutex<Vec<String>>>,
+ }
- impl Subscriber for ErrorReports {
+ impl Subscriber for LevelReports {
fn enabled(&self, metadata: &Metadata<'_>) -> bool {
- *metadata.level() == Level::ERROR
+ *metadata.level() == self.level
}
fn new_span(&self, _: &Attributes<'_>) -> Id {
@@ -2689,7 +2702,7 @@ mod tests {
event.record(&mut |field: &Field, value: &dyn Debug| {
fields.push(format!("{field}: {value:?}"));
});
- self.0.lock().unwrap().push(fields.join(", "));
+ self.events.lock().unwrap().push(fields.join(", "));
}
fn enter(&self, _: &Id) {}
@@ -2697,12 +2710,26 @@ mod tests {
fn exit(&self, _: &Id) {}
}
- impl ErrorReports {
+ impl LevelReports {
+ fn errors() -> Self {
+ Self {
+ level: Level::ERROR,
+ events: Arc::default(),
+ }
+ }
+
+ fn warnings() -> Self {
+ Self {
+ level: Level::WARN,
+ events: Arc::default(),
+ }
+ }
+
/// Waits for the first report, or returns none after [`POLL_TIMEOUT`].
async fn wait(&self) -> Vec<String> {
let first = async {
loop {
- let reports = self.0.lock().unwrap().clone();
+ let reports = self.events.lock().unwrap().clone();
if !reports.is_empty() {
return reports;
}
@@ -2715,7 +2742,7 @@ mod tests {
#[tokio::test]
async fn interval_commit_should_report_a_failed_store() {
- let recorder = ErrorReports::default();
+ let recorder = LevelReports::errors();
let _subscriber = tracing::subscriber::set_default(recorder.clone());
let consumer = builder().partition(Some(1)).build();
consumer
@@ -2742,7 +2769,7 @@ mod tests {
#[tokio::test]
async fn queued_commit_should_report_a_failed_store() {
- let recorder = ErrorReports::default();
+ let recorder = LevelReports::errors();
let _subscriber = tracing::subscriber::set_default(recorder.clone());
let mut consumer = builder().partition(Some(1)).build();
consumer.initialized = true;
@@ -3812,4 +3839,31 @@ mod tests {
assert_eq!(result, Ok(Err(IggyError::InvalidConfiguration)));
}
+
+ #[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();
+ let recorder = LevelReports::warnings();
+ let _subscriber = tracing::subscriber::set_default(recorder.clone());
+
+ let first = poll_once(&mut consumer).await;
+ let second = poll_once(&mut consumer).await;
+
+ for polled in [first, second] {
+ assert!(
+ matches!(polled, Err(IggyError::InvalidConfiguration)),
+ "every poll before init fails: {polled:?}"
+ );
+ }
+ let warnings = recorder.events.lock().unwrap().clone();
+ assert_eq!(
+ warnings.len(),
+ 1,
+ "only the first poll before init warns: {warnings:?}"
+ );
+ assert!(
+ warnings[0].contains("was polled before init()"),
+ "{warnings:?}"
+ );
+ }
}