hubcio commented on code in PR #3948:
URL: https://github.com/apache/iggy/pull/3948#discussion_r3901861703
##########
core/common/src/traits/message_client.rs:
##########
@@ -41,6 +42,38 @@ pub trait MessageClient {
auto_commit: bool,
) -> Result<PolledMessages, IggyError>;
+ /// Poll messages and wait up to `wait_timeout` when no messages are
+ /// immediately available. A zero timeout preserves immediate polling.
+ /// Transports without deferred-poll support return `FeatureUnavailable`.
+ #[allow(clippy::too_many_arguments)]
+ async fn poll_messages_with_timeout(
Review Comment:
any non-zero timeout fails on every transport, so this is a public method
with no working path. it also freezes the shape (8 positional args, `Duration`)
before a server exists to validate it - hold the trait change until the wait
lands.
##########
core/server/src/dispatch.rs:
##########
@@ -2491,6 +2521,9 @@ where
S: 'static,
SB: SuperblockStore + 'static,
{
+ if wire.wait_timeout_us != 0 {
Review Comment:
dead check - `handle_poll_messages` already returned at line 2175 and the
http wire hardcodes 0. drop it.
##########
core/integration/tests/server/poll_semantics_vsr.rs:
##########
@@ -88,6 +88,49 @@ async fn
given_missing_partition_when_polling_should_reject_partition_not_found(
assert_eq!(valid.messages.len(), 0, "empty topic polls empty");
}
+#[iggy_harness(test_client_transport = [Tcp])]
+async fn
given_non_zero_wait_timeout_when_polling_should_reject_feature_unavailable(
Review Comment:
this pins the stub, not a behavior anyone wants. it goes away the day the
wait lands - drop it, or make it the real wait test.
##########
core/common/src/traits/binary_impls/messages.rs:
##########
@@ -37,11 +37,27 @@ use iggy_binary_protocol::requests::messages::{
FlushUnsavedBufferRequest, PollMessagesRequest, RawMessage,
SendMessagesEncoder,
};
use
iggy_binary_protocol::responses::consumer_groups::SyncConsumerGroupResponse;
+use std::time::Duration;
/// Max attempts to resolve a fenced consumer-group poll: one re-sync after the
/// coordinator rejects a stale assignment, then retry once.
const GROUP_POLL_MAX_ATTEMPTS: usize = 2;
+struct PollGroupOptions<'a> {
+ consumer: &'a Consumer,
+ strategy: &'a PollingStrategy,
+ count: u32,
+ auto_commit: bool,
+ wait_timeout_us: u64,
+}
+
+fn duration_to_wait_timeout_us(wait_timeout: Duration) -> Result<u64,
IggyError> {
Review Comment:
sub-microsecond timeouts truncate to 0 and silently become an immediate
poll. round up or reject.
##########
core/server/src/dispatch.rs:
##########
@@ -2115,6 +2115,32 @@ async fn evict_stale_client<B, MJ, S, SB>(
}
}
+/// Reject a non-zero wait timeout until active-server deferred waits are
implemented.
+async fn reject_deferred_poll<B, MJ, S, SB>(
Review Comment:
this is the whole server side: log and deny. the issue asks for the wait
itself - port the #3605 waiter to this server instead of stubbing.
##########
core/binary_protocol/src/requests/messages/poll_messages.rs:
##########
@@ -104,6 +108,13 @@ impl WireDecode for PollMessagesRequest {
pos += 4;
let auto_commit = read_u8(buf, pos)? != 0;
pos += 1;
+ let wait_timeout_us = if buf.len() == pos {
Review Comment:
presence is sniffed from the remaining length, so the next optional field
has to repeat the trick. a flags byte or a version would age better.
##########
core/common/src/traits/binary_impls/messages.rs:
##########
@@ -205,6 +225,7 @@ async fn poll_group_messages<B: BinaryClient>(
strategy: polling_strategy_to_wire(strategy),
count,
auto_commit,
+ wait_timeout_us,
Review Comment:
group polls pick one assigned partition round-robin, so a wait here parks
the consumer on that partition while the others may already have data. the
group case needs a design answer before the field goes on the wire.
--
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]
