feat(server)!: add bounded deferred (long) polling

.claude/skills/connector-runtime/SKILL.md

@@ -133,7 +133,7 @@ Per `[[streams]]` entry per sink:
 
 1. Build / ensure consumer group `iggy-connect-sink-{key}` (`connect` matches the docker image name `apache/iggy-connect` - see `runtime/src/sink.rs::default_consumer_group`). Overridable via the stream config's `consumer_group`.
 2. Spawn one task per topic (`spawn_consume_tasks`).
-3. Poll Iggy → batch messages (default `batch_length = 1000`, `poll_interval = 5ms`). A `consumer.next()` `Err` (transport/decode at the Iggy client boundary) bumps `errors` and `continue`s - offset already auto-committed via `AutoCommit::When(AutoCommitWhen::PollingMessages)`, message effectively dropped.
+3. Poll Iggy → batch messages (default `batch_length = 1000`). A `consumer.next()` `Err` (transport/decode at the Iggy client boundary) bumps `errors` and `continue`s. A rejected poll committed nothing, so the consumer retries the same offset. Polls commit via `AutoCommit::When(AutoCommitWhen::PollingMessages)`, so only a reply lost after the server committed can skip its batch.
 4. Decode via `Schema::decoder()`. Decode failure logs + bumps `errors`. `messages_filtered` is reserved for the transform filter contract (`Ok(None)`).
 5. Apply transforms (chain). A transform `Err` is logged, bumps `errors`, and drops only that message (drop-and-continue, mirroring the source) - one bad payload never kills the batch. `Ok(None)` filters the message - bumps `messages_filtered{connector_type="sink"}`.
 6. Postcard-encode batch as `RawMessages`. Headers as a separate postcard blob. Per-message failures (missing field, payload conversion, header serialization) bump `errors` + skip. Batch-level failures (postcard serialization of metadata / `RawMessages`) propagate `Err` -> `spawn_consume_tasks` wrapper bumps `errors` + `set_error`. The total histogram and benchmark emit still fire for the failed batch before the `Err` propagates.
@@ -146,8 +146,9 @@ Stages (both labels snake_case): `decode` (schema decode loop), `prepare` (trans
 
 Gotchas:
 
-- Batch flushes on `current_offset != message_offset` (gap) OR batch full - higher latency for low-volume topics.
-- Auto-commit happens on poll - idempotency at the sink protects against re-delivery on crash before plugin write.
+- Batch flushes when a message is the last one in its partition (`message_offset == current_offset`), when the batch is full, OR before a message of another partition, because the plugin gets one partition id per batch. `max_wait` and `min_count` on the stream make the server collect larger replies.
+- Auto-commit happens on poll, so sink delivery is at-most-once: a crash after the poll and before the plugin write loses that batch.
+- A stop still flushes the batch the loop holds, and the task then awaits `IggyConsumer::shutdown`, which rewinds over the batches the consumer holds. Without either step, the next run skips messages that the polls committed.
 
 ## Logging format
 

Cargo.toml

@@ -213,11 +213,11 @@ hyper-util = { version = "0.1.21", features = ["server-auto", "service"] }
 iceberg = "0.10.1"
 iceberg-catalog-rest = "0.10.1"
 iceberg-storage-opendal = "0.10.1"
-iggy = { path = "core/sdk", version = "0.11.1-edge.4" }
-iggy-cli = { path = "core/cli", version = "0.14.1-edge.3" }
+iggy = { path = "core/sdk", version = "0.11.1-edge.5" }
+iggy-cli = { path = "core/cli", version = "0.14.1-edge.4" }
 iggy-gateway-kafka = { path = "gateways/kafka" }
-iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.1-edge.3" }
-iggy_common = { path = "core/common", version = "0.11.1-edge.4" }
+iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.1-edge.4" }
+iggy_common = { path = "core/common", version = "0.11.1-edge.5" }
 iggy_connector_doris_sink = { path = "core/connectors/sinks/doris_sink" }
 iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.5.1-edge.2" }
 indexmap = "2.14.2"

README.md

@@ -198,7 +198,6 @@ stream = "qw"
 topics = ["records"]
 schema = "json"
 batch_length = 1000
-poll_interval = "5ms"
 consumer_group = "qw_sink_connector"
 
 [transforms.add_fields]
@@ -482,7 +481,6 @@ let mut consumer = client
     .create_consumer_group_if_not_exists()
     .auto_join_consumer_group()
     .polling_strategy(PollingStrategy::next())
-    .poll_interval(IggyDuration::from_str("1ms")?)
     .batch_length(1000)
     .build();
 

core/ai/mcp/Cargo.toml

@@ -17,7 +17,7 @@
 
 [package]
 name = "iggy-mcp"
-version = "0.5.1-edge.3"
+version = "0.5.1-edge.4"
 description = "MCP Server for Iggy message streaming platform"
 edition = "2024"
 license = "Apache-2.0"

core/bench/Cargo.toml

@@ -17,7 +17,7 @@
 
 [package]
 name = "iggy-bench"
-version = "0.6.1-edge.3"
+version = "0.6.1-edge.4"
 edition = "2024"
 license = "Apache-2.0"
 repository = "https://github.com/apache/iggy"
@@ -46,6 +46,7 @@ governor = { workspace = true }
 hostname = { workspace = true }
 human-repr = { workspace = true }
 iggy = { workspace = true }
+iggy_binary_protocol = { workspace = true }
 nonzero_lit = { workspace = true }
 rand = { workspace = true }
 rayon = { workspace = true }

core/bench/src/actors/consumer/benchmark_consumer.rs

@@ -483,6 +483,7 @@ mod tests {
                 messages_per_batch: args.messages_per_batch(),
                 warmup_time: args.warmup_time(),
                 polling_kind: PollingKind::Next,
+                poll_options: DeferredPollOptions::default(),
                 origin_timestamp_latency_calculation,
                 pretty: false,
             },

core/bench/src/actors/consumer/client/high_level.rs

@@ -26,11 +26,12 @@ use crate::actors::{
 use crate::utils::ClientFactory;
 use futures_util::StreamExt;
 use iggy::prelude::*;
-use std::{sync::Arc, time::Duration};
+use std::sync::Arc;
 use tokio::time::{Instant, timeout};
 use tracing::{error, warn};
 
 const TOPIC_ID: &str = "topic-1";
+const PREFETCH_BATCHES: u32 = 16;
 
 pub struct HighLevelConsumerClient {
     client_factory: Arc<dyn ClientFactory>,
@@ -81,21 +82,41 @@ impl HighLevelConsumerClient {
         };
         Ok(builder
             .batch_length(self.config.messages_per_batch.get())
+            .poll_options(self.config.poll_options)
+            .prefetch_messages(
+                self.config
+                    .messages_per_batch
+                    .get()
+                    .saturating_mul(PREFETCH_BATCHES),
+            )
+            .prefetch_bytes(
+                self.config
+                    .poll_options
+                    .max_bytes
+                    .saturating_mul(PREFETCH_BATCHES),
+            )
             .build())
     }
 }
 
 impl ConsumerClient for HighLevelConsumerClient {
+    #[expect(
+        clippy::significant_drop_tightening,
+        reason = "The borrowed consumer retains reservations until its buffered messages are drained."
+    )]
     async fn consume_batch(&mut self) -> Result<Option<BatchMetrics>, IggyError> {
         let consumer = self.consumer.as_mut().expect("Consumer not initialized");
 
+        // One poll can wait up to its request timeout, so a shorter timeout would restart the
+        // batch clock in the middle of a deliberate wait.
+        let poll_timeout = self.config.poll_options.request_timeout.get_duration();
         let batch_start = Instant::now();
         let mut batch_messages = 0;
         let mut batch_user_bytes = 0;
         let mut batch_total_bytes = 0;
 
         while batch_messages < self.config.messages_per_batch.get() {
-            let timeout_result = timeout(Duration::from_secs(1), consumer.next()).await;
+            let timeout_result = timeout(poll_timeout, consumer.next()).await;
 
             match timeout_result {
                 Ok(Some(message_result)) => match message_result {
@@ -141,6 +162,10 @@ impl ConsumerClient for HighLevelConsumerClient {
         }
     }
 
+    #[expect(
+        clippy::significant_drop_tightening,
+        reason = "The replacement must finish initialization before taking over the consumer slot"
+    )]
     async fn reset_offsets(&mut self) -> Result<(), IggyError> {
         let client = self.client.as_deref().expect("Client not initialized");
 
@@ -157,6 +182,10 @@ impl ConsumerClient for HighLevelConsumerClient {
 }
 
 impl BenchmarkInit for HighLevelConsumerClient {
+    #[expect(
+        clippy::significant_drop_tightening,
+        reason = "The initialized consumer moves into self and must retain its prefetch reservations."
+    )]
     async fn setup(&mut self) -> Result<(), IggyError> {
         let client = self.client_factory.create_authenticated_client().await?;
         let mut consumer = self.build_consumer(&client, PollingStrategy::next())?;

core/bench/src/actors/consumer/client/interface.rs

@@ -28,6 +28,7 @@ pub struct BenchmarkConsumerConfig {
     pub messages_per_batch: BenchmarkNumericParameter,
     pub warmup_time: IggyDuration,
     pub polling_kind: PollingKind,
+    pub poll_options: DeferredPollOptions,
     pub origin_timestamp_latency_calculation: bool,
     pub pretty: bool,
 }

core/bench/src/actors/consumer/client/low_level.rs

@@ -75,14 +75,15 @@ impl ConsumerClient for LowLevelConsumerClient {
         };
         let before_poll = Instant::now();
         let polled = client
-            .poll_messages_with_strategy_for(
+            .poll_messages_with_strategy_for_and_options(
                 &self.stream_id,
                 &self.topic_id,
                 self.partition_id,
                 consumer,
                 &strategy_for,
                 messages_to_receive,
                 self.auto_commit,
+                Some(self.config.poll_options),
             )
             .await;
 
@@ -189,8 +190,8 @@ mod tests {
     use crate::utils::client_factory::TcpClientFactory;
     use bench_report::numeric_parameter::BenchmarkNumericParameter;
     use iggy::prelude::{
-        ClientWrapper, Consumer, Identifier, IggyClient, IggyDuration, PollingStrategy, TcpClient,
-        TcpClientConfig,
+        ClientWrapper, Consumer, DeferredPollOptions, Identifier, IggyClient, IggyDuration,
+        PollingStrategy, TcpClient, TcpClientConfig,
     };
     use std::{collections::HashMap, sync::Arc};
 
@@ -213,6 +214,7 @@ mod tests {
                     messages_per_batch: BenchmarkNumericParameter::Value(1),
                     warmup_time: IggyDuration::default(),
                     polling_kind: polling_strategy.kind,
+                    poll_options: DeferredPollOptions::default(),
                     origin_timestamp_latency_calculation: false,
                     pretty: false,
                 },

core/bench/src/actors/consumer/typed_benchmark_consumer.rs

@@ -54,6 +54,7 @@ impl TypedBenchmarkConsumer {
         sampling_time: IggyDuration,
         moving_average_window: u32,
         polling_kind: PollingKind,
+        poll_options: DeferredPollOptions,
         limit_bytes_per_second: Option<IggyByteSize>,
         origin_timestamp_latency_calculation: bool,
         pretty: bool,
@@ -65,6 +66,7 @@ impl TypedBenchmarkConsumer {
             messages_per_batch,
             warmup_time,
             polling_kind,
+            poll_options,
             origin_timestamp_latency_calculation,
             pretty,
         };

core/bench/src/actors/producing_consumer/typed_benchmark_producing_consumer.rs

@@ -40,8 +40,8 @@ use bench_report::{
 use iggy::prelude::*;
 
 pub enum TypedBenchmarkProducingConsumer {
-    High(BenchmarkProducingConsumer<HighLevelProducerClient, HighLevelConsumerClient>),
-    Low(BenchmarkProducingConsumer<LowLevelProducerClient, LowLevelConsumerClient>),
+    High(Box<BenchmarkProducingConsumer<HighLevelProducerClient, HighLevelConsumerClient>>),
+    Low(Box<BenchmarkProducingConsumer<LowLevelProducerClient, LowLevelConsumerClient>>),
 }
 impl TypedBenchmarkProducingConsumer {
     #[allow(clippy::too_many_arguments)]
@@ -62,6 +62,7 @@ impl TypedBenchmarkProducingConsumer {
         moving_average_window: u32,
         limit_bytes_per_second: Option<IggyByteSize>,
         polling_kind: PollingKind,
+        poll_options: DeferredPollOptions,
         origin_timestamp_latency_calculation: bool,
         pretty: bool,
     ) -> Self {
@@ -82,6 +83,7 @@ impl TypedBenchmarkProducingConsumer {
             messages_per_batch,
             warmup_time,
             polling_kind,
+            poll_options,
             origin_timestamp_latency_calculation,
             pretty,
         };
@@ -90,7 +92,7 @@ impl TypedBenchmarkProducingConsumer {
             let producer =
                 HighLevelProducerClient::new(client_factory.clone(), producer_config.clone());
             let consumer = HighLevelConsumerClient::new(client_factory, consumer_config.clone());
-            Self::High(BenchmarkProducingConsumer::new(
+            Self::High(Box::new(BenchmarkProducingConsumer::new(
                 producer,
                 consumer,
                 benchmark_kind,
@@ -101,12 +103,12 @@ impl TypedBenchmarkProducingConsumer {
                 limit_bytes_per_second,
                 producer_config,
                 consumer_config,
-            ))
+            )))
         } else {
             let producer =
                 LowLevelProducerClient::new(client_factory.clone(), producer_config.clone());
             let consumer = LowLevelConsumerClient::new(client_factory, consumer_config.clone());
-            Self::Low(BenchmarkProducingConsumer::new(
+            Self::Low(Box::new(BenchmarkProducingConsumer::new(
                 producer,
                 consumer,
                 benchmark_kind,
@@ -117,7 +119,7 @@ impl TypedBenchmarkProducingConsumer {
                 limit_bytes_per_second,
                 producer_config,
                 consumer_config,
-            ))
+            )))
         }
     }