numinnex commented on code in PR #4706:
URL: https://github.com/apache/iggy/pull/4706#discussion_r4447277377
##########
gateways/kafka/src/protocol/handlers/fetch.rs:
##########
@@ -812,16 +830,108 @@ impl<'a> Reader<'a> {
(answers, self.slot)
}
- /// [`position`], with a partition that reads as loading refused 6 for
[`UNPLACED_GRACE`].
- /// After that a poll decides, so counters that drifted cannot stall a
consumer.
+ /// [`position`], refused when no poll may run: 6 for a partition that
reads as loading, for
+ /// [`UNPLACED_GRACE`], and for one whose poll ran out of time, for
[`UNREACHABLE_FOR`]. -1
+ /// while this connection holds the partition at the offset.
+ ///
+ /// Past the grace a poll decides a load, so counters that drifted cannot
stall a consumer.
fn place(&self, want: &Wanted<'_>, probes: &Probes) -> Position {
let position = position(want, probes);
if let Position::Loading { at, .. } = position
&& self.fetch.loading(want.topic, at.index)
{
return Position::Refused(ERROR_NOT_LEADER_OR_FOLLOWER);
}
- position
+ let (Position::Behind { at, offset, .. }
+ | Position::Unsure { at, offset, .. }
+ | Position::Loading { at, offset, .. }) = position
+ else {
+ return position;
+ };
+ let (stuck, now) = (&self.fetch.connection.stuck_offsets,
Instant::now());
+ if held(stuck, want.topic, at.index, offset, now) {
+ Position::Refused(ERROR_UNKNOWN_SERVER_ERROR)
+ } else if unreachable(&self.fetch.state.unreachable, want.topic,
at.index, now) {
+ Position::Refused(ERROR_NOT_LEADER_OR_FOLLOWER)
Review Comment:
**Critical.** `place` checks `unreachable` once, when the request starts
(`read_all` at :790-795), before `first_polls` waits for the FIFO slot at :919.
Nothing checks again before the poll, and nothing stops every queued request
from polling the stalled partition the moment the mark runs out. With
`UNREACHABLE_FOR == POLL_DEADLINE` (:85, :90), each consumer of a stalled
partition holds a slot about half the time.
Repro on this head: stall a partition, let the mark expire, then run 8
Fetches of it with 1 Fetch of a healthy partition. The healthy one takes 3.99 s
at the default 4 slots, 16.04 s at 1 slot. That is about K x 2 s / slots, so at
about 40 such connections healthy reads hit the 20 s deadline and come back
empty. README:679 says these consumers do not keep the slots others need.
Fix: check the mark again once the slot is taken, let one poll per partition
re-check it while the rest answer 6, and lengthen the mark after repeated
timeouts.
##########
gateways/kafka/src/protocol/handlers/fetch.rs:
##########
@@ -812,16 +830,108 @@ impl<'a> Reader<'a> {
(answers, self.slot)
}
- /// [`position`], with a partition that reads as loading refused 6 for
[`UNPLACED_GRACE`].
- /// After that a poll decides, so counters that drifted cannot stall a
consumer.
+ /// [`position`], refused when no poll may run: 6 for a partition that
reads as loading, for
+ /// [`UNPLACED_GRACE`], and for one whose poll ran out of time, for
[`UNREACHABLE_FOR`]. -1
+ /// while this connection holds the partition at the offset.
+ ///
+ /// Past the grace a poll decides a load, so counters that drifted cannot
stall a consumer.
fn place(&self, want: &Wanted<'_>, probes: &Probes) -> Position {
let position = position(want, probes);
if let Position::Loading { at, .. } = position
&& self.fetch.loading(want.topic, at.index)
{
return Position::Refused(ERROR_NOT_LEADER_OR_FOLLOWER);
}
- position
+ let (Position::Behind { at, offset, .. }
+ | Position::Unsure { at, offset, .. }
+ | Position::Loading { at, offset, .. }) = position
+ else {
+ return position;
+ };
+ let (stuck, now) = (&self.fetch.connection.stuck_offsets,
Instant::now());
+ if held(stuck, want.topic, at.index, offset, now) {
+ Position::Refused(ERROR_UNKNOWN_SERVER_ERROR)
+ } else if unreachable(&self.fetch.state.unreachable, want.topic,
at.index, now) {
+ Position::Refused(ERROR_NOT_LEADER_OR_FOLLOWER)
+ } else {
+ position
+ }
+ }
+
+ /// The first poll of each partition that polls, at most [`POLLS_AT_ONCE`]
at a time. Read in
+ /// turn, a partition that Iggy holds up would hold up each partition
after it, by up to
+ /// [`POLL_DEADLINE`] each.
+ ///
+ /// Plans the byte budget in request order, the way [`Self::read`] spends
it, from what the
+ /// probes expect. A partition with no budget in the plan polls in its
turn, if the reads before
+ /// it leave room. A first poll asks for what fits the planned budget, and
its read keeps what
+ /// fits the budget it really gets. So a request reads at most one page
more per partition than
+ /// it serves.
+ async fn first_polls(&mut self, positions: &[Position]) ->
Vec<Option<PollOutcome>> {
+ let fetch = self.fetch;
+ let mut planned = Vec::new();
+ let (mut bytes_left, mut owes_first_record) = (self.bytes_left,
self.owes_first_record);
+ for (index, (want, position)) in
fetch.wanted.iter().zip(positions).enumerate() {
+ let (at, offset, probe, behind) = match *position {
+ Position::Behind { at, offset, probe } => (at, offset, probe,
true),
+ Position::Unsure { at, offset, probe }
+ | Position::Loading { at, offset, probe } => (at, offset,
probe, false),
+ Position::Refused(_) | Position::CaughtUp { .. } => continue,
+ };
+ let budget = want.max_bytes.min(bytes_left);
+ if !worth_polling(owes_first_record, budget, probe.average_size) {
+ continue;
+ }
+ let count = if behind {
+ let expected = bytes_waiting(offset, probe, budget);
+ bytes_left =
+
bytes_left.saturating_sub(usize::try_from(expected).unwrap_or(usize::MAX));
+ owes_first_record &= expected == 0;
+ next_page(
+ &Polled::new(offset, probe.high_watermark),
+ budget,
+ probe.average_size,
+ )
+ } else {
+ Some(poll_count(
+ budget,
+ probe.average_size,
+ u64::from(MAX_POLL_COUNT),
+ ))
+ };
Review Comment:
**Warning.** Unsure and Loading partitions never lower the plan's
`bytes_left`, and Behind ones lower it only by the probe's estimate (:886-888),
while each page can fill `page_max_bytes`, at least 1 MiB. Every planned page
is then held until its partition's turn in `read_all`.
Repro on this head, partitions without stats: 20 pages (20.56 MB) held to
serve 1.02 MB. With stats it held 1 page. Before, a request held about one
response budget. Now it holds a page per planned partition, times
`IGGY_KAFKA_FETCH_SLOTS`, and the README line "response memory grows with
connections: `max_bytes` each" no longer holds.
Fix: prefetch only a window of `POLLS_AT_ONCE` partitions ahead of the read
cursor. Charging each page's `max_bytes` to the plan would cap it near 8
partitions and bring the head-of-line wait back.
##########
gateways/kafka/src/protocol/handlers/fetch.rs:
##########
@@ -812,16 +830,108 @@ impl<'a> Reader<'a> {
(answers, self.slot)
}
- /// [`position`], with a partition that reads as loading refused 6 for
[`UNPLACED_GRACE`].
- /// After that a poll decides, so counters that drifted cannot stall a
consumer.
+ /// [`position`], refused when no poll may run: 6 for a partition that
reads as loading, for
+ /// [`UNPLACED_GRACE`], and for one whose poll ran out of time, for
[`UNREACHABLE_FOR`]. -1
+ /// while this connection holds the partition at the offset.
+ ///
+ /// Past the grace a poll decides a load, so counters that drifted cannot
stall a consumer.
fn place(&self, want: &Wanted<'_>, probes: &Probes) -> Position {
let position = position(want, probes);
if let Position::Loading { at, .. } = position
&& self.fetch.loading(want.topic, at.index)
{
return Position::Refused(ERROR_NOT_LEADER_OR_FOLLOWER);
}
- position
+ let (Position::Behind { at, offset, .. }
+ | Position::Unsure { at, offset, .. }
+ | Position::Loading { at, offset, .. }) = position
+ else {
+ return position;
+ };
+ let (stuck, now) = (&self.fetch.connection.stuck_offsets,
Instant::now());
+ if held(stuck, want.topic, at.index, offset, now) {
+ Position::Refused(ERROR_UNKNOWN_SERVER_ERROR)
+ } else if unreachable(&self.fetch.state.unreachable, want.topic,
at.index, now) {
+ Position::Refused(ERROR_NOT_LEADER_OR_FOLLOWER)
+ } else {
+ position
+ }
+ }
+
+ /// The first poll of each partition that polls, at most [`POLLS_AT_ONCE`]
at a time. Read in
+ /// turn, a partition that Iggy holds up would hold up each partition
after it, by up to
+ /// [`POLL_DEADLINE`] each.
+ ///
+ /// Plans the byte budget in request order, the way [`Self::read`] spends
it, from what the
+ /// probes expect. A partition with no budget in the plan polls in its
turn, if the reads before
+ /// it leave room. A first poll asks for what fits the planned budget, and
its read keeps what
+ /// fits the budget it really gets. So a request reads at most one page
more per partition than
+ /// it serves.
+ async fn first_polls(&mut self, positions: &[Position]) ->
Vec<Option<PollOutcome>> {
+ let fetch = self.fetch;
+ let mut planned = Vec::new();
+ let (mut bytes_left, mut owes_first_record) = (self.bytes_left,
self.owes_first_record);
+ for (index, (want, position)) in
fetch.wanted.iter().zip(positions).enumerate() {
+ let (at, offset, probe, behind) = match *position {
+ Position::Behind { at, offset, probe } => (at, offset, probe,
true),
+ Position::Unsure { at, offset, probe }
+ | Position::Loading { at, offset, probe } => (at, offset,
probe, false),
+ Position::Refused(_) | Position::CaughtUp { .. } => continue,
+ };
+ let budget = want.max_bytes.min(bytes_left);
+ if !worth_polling(owes_first_record, budget, probe.average_size) {
+ continue;
+ }
+ let count = if behind {
+ let expected = bytes_waiting(offset, probe, budget);
+ bytes_left =
+
bytes_left.saturating_sub(usize::try_from(expected).unwrap_or(usize::MAX));
+ owes_first_record &= expected == 0;
+ next_page(
+ &Polled::new(offset, probe.high_watermark),
+ budget,
+ probe.average_size,
+ )
+ } else {
+ Some(poll_count(
+ budget,
+ probe.average_size,
+ u64::from(MAX_POLL_COUNT),
+ ))
+ };
+ if let Some(count) = count {
+ planned.push(PlannedPoll {
+ index,
+ topic: want.topic,
+ partition: at.index,
+ offset,
+ options: page_options(count, budget),
+ });
+ }
+ }
+ let mut first_polls: Vec<Option<PollOutcome>> =
std::iter::repeat_with(|| None)
+ .take(fetch.wanted.len())
+ .collect();
+ if planned.is_empty() {
+ return first_polls;
+ }
+ // With no slot by the deadline, each read finds none in its turn
either.
+ let Some(slot) = self.slot().await else {
+ return first_polls;
+ };
+ let mut planned = planned.into_iter();
+ let mut running: FuturesUnordered<_> = planned
+ .by_ref()
+ .take(POLLS_AT_ONCE)
+ .map(|poll| first_poll(fetch, slot, poll))
+ .collect();
+ while let Some((index, outcome)) = running.next().await {
+ first_polls[index] = Some(outcome);
+ if let Some(poll) = planned.next() {
+ running.push(first_poll(fetch, slot, poll));
+ }
+ }
Review Comment:
**Critical.** Each first poll gets its own 2 s (`poll_page` :1245), 8 at a
time, so a request with S stalled partitions holds its slot for about ceil(S/8)
x 2 s before the healthy partitions queued behind them get polled.
Repro on this head, default 4 slots: 9 stalled partitions in one request
take 4.07 s. 80 take 20.0 s, and a healthy partition with records comes back
empty at `give_up_at`, which is the #4515 symptom. README:676 says a slow
partition does not hold up the partitions after it.
Fix: one deadline for the whole first-poll phase. Partitions whose poll
never started answer 0 with no records rather than 6.
##########
gateways/kafka/src/bridge/config.rs:
##########
@@ -260,25 +295,22 @@ impl IggyBridgeConfig {
})?,
};
+ let fetch_slots = match std::env::var("IGGY_KAFKA_FETCH_SLOTS") {
+ Err(_) => DEFAULT_FETCH_SLOTS,
+ // With 0, no Fetch could read.
+ Ok(value) => value
+ .parse()
+ .ok()
+ .filter(|slots| *slots > 0)
+ .ok_or_else(|| {
+ BridgeError::InvalidConfig(format!(
+ "IGGY_KAFKA_FETCH_SLOTS '{value}' is not a count above
0, for example 4"
+ ))
+ })?,
+ };
Review Comment:
**Warning.** `IGGY_KAFKA_FETCH_SLOTS` is checked only for > 0. A huge value
reaches `FetchPool::new` (`iggy_bridge/fetch.rs:55-58`), where the `Vec` and
`Semaphore::new` panic or exhaust memory at startup instead of returning
`InvalidConfig`, and a typo like 40000 opens 40000 Iggy clients. `pub
fetch_slots: usize` (:166) also accepts 0 from a struct literal, and then every
Fetch waits 20 s and answers empty with no log.
`main.rs:315-323` already caps `IGGY_KAFKA_MAX_CONNECTIONS` at
`Semaphore::MAX_PERMITS` for this reason.
Fix: an upper bound here with a test and the README:712 row, and
`NonZeroUsize` for the field.
##########
gateways/kafka/tests/fetch_real_bridge_tests.rs:
##########
@@ -840,3 +860,138 @@ async fn
given_read_committed_when_fetching_should_set_last_stable_offset_to_hig
assert_eq!(partition.aborted_transactions, None);
assert_eq!(offsets(partition), vec![0, 1]);
}
+
+/// `server`'s bridge config, with every Iggy connection through a relay that
drops each poll of
+/// the `stalled` partitions while `stall` holds `true`. Iggy then never
answers those polls, as
+/// when the node of a partition stops.
+async fn relay_stalling(
+ server: &TestServer,
+ stalled: &'static [u32],
+ stall: watch::Receiver<bool>,
+) -> IggyBridgeConfig {
+ let listener = TcpListener::bind("127.0.0.1:0").await.expect("relay");
+ let mut config = server.test_config();
+ let relay = listener.local_addr().expect("relay address").to_string();
+ let upstream = std::mem::replace(&mut config.address, relay);
+ tokio::spawn(async move {
+ while let Ok((inbound, _)) = listener.accept().await {
+ let outbound =
TcpStream::connect(&upstream).await.expect("upstream");
+ let stall = stall.clone();
+ tokio::spawn(async move {
+ let (mut requests, mut replies) = inbound.into_split();
+ let (mut responses, mut writes) = outbound.into_split();
+ tokio::select! {
+ () = pass_requests(&mut requests, &mut writes, stalled,
&stall) => {}
+ _ = copy(&mut responses, &mut replies) => {}
+ }
+ });
+ }
+ });
+ config
+}
+
+/// Passes request frames on until a side closes, except each poll of a
`stalled` partition while
+/// `stall` holds `true`.
+async fn pass_requests(
+ requests: &mut OwnedReadHalf,
+ writes: &mut OwnedWriteHalf,
+ stalled: &[u32],
+ stall: &watch::Receiver<bool>,
+) {
+ let mut header = [0; HEADER_SIZE];
+ let size_at = offset_of!(RequestHeader, size);
+ loop {
+ if requests.read_exact(&mut header).await.is_err() {
+ return;
+ }
+ let size = u32::from_le_bytes(header[size_at..size_at +
4].try_into().expect("4 bytes"));
+ let mut body = vec![0; usize::try_from(size).expect("a frame size") -
HEADER_SIZE];
+ if requests.read_exact(&mut body).await.is_err() {
+ return;
+ }
+ if *stall.borrow()
+ && polled_partition(&header, &body).is_some_and(|polled|
stalled.contains(&polled))
+ {
+ continue;
+ }
+ if writes.write_all(&header).await.is_err() ||
writes.write_all(&body).await.is_err() {
+ return;
+ }
+ }
+}
+
+/// The partition that a request frame polls, if it is a poll. The SDK sends a
poll with options,
+/// as the bridge's are, as a deferred poll.
+fn polled_partition(header: &[u8; HEADER_SIZE], body: &[u8]) -> Option<u32> {
+ let code_at = offset_of!(RequestHeader, reserved);
+ let code = u32::from_le_bytes(header[code_at..code_at +
4].try_into().ok()?);
+ if header[offset_of!(RequestHeader, operation)] !=
Operation::NonReplicated as u8 {
+ return None;
+ }
+ match code {
+ POLL_MESSAGES_CODE =>
PollMessagesRequest::decode(body).ok()?.0.partition_id,
+ POLL_MESSAGES_DEFERRED_CODE => {
+ DeferredPollMessagesRequest::decode(body)
+ .ok()?
+ .0
+ .poll
+ .partition_id
+ }
+ _ => None,
+ }
+}
+
+#[tokio::test]
+#[serial]
+async fn
given_partitions_whose_polls_go_unanswered_when_fetching_should_serve_the_others_at_once()
Review Comment:
**Warning.** This test passes on this head while the two critical repros
above fail, and 7 mutants survive it: dropping the `mark_reachable` call in
`poll_page`, `POLLS_AT_ONCE = 2`, the old drop-client-on-timeout path through
`until`, discarding the client on timeout, no pre-connect in `fetch_slot`, and
both plan-budget mutants (no `bytes_left` drain, no `owes_first_record`
update). It only catches the `unreachable` check removed from `place` and
`POLLS_AT_ONCE = 1`.
The relay drops request frames and keeps the link up, so it cannot show a
stopped node, a reset or a failover either.
Fix: add a phase after the mark expires with more stalled Fetches than
slots, a request with more than 8 stalled partitions, and a unit test of the
`first_polls` budget plan.
##########
gateways/kafka/src/protocol/handlers/fetch.rs:
##########
@@ -1333,6 +1509,54 @@ pub(crate) fn sight_first(
.is_new()
}
+/// Per Kafka topic, then partition, until when a partition answers 6 without
a poll. See
+/// [`mark_unreachable`].
+pub(crate) type Unreachable = PartitionMap<Instant>;
+
+/// Whether `partition` of `topic` answers 6 without a poll at `now`.
+fn unreachable(map: &Mutex<Unreachable>, topic: &str, partition: u32, now:
Instant) -> bool {
+ map.lock()
+ .unwrap_or_else(PoisonError::into_inner)
+ .get(topic, partition)
+ .is_some_and(|until| now < *until)
+}
+
+/// Makes `partition` of `topic` answer 6 without a poll for
[`UNREACHABLE_FOR`] from `now`: Iggy
+/// did not answer a poll of it within [`POLL_DEADLINE`].
+///
+/// Gateway-wide, since the node behind a partition answers no consumer once
it stops. Warns when
+/// the partition answered a poll since, or ran out of time last more than
[`UNPLACED_GRACE`] ago.
+fn mark_unreachable(map: &Mutex<Unreachable>, topic: &str, partition: u32,
now: Instant) {
+ let before = map.lock().unwrap_or_else(PoisonError::into_inner).insert(
+ topic,
+ partition,
+ now + UNREACHABLE_FOR,
+ |until| now.saturating_duration_since(*until) < UNPLACED_GRACE,
+ );
Review Comment:
**Warning.** The mark is gateway-wide, but the timeout behind it can be
local to one slot's connection: a reply queued behind up to 7 others on the
same link, or a dead link to a live node. One slot's timeout then answers 6 for
that partition to every consumer for 2 s.
`insert` is also last-writer-wins. An older poll that times out after a
newer poll of the same partition answered re-marks a partition that is serving.
Fix: the same `answered_at` check as in `poll_page`. Mark only when no poll
of the partition has answered since this one went out.
##########
gateways/kafka/src/protocol/handlers/fetch.rs:
##########
@@ -1039,47 +1167,95 @@ impl<'a> Reader<'a> {
refused(want.partition, code)
}
- /// Polls one page of at most `count` messages and [`page_max_bytes`]
bytes, in the read slot.
- /// `None` once `give_up_at` passes.
- async fn poll(
+ /// `first`, if the partition polled ahead, or else a poll by `options` at
`offset`.
+ async fn poll_or(
&mut self,
+ first: Option<PollOutcome>,
topic: &str,
index: u32,
offset: u64,
- count: u32,
- budget: usize,
- ) -> Option<core::result::Result<PolledMessages, BridgeError>> {
- let give_up_at = self.fetch.give_up_at;
- if Instant::now() >= give_up_at {
- return None;
+ options: PollOptions,
+ ) -> PollOutcome {
+ if let Some(first) = first {
+ return first;
}
- let options = PollOptions {
- count,
- max_bytes: page_max_bytes(count, budget),
- ..PollOptions::default()
+ let asked = Instant::now();
+ let fetch = self.fetch;
+ let page = match self.slot().await {
+ Some(slot) => poll_page(fetch, slot, topic, index, offset,
options).await,
+ None => None,
};
- let bridge = self.fetch.bridge;
- let (page, slot) = bridge
- .poll(
- self.take_slot().await?,
- topic,
- index,
- offset,
- options,
- give_up_at,
- )
- .await?;
- self.slot = Some(slot);
- Some(page)
+ PollOutcome { asked, page }
+ }
+
+ /// The read slot, taken at the first poll. `None` if none frees by
`give_up_at`.
+ async fn slot(&mut self) -> Option<&FetchSlot> {
+ if self.slot.is_none() {
+ self.slot =
self.fetch.bridge.fetch_slot(self.fetch.give_up_at).await;
+ }
+ self.slot.as_ref()
}
+}
- /// The read slot, taken at the first poll. `None` once `give_up_at`
passes.
- async fn take_slot(&mut self) -> Option<FetchSlot> {
- match self.slot.take() {
- Some(slot) => Some(slot),
- None => self.fetch.bridge.fetch_slot(self.fetch.give_up_at).await,
+/// A first poll that [`Reader::first_polls`] plans for the partition at
`index` of the request.
+struct PlannedPoll<'a> {
+ index: usize,
+ topic: &'a str,
+ partition: u32,
+ offset: u64,
+ options: PollOptions,
+}
+
+/// Runs `poll`, and returns it with the place of its partition in the request.
+async fn first_poll(
+ fetch: &Fetch<'_>,
+ slot: &FetchSlot,
+ poll: PlannedPoll<'_>,
+) -> (usize, PollOutcome) {
+ let asked = Instant::now();
+ let page = poll_page(
+ fetch,
+ slot,
+ poll.topic,
+ poll.partition,
+ poll.offset,
+ poll.options,
+ )
+ .await;
+ (poll.index, PollOutcome { asked, page })
+}
+
+/// Polls one page in `slot`, for at most [`POLL_DEADLINE`]. `None` once the
request's
+/// `give_up_at` passes.
+///
+/// A poll that runs out of its own time makes the partition unreachable, and
one that answers
+/// makes it reachable again. See [`mark_unreachable`].
+async fn poll_page(
+ fetch: &Fetch<'_>,
+ slot: &FetchSlot,
+ topic: &str,
+ partition: u32,
+ offset: u64,
+ options: PollOptions,
+) -> Option<core::result::Result<PolledMessages, BridgeError>> {
+ let now = Instant::now();
+ if now >= fetch.give_up_at {
+ return None;
+ }
+ let deadline = (now + POLL_DEADLINE).min(fetch.give_up_at);
+ let page = fetch
+ .bridge
+ .poll_in(slot, topic, partition, offset, options, deadline)
+ .await;
+ let unreachable = &fetch.state.unreachable;
+ match page {
+ Err(BridgeError::Timeout) if deadline == fetch.give_up_at => return
None,
+ Err(BridgeError::Timeout) => {
+ mark_unreachable(unreachable, topic, partition, Instant::now());
}
+ _ => mark_reachable(unreachable, topic, partition),
}
Review Comment:
**Warning.** The 2 s cap has no back-off. A page that needs more than 2 s on
the wire is dropped, the partition answers 6 to every consumer, and the retry
asks for the same page, so it never lands.
Repro on this head: a 3 MiB message through a 1 MiB/s relay answered 6 on 4
of 4 attempts and was never served. On master the poll had the request's 20 s
and landed in about 3 s. The server finishes the disk read anyway
(`core/shard/src/lib.rs:2556-2557`), so each retry repeats only the transfer.
It needs one page over 2 s on its own: about 16 MiB on 100 Mbps, or a 64 MiB
record below about 256 Mbps.
Fix: record when each partition last answered, mark it only if nothing
answered since this poll went out, and let a single re-check use the request's
remaining time.
##########
gateways/kafka/src/bridge/iggy_bridge/fetch.rs:
##########
@@ -371,37 +373,54 @@ impl IggyBridge {
}
/// Reads up to `options.count` messages and `options.max_bytes` bytes of
`partition`, from
- /// `offset` on, with the client of `slot`. Hands the slot back with the
result. `None` if no
- /// result comes by `deadline`.
+ /// `offset` on, with the client of `slot`.
///
/// A plain consumer without auto commit, so Iggy stores no offset. Always
the request's
/// offset, never `Next`.
///
- /// The errors are those of [`Self::probe`]. A partition the topic lacks is
- /// [`BridgeError::Iggy`] here. The poll stops at `deadline` too, so a
result that comes then
- /// is [`BridgeError::Timeout`], and the slot connects a new client for
its next poll.
- pub(crate) async fn poll(
- self: &Arc<Self>,
- slot: FetchSlot,
+ /// Polls can share a slot at once. The SDK carries them on one
connection, and it drops the
+ /// reply of a poll that nobody waits for. So a poll that times out keeps
the client.
+ ///
+ /// # Errors
+ ///
+ /// The errors of [`Self::probe`]. A partition the topic lacks is
[`BridgeError::Iggy`] here.
+ /// [`BridgeError::Timeout`] if the poll does not end by `deadline`.
+ pub(crate) async fn poll_in(
+ &self,
+ slot: &FetchSlot,
kafka_topic: &str,
partition: u32,
offset: u64,
options: PollOptions,
deadline: Instant,
- ) -> Option<(Result<PolledMessages, BridgeError>, FetchSlot)> {
- self.poll_in_slot(
- slot,
- kafka_topic,
- partition,
- PollingStrategy::offset(offset),
- options,
- deadline,
- )
- .await
+ ) -> Result<PolledMessages, BridgeError> {
+ let target = self.topic_target(kafka_topic)?;
+ let (consumer, strategy) = (Consumer::default(),
PollingStrategy::offset(offset));
+ let poll = async {
+ let client = slot.client.connected(&self.config).await?;
+ client
+ .poll_messages(
+ &target.stream_id,
+ &target.topic_id,
+ partition,
+ &consumer,
+ &strategy,
+ options,
+ )
+ .await
+ .map_err(BridgeError::Iggy)
+ };
+ timeout_at(deadline, poll)
+ .await
+ .unwrap_or(Err(BridgeError::Timeout))
Review Comment:
**Warning.** Cutting the poll at 2 s while keeping the client stops the SDK
from ever handling the failure. The SDK marks a route failed only when the call
settles (`core/sdk/src/driver.rs:1124-1128`, `routes.failed` at :1204-1206),
and the dropped future never settles. Partition replies don't move the metadata
watermark either (`core/sdk/src/vsr.rs:103-104`).
When a primary's host goes silent without resetting connections, the slot's
client keeps sending to the dead node until its keepalive ping fails (5 s + 30
s). The partition then answers 6 for about 25-35 s after Iggy has failed over,
against about 10 s on master.
Fix: retire the slot's client on the second consecutive timeout of a
partition, or give the SDK a per-poll timeout so the call settles on its own.
--
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]
