spetz commented on code in PR #4706:
URL: https://github.com/apache/iggy/pull/4706#discussion_r4447204900
##########
gateways/kafka/src/protocol/handlers/fetch.rs:
##########
@@ -79,6 +80,15 @@ const UNPLACED_GRACE: Duration = Duration::from_secs(30);
/// How long the last good probe of a topic stands in while its refreshes fail.
const LAST_GOOD_TTL: Duration = Duration::from_secs(30);
+/// Longest one Iggy poll may take. A partition whose node stopped then
answers 6 long before the
+/// request deadline, and the other partitions of the request still answer.
+const POLL_DEADLINE: Duration = Duration::from_secs(2);
+/// First polls that one request runs at once. See [`Reader::first_polls`].
+const POLLS_AT_ONCE: usize = 8;
+/// How long a partition whose poll ran out of [`POLL_DEADLINE`] answers 6
without a poll, for the
+/// whole gateway. Its consumers then do not each spend a read slot on it.
+const UNREACHABLE_FOR: Duration = Duration::from_secs(2);
Review Comment:
The mark lasts exactly as long as one poll, and each request polls on its
own, so once a mark lapses the next Fetch that names the partition polls it
again and holds a slot for another 2 s before its timeout renews the mark.
Under a stall longer than 2 s the mark covers the slots for at most half of
every 4 s cycle, and a consumer that reads healthy and stalled partitions gets
every other response 2 s late, because its healthy pages wait for the stalled
first polls. Past 8 stalled partitions in one request the first polls run in
rounds of 8 (`gateways/kafka/src/protocol/handlers/fetch.rs:925`) and the
rounds are marked 2 s apart, so each later Fetch finds one round lapsed and
pays at least 2 s again. The README lines that say these consumers no longer
keep the slots and that a slow partition no longer holds up the ones after it
(`gateways/kafka/README.md:676`, `gateways/kafka/README.md:679`) hold only in
the first 2 s of a stall.
##########
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 {
Review Comment:
`place` reads the unreachable map at
`gateways/kafka/src/protocol/handlers/fetch.rs:854`, before this line waits on
the FIFO slot semaphore, and nothing reads it again once the slot is taken. A
Fetch that queued for a slot while the first timeout was running still polls
the partition, holds its slot for 2 s and renews the mark, and every healthy
Fetch queued behind it waits as well. With one slot and the relay from the new
test, a stalled Fetch at 0 ms, a second stalled Fetch at 100 ms and a healthy
Fetch at 200 ms answered at 2.04 s, 4.05 s and 4.05 s locally, so the healthy
consumer waited out two poll limits instead of one. With N such Fetches queued,
the 4 default slots stay busy for about N/4 times 2 s, which is the starvation
the description says this change removes.
##########
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());
Review Comment:
Any timeout marks the partition for every consumer of the gateway, but the
cause can be local to the one slot that timed out. Each slot client keeps its
own link per node and redials a lost one on its own jittered backoff
(`core/sdk/src/links/replicas.rs:1115`), and until it does, its polls retry in
`OperationDriver::hop` and hit the 2 s cut, so after a node restart one slow
slot refuses that node's partitions to consumers on slots whose links work. The
cut is also fixed, where before this PR a page had the 20 s request deadline. A
page that needs longer than 2 s, such as one large record over a slow link to
Iggy, times out on every retry, so that partition never serves again and
answers 6 to all its consumers. `mark_unreachable` warns only for the first
timeout and logs the rest at debug
(`gateways/kafka/src/protocol/handlers/fetch.rs:1536`), so a consumer stuck
this way leaves one warn line.
##########
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()
+{
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (stall, stalled) = watch::channel(true);
+ let mut config = relay_stalling(&server, &[1, 2], stalled).await;
+ // One read slot, so a poll that keeps it would hold up every other read
too.
+ config.fetch_slots = 1;
+ let state = gateway_on(config, 3).await;
+ for (partition, value) in [(0, b"a"), (1, b"b"), (2, b"c")] {
+ store(&server, partition, &[keyed(value)]).await;
+ }
+
+ // The stalled partitions come first, and their polls run at once. Read in
turn, each would
+ // hold up the partitions after it.
+ let all = request(TOPIC, &[(1, 0), (2, 0), (0, 0)]).with_max_wait_ms(0);
+ let started = Instant::now();
+ let response = fetch(&state, &all).await;
+ let took = started.elapsed();
+ let partitions = &response.responses[0].partitions;
+ assert_refused(&partitions[0], ERROR_NOT_LEADER_OR_FOLLOWER);
+ assert_refused(&partitions[1], ERROR_NOT_LEADER_OR_FOLLOWER);
+ assert_eq!(values(&partitions[2]), vec![Some(Bytes::from_static(b"a"))]);
+ assert!(
+ took >= POLL_DEADLINE && took < POLL_DEADLINE + AT_ONCE,
+ "the unanswered polls end together, at their own deadline: {took:?}"
+ );
+
+ // Their consumers now get 6 without a poll, and the one slot stays free
for partition 0.
+ let stalled_one = request(TOPIC, &[(1, 0)]).with_max_wait_ms(0);
+ let healthy = request(TOPIC, &[(0, 0)]).with_max_wait_ms(0);
+ let started = Instant::now();
+ let (refusals, other) = tokio::join!(
+ join_all((0..8).map(|_| fetch(&state, &stalled_one))),
+ fetch(&state, &healthy),
+ );
+ let took = started.elapsed();
+ for response in &refusals {
+ assert_refused(only(response), ERROR_NOT_LEADER_OR_FOLLOWER);
+ }
+ assert_eq!(values(only(&other)), vec![Some(Bytes::from_static(b"a"))]);
+ assert!(took < AT_ONCE, "nothing waits for partition 1: {took:?}");
+
+ // Once Iggy answers again, the first poll after the pause serves the
records.
+ stall.send(false).expect("the relay watches");
Review Comment:
The stall ends here, before the first mark lapses, so the test never shows
what a stall longer than 2 s does to the slots, which is when the mark stops
covering them. The relay also drops the stalled frames
(`gateways/kafka/tests/fetch_real_bridge_tests.rs:913`), so no late reply ever
comes back on the kept connection, and the test passes just the same if a
timed-out poll drops its client the way `LazyClient::until` does. Keeping the
client after a timeout has no test that fails without it.
##########
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))
}
- /// [`Self::poll`] of one message with any polling strategy, so
`ListOffsets` can read by
- /// timestamp or the last record through the same slots.
+ /// Reads one message of `partition` by `strategy`, with the client of
`slot`, so
Review Comment:
`ListOffsets` lookups by time still borrow these Fetch slots through
`poll_by`, and they did not get the new limits. Each lookup waits up to
`LOOKUP_BUDGET` (5 s) on a stalled partition, a request walks its partitions
one by one up to the 20 s deadline
(`gateways/kafka/src/protocol/handlers/list_offsets.rs:402`), nothing there
reads the unreachable map, and a timeout still drops the slot's client through
`LazyClient::until`. During the node outage this PR is about, four consumers
that seek by time on partitions of the stopped node hold every slot for up to
20 s, and Fetches of healthy partitions wait behind them as they did before.
##########
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(
Review Comment:
Only `Behind` partitions spend the planned `bytes_left`, so every `Unsure`
and `Loading` partition plans a first poll at the full remaining budget, at
least 1 MiB through `page_max_bytes`, and `first_polls` keeps every page until
the loop reaches it. A stale probe, which a failing `get_topic` leaves in place
for up to 30 s as in the node outage this PR targets, turns every caught-up
partition into `Unsure`, so a consumer of 300 partitions can hold up to 300 MiB
per request where the serial reads stopped near `max_bytes`. `read` and
`settle` then drop what no longer fits the real budget, so Iggy also served
those bytes for nothing.
##########
gateways/kafka/src/bridge/config.rs:
##########
@@ -260,25 +295,22 @@ impl IggyBridgeConfig {
})?,
};
+ let fetch_slots = match std::env::var("IGGY_KAFKA_FETCH_SLOTS") {
Review Comment:
Any count above 0 passes, and `FetchPool::new` builds every slot and a
semaphore of that size at startup
(`gateways/kafka/src/bridge/iggy_bridge/fetch.rs:55`). A value with a few zeros
too many runs the gateway out of memory, and one above tokio's
`Semaphore::MAX_PERMITS` panics, where every other bad value of this variable
gets an `InvalidConfig` that names it.
--
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]
