You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "arunsingh (via GitHub)" <gi...@apache.org> on 2026/08/01 07:26:08 UTC
Re: [PR] feat(server): add deferred PollMessages wait timeout [iggy]
arunsingh commented on code in PR #3605:
URL: https://github.com/apache/iggy/pull/3605#discussion_r3694996978
##########
core/integration/tests/server/scenarios/poll_messages_wait_timeout_scenario.rs:
##########
@@ -0,0 +1,429 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use iggy::prelude::*;
+use integration::harness::TestHarness;
+use std::time::{Duration, Instant};
+use tokio::time::sleep;
+#[cfg(not(feature = "vsr"))]
+use tokio::time::timeout;
+
+const STREAM_NAME: &str = "poll-wait-timeout-stream";
+const TOPIC_NAME: &str = "poll-wait-timeout-topic";
+const CONSUMER_GROUP_NAME: &str = "poll-wait-timeout-group";
+#[cfg(not(feature = "vsr"))]
+const PARTITION_ID: u32 = 0;
+
+#[cfg(not(feature = "vsr"))]
+pub async fn run_semantics_checks(harness: &TestHarness) {
+ let client = harness.root_client().await.expect("root client");
+ let producer = harness.root_client().await.expect("producer client");
+ setup_topic(&client, 1).await;
+
+ verify_immediate_empty_polling(&client).await;
+ verify_timeout_waits_and_returns_empty(&client).await;
+ verify_existing_messages_return_without_waiting(&producer, &client).await;
+ verify_timeout_auto_commit_boundaries(&producer, &client).await;
+
+ client
+ .delete_stream(&Identifier::named(STREAM_NAME).unwrap())
+ .await
+ .unwrap();
+}
+
+#[cfg(not(feature = "vsr"))]
+pub async fn run_wake_after_append_checks(harness: &TestHarness) {
+ let producer = harness.root_client().await.expect("producer client");
+ let consumer_client = harness.root_client().await.expect("consumer client");
+
+ producer.create_stream(STREAM_NAME).await.unwrap();
+ producer
+ .create_topic(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ TOPIC_NAME,
+ 1,
+ CompressionAlgorithm::default(),
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .unwrap();
+
+ let timeout = Duration::from_secs(5);
+ let start = Instant::now();
+ let stream = Identifier::named(STREAM_NAME).unwrap();
+ let topic = Identifier::named(TOPIC_NAME).unwrap();
+ let consumer = Consumer::default();
+ let strategy = PollingStrategy::offset(0);
+ let poll = consumer_client.poll_messages_with_timeout(
+ &stream,
+ &topic,
+ Some(0),
+ &consumer,
+ &strategy,
+ 1,
+ false,
+ timeout,
+ );
+ let send = async {
+ sleep(Duration::from_millis(100)).await;
+ let mut messages = vec![
+ IggyMessage::builder()
+ .id(1)
+ .payload("wake-after-append".into())
+ .build()
+ .unwrap(),
+ ];
+ producer
+ .send_messages(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ &Partitioning::partition_id(PARTITION_ID),
+ &mut messages,
+ )
+ .await
+ .unwrap();
+ };
+
+ let (polled, _) = tokio::join!(poll, send);
+ let polled = polled.unwrap();
+
+ assert_eq!(polled.partition_id, PARTITION_ID);
+ assert_eq!(polled.messages.len(), 1);
+ assert!(
+ start.elapsed() < timeout / 2,
+ "poll should wake after a successful append instead of waiting for timeout"
+ );
+
+ producer
+ .delete_stream(&Identifier::named(STREAM_NAME).unwrap())
+ .await
+ .unwrap();
+}
+
+#[cfg(not(feature = "vsr"))]
+async fn setup_topic(client: &IggyClient, partitions_count: u32) {
+ client.create_stream(STREAM_NAME).await.unwrap();
+ client
+ .create_topic(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ TOPIC_NAME,
+ partitions_count,
+ CompressionAlgorithm::default(),
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .unwrap();
+}
+
+#[cfg(not(feature = "vsr"))]
+async fn verify_immediate_empty_polling(client: &IggyClient) {
+ let stream = Identifier::named(STREAM_NAME).unwrap();
+ let topic = Identifier::named(TOPIC_NAME).unwrap();
+ let consumer = Consumer::new(Identifier::numeric(101).unwrap());
+ let strategy = PollingStrategy::offset(0);
+
+ let immediate = timeout(
+ Duration::from_secs(1),
+ client.poll_messages(
+ &stream,
+ &topic,
+ Some(PARTITION_ID),
+ &consumer,
+ &strategy,
+ 1,
+ false,
+ ),
+ )
+ .await
+ .expect("immediate poll should not wait")
+ .unwrap();
+ assert!(immediate.messages.is_empty());
+
+ let zero_timeout = timeout(
+ Duration::from_secs(1),
+ client.poll_messages_with_timeout(
+ &stream,
+ &topic,
+ Some(PARTITION_ID),
+ &consumer,
+ &strategy,
+ 1,
+ false,
+ Duration::ZERO,
+ ),
+ )
+ .await
+ .expect("zero timeout poll should not wait")
+ .unwrap();
+ assert!(zero_timeout.messages.is_empty());
+}
+
+#[cfg(not(feature = "vsr"))]
+async fn verify_timeout_waits_and_returns_empty(client: &IggyClient) {
+ let stream = Identifier::named(STREAM_NAME).unwrap();
+ let topic = Identifier::named(TOPIC_NAME).unwrap();
+ let consumer = Consumer::new(Identifier::numeric(102).unwrap());
+ let strategy = PollingStrategy::offset(0);
+ let poll = client.poll_messages_with_timeout(
+ &stream,
+ &topic,
+ Some(PARTITION_ID),
+ &consumer,
+ &strategy,
+ 1,
+ false,
+ Duration::from_millis(120),
+ );
+ tokio::pin!(poll);
+
+ tokio::select! {
+ result = &mut poll => panic!("non-zero timeout returned before its wait window: {result:?}"),
+ () = sleep(Duration::from_millis(20)) => {}
+ }
+
+ let polled = poll.await.unwrap();
+ assert!(polled.messages.is_empty());
+}
+
+#[cfg(not(feature = "vsr"))]
+async fn verify_existing_messages_return_without_waiting(
+ producer: &IggyClient,
+ client: &IggyClient,
+) {
+ let stream = Identifier::named(STREAM_NAME).unwrap();
+ let topic = Identifier::named(TOPIC_NAME).unwrap();
+ send_one(producer, 1, "already-readable").await;
+
+ let polled = timeout(
+ Duration::from_secs(1),
+ client.poll_messages_with_timeout(
+ &stream,
+ &topic,
+ Some(PARTITION_ID),
+ &Consumer::new(Identifier::numeric(103).unwrap()),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ Duration::from_secs(5),
+ ),
+ )
+ .await
+ .expect("readable data should return without waiting for timeout")
+ .unwrap();
+
+ assert_eq!(polled.partition_id, PARTITION_ID);
+ assert_eq!(polled.messages.len(), 1);
+ assert_eq!(polled.messages[0].payload.as_ref(), b"already-readable");
+}
+
+#[cfg(not(feature = "vsr"))]
+async fn verify_timeout_auto_commit_boundaries(producer: &IggyClient, client: &IggyClient) {
+ let stream = Identifier::named(STREAM_NAME).unwrap();
+ let topic = Identifier::named(TOPIC_NAME).unwrap();
+ let empty_timeout_consumer = Consumer::new(Identifier::numeric(104).unwrap());
+ let empty = client
+ .poll_messages_with_timeout(
+ &stream,
+ &topic,
+ Some(PARTITION_ID),
+ &empty_timeout_consumer,
+ &PollingStrategy::offset(1),
+ 1,
+ true,
+ Duration::from_millis(50),
+ )
+ .await
+ .unwrap();
+
+ assert!(empty.messages.is_empty());
+ let offset_after_empty = client
+ .get_consumer_offset(&empty_timeout_consumer, &stream, &topic, Some(PARTITION_ID))
+ .await
+ .unwrap();
+ assert!(offset_after_empty.is_none());
+
+ let auto_commit_consumer = Consumer::new(Identifier::numeric(105).unwrap());
+ let auto_commit_strategy = PollingStrategy::offset(1);
+ let poll = client.poll_messages_with_timeout(
+ &stream,
+ &topic,
+ Some(PARTITION_ID),
+ &auto_commit_consumer,
+ &auto_commit_strategy,
+ 1,
+ true,
+ Duration::from_secs(5),
+ );
+ let send = async {
+ sleep(Duration::from_millis(100)).await;
+ send_one(producer, 2, "wake-auto-commit").await;
+ };
+ let (polled, _) = tokio::join!(poll, send);
+ let polled = polled.unwrap();
+
+ assert_eq!(polled.messages.len(), 1);
+ let committed = client
+ .get_consumer_offset(&auto_commit_consumer, &stream, &topic, Some(PARTITION_ID))
+ .await
+ .unwrap()
+ .expect("wake with returned messages should auto-commit");
+ assert_eq!(committed.stored_offset, polled.messages[0].header.offset);
+}
+
+#[cfg(not(feature = "vsr"))]
+async fn send_one(client: &IggyClient, id: u128, payload: &str) {
+ let mut messages = vec![
+ IggyMessage::builder()
+ .id(id)
+ .payload(payload.to_owned().into())
+ .build()
+ .unwrap(),
+ ];
+ client
+ .send_messages(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ &Partitioning::partition_id(PARTITION_ID),
+ &mut messages,
+ )
+ .await
+ .unwrap();
+}
+
+pub async fn run_consumer_group_checks(harness: &TestHarness) {
+ let client = harness.root_client().await.expect("root client");
+
+ client.create_stream(STREAM_NAME).await.unwrap();
+ client
+ .create_topic(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ TOPIC_NAME,
+ 2,
+ CompressionAlgorithm::default(),
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .unwrap();
+ client
+ .create_consumer_group(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ CONSUMER_GROUP_NAME,
+ )
+ .await
+ .unwrap();
+ client
+ .join_consumer_group(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ &Identifier::named(CONSUMER_GROUP_NAME).unwrap(),
+ )
+ .await
+ .unwrap();
+
+ let mut messages = vec![
+ IggyMessage::builder()
+ .id(1)
+ .payload("ready-on-partition-1".into())
+ .build()
+ .unwrap(),
+ ];
+ client
+ .send_messages(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ &Partitioning::partition_id(1),
+ &mut messages,
+ )
+ .await
+ .unwrap();
+
+ let consumer = Consumer::group(Identifier::named(CONSUMER_GROUP_NAME).unwrap());
+ let group = client
+ .get_consumer_group(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ &Identifier::named(CONSUMER_GROUP_NAME).unwrap(),
+ )
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(group.members[0].partitions, vec![0, 1]);
+
+ wait_until_partition_readable(&client).await;
+
+ let timeout = Duration::from_millis(750);
+ let start = Instant::now();
+ let polled = client
+ .poll_messages_with_timeout(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ Some(0),
+ &consumer,
+ &PollingStrategy::offset(0),
+ 1,
+ true,
+ timeout,
+ )
+ .await
+ .unwrap();
+
+ assert_eq!(polled.partition_id, 1);
Review Comment:
Rebased and gated this CG case out of VSR; I kept full server-ng owned-partition scan/extra CG wake coverage as follow-up unless you want it here.
##########
core/server/src/shard/handlers.rs:
##########
@@ -74,6 +74,7 @@ async fn handle_request(
.await?;
shard.metrics.increment_messages(messages_count as u64);
+ shard.wake_poll_waiters(&namespace);
Review Comment:
Added a live-waiter fast path so append skips the registry mutex when nobody is waiting.
##########
core/server/src/binary/handlers/messages/poll_messages_handler.rs:
##########
@@ -41,21 +46,76 @@ pub async fn handle_poll_messages(
let partition_id = req.partition_id;
let count = req.count;
let auto_commit = req.auto_commit;
+ let wait_timeout = Duration::from_micros(req.wait_timeout_us);
debug!(
"session: {session}, command: poll_messages, stream_id: {stream_id}, topic_id: {topic_id}, partition_id: {partition_id:?}"
);
shard.ensure_authenticated(session)?;
- let args = PollingArgs::new(strategy, count, auto_commit);
-
let user_id = session.get_user_id();
let client_id = session.client_id;
let topic = shard.resolve_topic_for_poll(user_id, &stream_id, &topic_id)?;
- let (metadata, mut batch) = shard
- .poll_messages(client_id, topic, consumer, partition_id, args)
+ let (mut metadata, mut batch) = shard
+ .poll_messages(
+ client_id,
+ topic,
+ consumer.clone(),
Review Comment:
Changed shard polling to borrow Consumer; immediate polls no longer clone it.
##########
core/server/src/binary/handlers/messages/poll_messages_handler.rs:
##########
@@ -41,21 +46,76 @@ pub async fn handle_poll_messages(
let partition_id = req.partition_id;
let count = req.count;
let auto_commit = req.auto_commit;
+ let wait_timeout = Duration::from_micros(req.wait_timeout_us);
Review Comment:
Added a server-side 30s clamp and warning on waiter-cap fallback; happy to wire the cap to config if preferred.
##########
core/server/src/binary/handlers/messages/poll_messages_handler.rs:
##########
@@ -41,21 +46,76 @@ pub async fn handle_poll_messages(
let partition_id = req.partition_id;
let count = req.count;
let auto_commit = req.auto_commit;
+ let wait_timeout = Duration::from_micros(req.wait_timeout_us);
debug!(
"session: {session}, command: poll_messages, stream_id: {stream_id}, topic_id: {topic_id}, partition_id: {partition_id:?}"
);
shard.ensure_authenticated(session)?;
- let args = PollingArgs::new(strategy, count, auto_commit);
-
let user_id = session.get_user_id();
let client_id = session.client_id;
let topic = shard.resolve_topic_for_poll(user_id, &stream_id, &topic_id)?;
- let (metadata, mut batch) = shard
- .poll_messages(client_id, topic, consumer, partition_id, args)
+ let (mut metadata, mut batch) = shard
+ .poll_messages(
+ client_id,
+ topic,
+ consumer.clone(),
+ partition_id,
+ PollingArgs::with_wait_timeout(strategy, count, auto_commit, wait_timeout),
+ )
.await?;
+ if count > 0 && !wait_timeout.is_zero() && batch.is_empty() {
+ let namespaces =
+ shard.resolve_poll_wait_namespaces(topic, &consumer, client_id, partition_id)?;
+ let waiters = namespaces
+ .iter()
+ .filter_map(|namespace| shard.register_poll_waiter(*namespace, wait_timeout))
+ .collect::<Vec<_>>();
+
+ if let Some((next_metadata, next_batch)) = poll_wait_namespaces(
+ shard,
+ client_id,
+ topic,
+ &consumer,
+ &namespaces,
+ &strategy,
+ count,
+ auto_commit,
+ )
+ .await?
+ {
+ metadata = next_metadata;
+ batch = next_batch;
+ }
+
+ if batch.is_empty() && !waiters.is_empty() {
Review Comment:
Changed wake handling to loop and re-poll until the deadline.
##########
core/server/src/shard/waiters.rs:
##########
@@ -0,0 +1,218 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::shard::IggyShard;
+use ahash::AHashMap;
+use async_channel::{Receiver, Sender};
+use server_common::sharding::IggyNamespace;
+use std::{sync::Arc, sync::Mutex, time::Duration, time::Instant};
+
+const MAX_WAITERS_PER_NAMESPACE: usize = 1024;
+
+#[derive(Debug)]
+struct PollWaiter {
+ id: u64,
+ wake_sender: Sender<()>,
+ deadline: Option<Instant>,
+}
+
+#[derive(Debug, Default)]
+pub struct PollWaiterRegistry {
+ next_id: u64,
+ waiters: AHashMap<IggyNamespace, Vec<PollWaiter>>,
+}
+
+impl PollWaiterRegistry {
+ fn register(
+ &mut self,
+ namespace: IggyNamespace,
+ timeout: Duration,
+ ) -> Option<(u64, Receiver<()>)> {
+ self.prune_namespace(&namespace);
+ let waiters = self.waiters.entry(namespace).or_default();
+ if waiters.len() >= MAX_WAITERS_PER_NAMESPACE {
+ return None;
+ }
+
+ self.next_id = self.next_id.wrapping_add(1);
+ let id = self.next_id;
+ let (wake_sender, wake_receiver) = async_channel::bounded(1);
+ waiters.push(PollWaiter {
+ id,
+ wake_sender,
+ deadline: Instant::now().checked_add(timeout),
+ });
+
+ Some((id, wake_receiver))
+ }
+
+ fn remove(&mut self, namespace: &IggyNamespace, id: u64) {
+ let should_remove = if let Some(waiters) = self.waiters.get_mut(namespace) {
+ waiters.retain(|waiter| waiter.id != id);
+ waiters.is_empty()
+ } else {
+ false
+ };
+
+ if should_remove {
+ self.waiters.remove(namespace);
+ }
+ }
+
+ fn wake_namespace(&mut self, namespace: &IggyNamespace) {
+ let Some(waiters) = self.waiters.remove(namespace) else {
+ return;
+ };
+ for waiter in waiters {
+ let _ = waiter.wake_sender.try_send(());
+ }
+ }
+
+ fn wake_topic(&mut self, stream_id: usize, topic_id: usize) {
+ let namespaces = self
+ .waiters
+ .keys()
+ .copied()
+ .filter(|namespace| {
+ namespace.stream_id() == stream_id && namespace.topic_id() == topic_id
+ })
+ .collect::<Vec<_>>();
+ for namespace in namespaces {
+ self.wake_namespace(&namespace);
+ }
+ }
+
+ fn wake_stream(&mut self, stream_id: usize) {
+ let namespaces = self
+ .waiters
+ .keys()
+ .copied()
+ .filter(|namespace| namespace.stream_id() == stream_id)
+ .collect::<Vec<_>>();
+ for namespace in namespaces {
+ self.wake_namespace(&namespace);
+ }
+ }
+
+ fn prune_namespace(&mut self, namespace: &IggyNamespace) {
+ let now = Instant::now();
+ let should_remove = if let Some(waiters) = self.waiters.get_mut(namespace) {
+ waiters.retain(|waiter| {
+ !waiter.wake_sender.is_closed()
+ && waiter.deadline.is_none_or(|deadline| deadline > now)
+ });
+ waiters.is_empty()
+ } else {
+ false
+ };
+
+ if should_remove {
+ self.waiters.remove(namespace);
+ }
+ }
+}
+
+pub(crate) struct PollWaiterRegistration {
+ namespace: IggyNamespace,
+ id: u64,
+ receiver: Receiver<()>,
+ registry: Arc<Mutex<PollWaiterRegistry>>,
+}
+
+impl PollWaiterRegistration {
+ pub(crate) async fn wait(&self) -> bool {
+ self.receiver.recv().await.is_ok()
+ }
+}
+
+impl Drop for PollWaiterRegistration {
+ fn drop(&mut self) {
+ self.registry
+ .lock()
+ .expect("poll waiter registry poisoned")
Review Comment:
Replaced the waiter registry expects with poison recovery via into_inner().
##########
core/server/src/shard/waiters.rs:
##########
@@ -0,0 +1,218 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::shard::IggyShard;
+use ahash::AHashMap;
+use async_channel::{Receiver, Sender};
+use server_common::sharding::IggyNamespace;
+use std::{sync::Arc, sync::Mutex, time::Duration, time::Instant};
+
+const MAX_WAITERS_PER_NAMESPACE: usize = 1024;
+
+#[derive(Debug)]
+struct PollWaiter {
+ id: u64,
+ wake_sender: Sender<()>,
+ deadline: Option<Instant>,
+}
+
+#[derive(Debug, Default)]
+pub struct PollWaiterRegistry {
+ next_id: u64,
+ waiters: AHashMap<IggyNamespace, Vec<PollWaiter>>,
+}
+
+impl PollWaiterRegistry {
+ fn register(
+ &mut self,
+ namespace: IggyNamespace,
+ timeout: Duration,
+ ) -> Option<(u64, Receiver<()>)> {
+ self.prune_namespace(&namespace);
+ let waiters = self.waiters.entry(namespace).or_default();
+ if waiters.len() >= MAX_WAITERS_PER_NAMESPACE {
+ return None;
+ }
+
+ self.next_id = self.next_id.wrapping_add(1);
+ let id = self.next_id;
+ let (wake_sender, wake_receiver) = async_channel::bounded(1);
+ waiters.push(PollWaiter {
+ id,
+ wake_sender,
+ deadline: Instant::now().checked_add(timeout),
+ });
+
+ Some((id, wake_receiver))
+ }
+
+ fn remove(&mut self, namespace: &IggyNamespace, id: u64) {
+ let should_remove = if let Some(waiters) = self.waiters.get_mut(namespace) {
+ waiters.retain(|waiter| waiter.id != id);
+ waiters.is_empty()
+ } else {
+ false
+ };
+
+ if should_remove {
+ self.waiters.remove(namespace);
+ }
+ }
+
+ fn wake_namespace(&mut self, namespace: &IggyNamespace) {
+ let Some(waiters) = self.waiters.remove(namespace) else {
+ return;
+ };
+ for waiter in waiters {
+ let _ = waiter.wake_sender.try_send(());
+ }
+ }
+
+ fn wake_topic(&mut self, stream_id: usize, topic_id: usize) {
+ let namespaces = self
+ .waiters
+ .keys()
+ .copied()
+ .filter(|namespace| {
+ namespace.stream_id() == stream_id && namespace.topic_id() == topic_id
+ })
+ .collect::<Vec<_>>();
+ for namespace in namespaces {
+ self.wake_namespace(&namespace);
+ }
+ }
+
+ fn wake_stream(&mut self, stream_id: usize) {
+ let namespaces = self
+ .waiters
+ .keys()
+ .copied()
+ .filter(|namespace| namespace.stream_id() == stream_id)
+ .collect::<Vec<_>>();
+ for namespace in namespaces {
+ self.wake_namespace(&namespace);
+ }
+ }
+
+ fn prune_namespace(&mut self, namespace: &IggyNamespace) {
Review Comment:
Removed the dead deadline/prune path and the unused timeout argument.
##########
core/server/src/shard/system/messages.rs:
##########
@@ -682,14 +682,25 @@ pub struct PollingArgs {
pub strategy: PollingStrategy,
pub count: u32,
pub auto_commit: bool,
+ pub wait_timeout: Duration,
Review Comment:
Dropped the write-only timeout field from PollingArgs; timeout now stays in the binary handler.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org