This is an automated email from the ASF dual-hosted git repository. krishvishal pushed a commit to branch kafka-fetch-stall in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 46867d8ef803c06d388830b58aa95ead62cce358 Author: Krishna Vishal <[email protected]> AuthorDate: Wed Oct 28 11:18:00 2026 +0530 fix(gateways): keep Kafka Fetch answering when one partition stalls Each Iggy poll gets 2 s, the first polls of a request run together, and a partition whose poll timed out answers 6 for 2 s. --- gateways/kafka/Cargo.toml | 3 +- gateways/kafka/README.md | 31 +- gateways/kafka/docs/TEST_SUITE.md | 2 +- gateways/kafka/src/bridge/config.rs | 104 ++++-- gateways/kafka/src/bridge/iggy_bridge/fetch.rs | 91 +++-- gateways/kafka/src/bridge/iggy_bridge/mod.rs | 5 +- gateways/kafka/src/bridge/mod.rs | 4 +- gateways/kafka/src/protocol/api.rs | 5 + gateways/kafka/src/protocol/handlers/fetch.rs | 373 +++++++++++++++++---- .../kafka/tests/bridge_iggy_integration_tests.rs | 8 +- gateways/kafka/tests/common/iggy_server.rs | 6 +- gateways/kafka/tests/fetch_real_bridge_tests.rs | 159 ++++++++- gateways/kafka/tests/kafka_client_e2e_tests.rs | 6 +- 13 files changed, 657 insertions(+), 140 deletions(-) diff --git a/gateways/kafka/Cargo.toml b/gateways/kafka/Cargo.toml index 113598cc5..1861e0267 100644 --- a/gateways/kafka/Cargo.toml +++ b/gateways/kafka/Cargo.toml @@ -72,7 +72,8 @@ uuid = { workspace = true } x509-parser = { workspace = true } [dev-dependencies] -# Partition route requests for `examples/partition_routes.rs`. +# Partition route requests for `examples/partition_routes.rs`, and the Iggy request frames that a +# test relay holds back. iggy_binary_protocol = { workspace = true } # The client half, so a test encodes requests and decodes responses the way a consumer does. kafka-protocol = { workspace = true, features = ["client"] } diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md index 35f2b5957..c09cfc363 100644 --- a/gateways/kafka/README.md +++ b/gateways/kafka/README.md @@ -651,7 +651,7 @@ records, in request order. One uncompressed batch per partition. | ---- | ---- | ------------- | | `OFFSET_OUT_OF_RANGE` (1) | Offset below 0. Or an offset still past `high_watermark` after 30 s of 6 on the same connection, counted from its first 6 there. A read there in range starts the 30 s over. A node that lags clears within them. A partition with no offset past 0 never answers 1, because Iggy reports that while it loads, fences or rebuilds one. | Resets its offset (`auto.offset.reset`). | | `UNKNOWN_TOPIC_OR_PARTITION` (3) | Topic or partition not in Iggy, or a name Kafka refuses | Refreshes metadata, retries. | -| `NOT_LEADER_OR_FOLLOWER` (6) | Iggy unreachable, too slow, not signed in, loading the partition, returning no records where it reports some, or an offset past the end (see 1). Alone, it goes out after `max_wait_ms`. | Refreshes metadata, retries. | +| `NOT_LEADER_OR_FOLLOWER` (6) | Iggy unreachable, too slow (see the poll limit below), not signed in, loading the partition, returning no records where it reports some, or an offset past the end (see 1). Alone, it goes out after `max_wait_ms`. | Refreshes metadata, retries. | | `TOPIC_AUTHORIZATION_FAILED` (29) | The bridge's Iggy user lacks permission. Fetch reads as that user, not as the client. | Fails. | | `UNKNOWN_SERVER_ERROR` (-1) | Anything else, such as a stored message the gateway cannot map. Alone, it goes out after `max_wait_ms`. Each connection reads the partition again at most once per `max_wait_ms`. | Retries the same offset. | | `FETCH_SESSION_ID_NOT_FOUND` (70) | Top level. The request continues a session. | Sends a full request. | @@ -662,17 +662,21 @@ Limits: - 20 s per request, wait included, so a waiting Fetch fits the 25 s shutdown drain. Partitions not read by then answer `0` with no records. -- 4 requests read and encode at once. A waiting request, the probes that decide empty polls, or a - response on a slow socket hold no slot. So response memory grows with connections: `max_bytes` - each, or one larger record. +- `IGGY_KAFKA_FETCH_SLOTS` (4) requests read and encode at once. A waiting request, the probes + that decide empty polls, or a response on a slow socket hold no slot. So response memory grows + with connections: `max_bytes` each, or one larger record. - A message the gateway cannot map stops the consumer at its offset. Records before it are served. The log shows it once per connection at `warn`. -- Each slot polls on its own Iggy client, which connects at the slot's first poll. A poll ends at - the request deadline, so a partition that Iggy refuses holds up only its own slot. The SDK can - hold that connection for up to 30 s more, so the slot drops the client and connects a new one. -- Iggy polls by count, not bytes. One poll can load 16 of the largest messages: about 1 GB, and - about 4 GB over the 4 slots. It also costs polls: 1000 small records take 63. When Iggy polls - take a byte cap, both limits go away. +- Each slot polls on its own Iggy client, which connects when a request first takes the slot. + The first poll of each partition runs at the same time as the others of its request, 8 at a + time. So a slow partition does not hold up the partitions after it. +- One poll gets 2 s. A partition whose poll gets no answer in that time answers 6. For the next + 2 s it answers 6 without a poll, to every consumer, so the consumers of a stopped node's + partitions do not keep the slots that other consumers need. A poll that answers ends it. +- A poll asks Iggy for at most 1000 messages, and for the bytes left in the partition's budget, + but never less than 1 MiB. A request keeps the first page of each partition that it reads until + that partition's turn. Each such page holds at most the larger of that budget and 1 MiB, or one + larger record. - Probes run on one more Iggy client of their own, so they never wait behind a Produce. A request waits at most 2 s for its probes, then reads with what it has. - A request loads at most 100 new topic probes per round, the oldest first. Other topics use an @@ -701,6 +705,7 @@ Limits: | `IGGY_KAFKA_IGGY_STREAM` | `kafka` | Default Iggy stream for a Kafka topic with no explicit mapping override | | `IGGY_KAFKA_TOPIC_MAP_PATH` | unset | Path to a topic-mapping TOML file (see below); omit to use only the default rule | | `IGGY_KAFKA_IGGY_MAX_MESSAGE_SIZE` | `64MiB` | Iggy's `message_bus.max_message_size`. Set both together. Larger partitions answer 10 | +| `IGGY_KAFKA_FETCH_SLOTS` | `4` | Fetch requests that read from Iggy at once, each on an Iggy client of its own. At least `1` | | `IGGY_KAFKA_PRODUCE_SESSIONS` | `256` | Producers that get an Iggy client of their own. The rest share one. `0`: all share one | | `IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS` | `600` | An unused Produce client closes after this long | | `IGGY_KAFKA_IDEMPOTENT_PRODUCERS` | `false` | `true` passes the producer of an idempotent batch to Iggy, so each batch lands once. Turn it on after every Iggy node, gateway and consumer of the topics is upgraded. See [docs/IDEMPOTENCE.md](docs/IDEMPOTENCE.md) | @@ -864,9 +869,9 @@ With auto-creation off, the default, nothing changes: a missing topic is answere - 4 clients sign in at once, 80 a second at most. More wait, up to their request's deadline. - `IGGY_KAFKA_PRODUCE_SESSIONS=0`: all producers share one client, one request at a time. - Metadata, CreateTopics, DeleteTopics, CreatePartitions, DescribeConfigs, AlterConfigs and - IncrementalAlterConfigs share 1 more client, one Iggy request at a time. Fetch polls use 4 - more, one per read slot, and a `ListOffsets` lookup by time borrows one of those slots for its - request. Topic probes use 1 + IncrementalAlterConfigs share 1 more client, one Iggy request at a time. Fetch polls use + `IGGY_KAFKA_FETCH_SLOTS` (4) more, one per read slot, and a `ListOffsets` lookup by time + borrows one of those slots for its request. Topic probes use 1 more. With auto topic creation on, creations use 1 more, opened at the first one. Offset calls use 8 more: a commit client and a read client per offset slot. Reading Iggy's metadata change log uses 1 more, opened at startup. It reads the log every 250 ms. The listing of diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md index 37b99b338..0009b0efb 100644 --- a/gateways/kafka/docs/TEST_SUITE.md +++ b/gateways/kafka/docs/TEST_SUITE.md @@ -83,7 +83,7 @@ file under `tests/` anymore. | [`sasl_real_server_tests.rs`](../tests/sasl_real_server_tests.rs) | `IggyAuthenticator` against a real, spawned `iggy-server`: twice as many logins as a small session table holds all succeed, because each check signs out | No (needs the `iggy-server` binary - see Prerequisites) | | [`bridge_iggy_integration_tests.rs`](../tests/bridge_iggy_integration_tests.rs) | `IggyBridge` against a real, spawned `iggy-server` — provisioning idempotency, high watermark, credential/connection edge cases, the bridge dialing a TLS-only server over TLS, and a client and a SASL login that dial the next known node when the configured one is down | No (needs the `iggy-server` binary - see Prerequisites) | | [`list_offsets_real_bridge_tests.rs`](../tests/list_offsets_real_bridge_tests.rs) | ListOffsets (key 2) through the whole handler against a real, spawned `iggy-server`: LATEST, EARLIEST, lookups by time and `MAX_TIMESTAMP`, the `-1` answers, code 3, several partitions looked up by time in one request, and a stalled lookup leaving the later partitions their answers | No (needs the `iggy-server` binary - see Prerequisites) | -| [`fetch_real_bridge_tests.rs`](../tests/fetch_real_bridge_tests.rs) | Fetch (key 1) through the whole handler against a real, spawned `iggy-server` — records go in through the Iggy SDK or the Produce handler and come back through `kafka_protocol`'s client decoder, at every version, with codes 0, 1, 3, 6, -1 and 70, paged polls, and the wait | No (needs the `iggy-server` binary - see Prerequisites) | +| [`fetch_real_bridge_tests.rs`](../tests/fetch_real_bridge_tests.rs) | Fetch (key 1) through the whole handler against a real, spawned `iggy-server` — records go in through the Iggy SDK or the Produce handler and come back through `kafka_protocol`'s client decoder, at every version, with codes 0, 1, 3, 6, -1 and 70, paged polls, and the wait. A relay that drops the polls of some partitions shows that the others still answer at once | No (needs the `iggy-server` binary - see Prerequisites) | | [`produce_real_bridge_tests.rs`](../tests/produce_real_bridge_tests.rs) | Produce (key 0) through the whole handler against a real, spawned `iggy-server` — records go in as Kafka wire bytes and come back through the Iggy SDK, plus one error code per partition. Through a relay that cuts Iggy off, a send that misses its deadline answers 6, and the client's retry stores the batch once | No (needs the `iggy-server` binary - see Prerequisites) | | [`create_partitions_real_bridge_tests.rs`](../tests/create_partitions_real_bridge_tests.rs) | CreatePartitions (key 37) through the whole handler against a real, spawned `iggy-server`: a grow appends indices and shows in Metadata's listing, the grow-only rule (37), a missing topic (3), `validate_only`, duplicate names (42), the manual assignment checks (39), the per-request and topic caps, the `manage_topics` gate (29), a duplicate answered before authorization (42, then 29 for the res [...] | [`offsets_real_bridge_tests.rs`](../tests/offsets_real_bridge_tests.rs) | OffsetCommit (key 8) and OffsetFetch (key 9) through the whole handler against a real, spawned `iggy-server`: offsets stored as sent, the Iggy external group key, a negative commit as a delete, -1 for nothing committed, a null topic list, and a gateway restart ([#3542](https://github.com/apache/iggy/issues/3542) acceptance) | No (needs the `iggy-server` binary - see Prerequisites) | diff --git a/gateways/kafka/src/bridge/config.rs b/gateways/kafka/src/bridge/config.rs index 221bb45fc..5383c9033 100644 --- a/gateways/kafka/src/bridge/config.rs +++ b/gateways/kafka/src/bridge/config.rs @@ -39,6 +39,8 @@ pub(crate) const DEFAULT_IGGY_ADDR: &str = "127.0.0.1:8090"; const DEFAULT_IGGY_USERNAME: &str = "iggy"; /// Iggy's default `message_bus.max_message_size`. pub const DEFAULT_MAX_MESSAGE_SIZE: u64 = 64 * 1024 * 1024; +/// Fetch requests that read from Iggy at once, when `IGGY_KAFKA_FETCH_SLOTS` is unset. +pub const DEFAULT_FETCH_SLOTS: usize = 4; /// How the gateway reaches Iggy: the bridge's own clients and the SASL credential check alike. /// @@ -117,6 +119,36 @@ impl Default for ProduceSessionLimits { } } +impl ProduceSessionLimits { + /// The defaults, with `IGGY_KAFKA_PRODUCE_SESSIONS` and `IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS` + /// applied. + /// + /// # Errors + /// + /// [`BridgeError::InvalidConfig`] if either does not parse. + pub fn from_env() -> Result<Self, BridgeError> { + let mut limits = Self::default(); + if let Ok(value) = std::env::var("IGGY_KAFKA_PRODUCE_SESSIONS") { + limits.max = value.parse().map_err(|_| { + BridgeError::InvalidConfig(format!( + "IGGY_KAFKA_PRODUCE_SESSIONS '{value}' is not a count, for example 256" + )) + })?; + } + if let Ok(value) = std::env::var("IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS") { + // 0 would close a session between two requests of a busy producer. + let seconds = value.parse().ok().filter(|seconds| *seconds > 0); + limits.idle = seconds.map(Duration::from_secs).ok_or_else(|| { + BridgeError::InvalidConfig(format!( + "IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS '{value}' is not a number of seconds \ + above 0, for example 600" + )) + })?; + } + Ok(limits) + } +} + /// Connection + topic-mapping config for [`IggyBridge`](crate::bridge::iggy_bridge::IggyBridge). /// /// `Debug` is safe to derive: `password` is `SecretString`, which redacts on `Debug` by design @@ -130,6 +162,8 @@ pub struct IggyBridgeConfig { pub topic_mapping: TopicMapping, /// Iggy's `message_bus.max_message_size`, in bytes. Larger partitions answer 10. pub max_message_size: u64, + /// Fetch requests that read from Iggy at once, each on an Iggy client of its own. At least 1. + pub fetch_slots: usize, /// Transport security of every client the bridge opens. pub tls: IggyTls, pub produce_sessions: ProduceSessionLimits, @@ -161,6 +195,7 @@ impl IggyBridgeConfig { "IGGY_KAFKA_IGGY_TLS_ENABLED", "IGGY_KAFKA_IGGY_TLS_DOMAIN", "IGGY_KAFKA_IGGY_TLS_CA_FILE", + "IGGY_KAFKA_FETCH_SLOTS", "IGGY_KAFKA_PRODUCE_SESSIONS", "IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS", "IGGY_KAFKA_IDEMPOTENT_PRODUCERS", @@ -190,8 +225,8 @@ impl IggyBridgeConfig { /// `IGGY_KAFKA_IGGY_USERNAME` is set but empty, or if `IGGY_KAFKA_TOPIC_MAP_PATH` is set but /// the file is missing, fails to parse, or fails validation (e.g. a malformed mapping key or /// an empty `default_stream`), or if `IGGY_KAFKA_IGGY_MAX_MESSAGE_SIZE`, - /// `IGGY_KAFKA_PRODUCE_SESSIONS`, `IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS` or - /// `IGGY_KAFKA_IDEMPOTENT_PRODUCERS` does not parse. + /// `IGGY_KAFKA_FETCH_SLOTS`, `IGGY_KAFKA_PRODUCE_SESSIONS`, + /// `IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS` or `IGGY_KAFKA_IDEMPOTENT_PRODUCERS` does not parse. pub fn from_env() -> Result<Self, BridgeError> { let address = std::env::var("IGGY_KAFKA_IGGY_ADDR").unwrap_or_else(|_| DEFAULT_IGGY_ADDR.to_string()); @@ -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" + )) + })?, + }; + let tls = IggyTls::from_env().map_err(BridgeError::InvalidConfig)?; - let mut produce_sessions = ProduceSessionLimits::default(); - if let Ok(value) = std::env::var("IGGY_KAFKA_PRODUCE_SESSIONS") { - produce_sessions.max = value.parse().map_err(|_| { - BridgeError::InvalidConfig(format!( - "IGGY_KAFKA_PRODUCE_SESSIONS '{value}' is not a count, for example 256" - )) - })?; - } - if let Ok(value) = std::env::var("IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS") { - // 0 would close a session between two requests of a busy producer. - let seconds = value.parse().ok().filter(|seconds| *seconds > 0); - produce_sessions.idle = seconds.map(Duration::from_secs).ok_or_else(|| { - BridgeError::InvalidConfig(format!( - "IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS '{value}' is not a number of seconds \ - above 0, for example 600" - )) - })?; - } + let produce_sessions = ProduceSessionLimits::from_env()?; let idempotent_producers = match std::env::var("IGGY_KAFKA_IDEMPOTENT_PRODUCERS") { Ok(raw) => parse_bool("IGGY_KAFKA_IDEMPOTENT_PRODUCERS", &raw) @@ -292,6 +324,7 @@ impl IggyBridgeConfig { password: SecretString::from(password), topic_mapping, max_message_size, + fetch_slots, tls, produce_sessions, idempotent_producers, @@ -329,6 +362,7 @@ mod tests { topic_mapping: TopicMapping::new("kafka".to_string(), std::collections::HashMap::new()) .expect("valid mapping for this test's fixture data"), max_message_size: DEFAULT_MAX_MESSAGE_SIZE, + fetch_slots: DEFAULT_FETCH_SLOTS, tls: IggyTls::default(), produce_sessions: ProduceSessionLimits::default(), idempotent_producers: false, @@ -370,6 +404,7 @@ mod tests { std::env::remove_var("IGGY_KAFKA_IGGY_TLS_ENABLED"); std::env::remove_var("IGGY_KAFKA_IGGY_TLS_DOMAIN"); std::env::remove_var("IGGY_KAFKA_IGGY_TLS_CA_FILE"); + std::env::remove_var("IGGY_KAFKA_FETCH_SLOTS"); std::env::remove_var("IGGY_KAFKA_PRODUCE_SESSIONS"); std::env::remove_var("IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS"); } @@ -380,6 +415,7 @@ mod tests { let config = result.expect("valid config from documented defaults alone"); assert_eq!(config.max_message_size, DEFAULT_MAX_MESSAGE_SIZE); + assert_eq!(config.fetch_slots, DEFAULT_FETCH_SLOTS); assert!( !config.tls.enabled, "the Iggy hop is in the clear unless switched on" @@ -419,6 +455,34 @@ mod tests { assert!(matches!(garbage, Err(BridgeError::InvalidConfig(_)))); } + #[test] + #[serial] + fn from_env_reads_the_fetch_slot_count() { + // Safety: process-wide, not per-variable - see the note on + // from_env_uses_documented_defaults_when_only_password_is_set above. + unsafe { + std::env::set_var("IGGY_KAFKA_IGGY_PASSWORD", "iggy"); + std::env::set_var("IGGY_KAFKA_FETCH_SLOTS", "16"); + } + let sixteen = IggyBridgeConfig::from_env(); + unsafe { + std::env::set_var("IGGY_KAFKA_FETCH_SLOTS", "0"); + } + let none = IggyBridgeConfig::from_env(); + unsafe { + std::env::set_var("IGGY_KAFKA_FETCH_SLOTS", "some"); + } + let garbage = IggyBridgeConfig::from_env(); + unsafe { + std::env::remove_var("IGGY_KAFKA_IGGY_PASSWORD"); + std::env::remove_var("IGGY_KAFKA_FETCH_SLOTS"); + } + + assert_eq!(sixteen.unwrap().fetch_slots, 16); + assert!(matches!(none, Err(BridgeError::InvalidConfig(_)))); + assert!(matches!(garbage, Err(BridgeError::InvalidConfig(_)))); + } + #[test] #[serial] fn from_env_reads_the_produce_session_limits() { diff --git a/gateways/kafka/src/bridge/iggy_bridge/fetch.rs b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs index eae169e43..f8d7f4751 100644 --- a/gateways/kafka/src/bridge/iggy_bridge/fetch.rs +++ b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs @@ -37,16 +37,12 @@ use super::{ use crate::bridge::config::IggyBridgeConfig; use crate::bridge::error::BridgeError; -/// Fetch reads that poll at once. Each has its own Iggy client, so a poll that Iggy holds up -/// stalls only its own slot, and only until the request deadline. -const FETCH_SLOTS: usize = 4; - /// The byte cap of a poll of one message: the smallest Iggy takes. The first message comes /// whatever its size, and a larger cap only makes its owner reserve more read budget. #[allow(clippy::cast_possible_truncation)] pub const ONE_MESSAGE_MAX_BYTES: u32 = MIN_DEFERRED_POLL_MAX_BYTES as u32; -/// The Fetch read slots, with a client for each. +/// The Fetch read slots, with a client for each. A slot reads for one request at a time. pub(super) struct FetchPool { permits: Arc<Semaphore>, /// The clients of the free slots. @@ -56,10 +52,10 @@ pub(super) struct FetchPool { } impl FetchPool { - pub(super) fn new() -> Self { - let all: Vec<LazyClient> = (0..FETCH_SLOTS).map(|_| LazyClient::default()).collect(); + pub(super) fn new(slots: usize) -> Self { + let all: Vec<LazyClient> = (0..slots).map(|_| LazyClient::default()).collect(); Self { - permits: Arc::new(Semaphore::new(FETCH_SLOTS)), + permits: Arc::new(Semaphore::new(slots)), idle: Mutex::new(all.clone()), all, } @@ -90,9 +86,8 @@ impl FetchPool { /// An Iggy client that connects at its first call, so a bridge that never needs it opens no /// extra connection. /// -/// The SDK cannot cancel a call. One given up on can hold the connection until the SDK's 30 s -/// read deadline, and every later call on it waits behind. So a call that times out takes the -/// client with it, and the next call connects a new one. +/// A call through [`Self::until`] that times out takes the client with it, and the next call +/// connects a new one. Fetch polls keep theirs. See [`IggyBridge::poll_in`]. #[derive(Clone, Default)] pub(super) struct LazyClient(Arc<AsyncMutex<Option<Arc<IggyClient>>>>); @@ -317,8 +312,15 @@ impl FromIterator<(u32, PartitionProbe)> for TopicProbe { impl IggyBridge { /// A Fetch read slot, with its own client. `None` if none frees by `deadline`. + /// + /// Connects the slot's client first, by `deadline`, so a slow sign-in never counts against + /// the deadline of one poll. A connect that fails is left to the first poll, which reports it. pub(crate) async fn fetch_slot(&self, deadline: Instant) -> Option<FetchSlot> { - self.fetch_pool.take(deadline).await + let slot = self.fetch_pool.take(deadline).await?; + if let Ok(Err(error)) = timeout_at(deadline, slot.client.connected(&self.config)).await { + tracing::debug!(%error, "a Fetch read slot could not connect to Iggy"); + } + Some(slot) } /// Probes every partition of the topic `kafka_topic` maps to, with one `get_topic` call, or @@ -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 + /// `ListOffsets` can read by timestamp or the last record through the Fetch slots. Hands the + /// slot back with the result. `None` if no result comes by `deadline`. + /// + /// The errors are those of [`Self::poll_in`]. A result that comes at `deadline` is + /// [`BridgeError::Timeout`], and the slot connects a new client for its next poll. pub(crate) async fn poll_by( self: &Arc<Self>, slot: FetchSlot, @@ -609,12 +628,14 @@ mod tests { assert!(Arc::ptr_eq(&kept, &new)); } + const SLOTS: usize = 4; + #[tokio::test] async fn given_every_slot_taken_when_one_is_dropped_should_hand_its_client_to_the_next_read() { - let pool = Arc::new(FetchPool::new()); + let pool = Arc::new(FetchPool::new(SLOTS)); let soon = || Instant::now() + Duration::from_millis(20); let mut taken = Vec::new(); - for _ in 0..FETCH_SLOTS { + for _ in 0..SLOTS { taken.push(pool.take(soon()).await.expect("a free slot")); } for (index, slot) in taken.iter().enumerate() { @@ -640,7 +661,7 @@ mod tests { #[tokio::test] async fn given_a_passed_deadline_when_a_slot_is_free_should_take_none_and_keep_its_client() { - let pool = Arc::new(FetchPool::new()); + let pool = Arc::new(FetchPool::new(SLOTS)); let held = unconnected(); for client in &pool.all { client.hold(Arc::clone(&held)).await; diff --git a/gateways/kafka/src/bridge/iggy_bridge/mod.rs b/gateways/kafka/src/bridge/iggy_bridge/mod.rs index 59959931f..e88091930 100644 --- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs +++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs @@ -271,9 +271,9 @@ impl IggyBridge { let bridge = Self { client, + fetch_pool: Arc::new(FetchPool::new(config.fetch_slots)), config, produce_pool, - fetch_pool: Arc::new(FetchPool::new()), probe_client: LazyClient::default(), lookup_client: LazyClient::default(), auto_create_client: LazyClient::default(), @@ -571,6 +571,7 @@ impl IggyBridge { ) .expect("a valid mapping"), max_message_size: crate::bridge::DEFAULT_MAX_MESSAGE_SIZE, + fetch_slots: crate::bridge::DEFAULT_FETCH_SLOTS, produce_sessions: crate::bridge::ProduceSessionLimits::default(), tls: crate::bridge::IggyTls::default(), idempotent_producers: false, @@ -584,8 +585,8 @@ impl IggyBridge { scan_client: LazyClient::holding(Arc::clone(&client)), client, produce_pool: Arc::new(ProducePool::new(config.produce_sessions)), + fetch_pool: Arc::new(FetchPool::new(config.fetch_slots)), config, - fetch_pool: Arc::new(FetchPool::new()), probe_client: LazyClient::default(), lookup_client: LazyClient::default(), offset_pool: OffsetPool::new(), diff --git a/gateways/kafka/src/bridge/mod.rs b/gateways/kafka/src/bridge/mod.rs index bd4c24944..8acbd2256 100644 --- a/gateways/kafka/src/bridge/mod.rs +++ b/gateways/kafka/src/bridge/mod.rs @@ -28,7 +28,9 @@ pub mod iggy_bridge; pub mod nodes; pub mod topic_map; -pub use config::{DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig, IggyTls, ProduceSessionLimits}; +pub use config::{ + DEFAULT_FETCH_SLOTS, DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig, IggyTls, ProduceSessionLimits, +}; pub use error::BridgeError; pub use iggy_bridge::{ CLUSTER_ID_TOPIC, COORDINATOR_LEASE_GROUP, COORDINATOR_LOG_TOPIC, COORDINATOR_OWNER_TOPIC, diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs index a05a69103..4a1086cfa 100644 --- a/gateways/kafka/src/protocol/api.rs +++ b/gateways/kafka/src/protocol/api.rs @@ -292,6 +292,10 @@ pub struct GatewayState { /// Per Kafka topic and partition, how long `ListOffsets` lookups by time poll empty without /// proof that no record matches. Only the first of a spell warns. pub(crate) unproven_empty: Mutex<fetch::Spells>, + /// Per Kafka topic and partition, until when Fetch answers 6 without a poll, after Iggy did not + /// answer a poll in time. Shared, so the consumers of a partition whose node stopped do not + /// each spend a read slot waiting for it. + pub(crate) unreachable: Mutex<fetch::Unreachable>, /// Topic probes that every Fetch shares. pub(crate) probe_board: ProbeBoard, /// Held by `CreatePartitions` from reading a topic's partition count until its write is @@ -355,6 +359,7 @@ impl GatewayState { groups, loading: Mutex::default(), unproven_empty: Mutex::default(), + unreachable: Mutex::default(), probe_board: ProbeBoard::default(), partition_growth: Arc::default(), cluster_id: ClusterId::random(), diff --git a/gateways/kafka/src/protocol/handlers/fetch.rs b/gateways/kafka/src/protocol/handlers/fetch.rs index 5d4a09701..1ab790411 100644 --- a/gateways/kafka/src/protocol/handlers/fetch.rs +++ b/gateways/kafka/src/protocol/handlers/fetch.rs @@ -24,6 +24,7 @@ use std::task::Poll; use std::time::Duration; use bytes::Bytes; +use futures::stream::{FuturesUnordered, StreamExt}; use iggy::prelude::{ DEFAULT_POLL_MAX_BYTES, IGGY_MESSAGE_HEADER_SIZE, IggyError, IggyMessage, PollOptions, PolledMessages, @@ -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); + /// Most messages one partition read takes from Iggy, over all its polls. const MAX_POLL_COUNT: u32 = 1000; @@ -655,7 +665,7 @@ fn bytes_waiting(offset: u64, probe: PartitionProbe, budget: usize) -> u64 { } /// Where a requested offset stands against the probe. -#[derive(Debug, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] enum Position { /// Answer this code, with no records. Refused(i16), @@ -716,6 +726,9 @@ fn position(want: &Wanted<'_>, probes: &Probes) -> Position { } /// Reads partitions in request order, as Kafka does from v3, and spends one byte budget on them. +/// +/// The first poll of each partition runs ahead, at the same time as the others. See +/// [`Reader::first_polls`]. struct Reader<'a> { fetch: &'a Fetch<'a>, bytes_left: usize, @@ -723,11 +736,17 @@ struct Reader<'a> { /// the limits say (KIP-74). A record larger than every limit would stall the consumer /// otherwise. owes_first_record: bool, - /// Taken at the first poll, and freed before the probes that decide empty polls. A poll given - /// up on keeps it until the poll ends, and the slot then connects a new Iggy client. + /// Taken at the first poll, and freed before the probes that decide empty polls. slot: Option<FetchSlot>, } +/// One poll's answer, with when it went out. +struct PollOutcome { + asked: Instant, + /// `None` once the request ran out of time. + page: Option<core::result::Result<PolledMessages, BridgeError>>, +} + /// What reading one partition gave. enum Read { Done(PartitionData), @@ -768,30 +787,29 @@ impl<'a> Reader<'a> { /// One answer per requested partition, in request order, and the slot, if taken. async fn read_all(mut self, probes: &Probes) -> (Vec<PartitionData>, Option<FetchSlot>) { let fetch = self.fetch; - let stuck = &fetch.connection.stuck_offsets; + let positions: Vec<Position> = fetch + .wanted + .iter() + .map(|want| self.place(want, probes)) + .collect(); + let mut first_polls = self.first_polls(&positions).await; let mut answers = Vec::with_capacity(fetch.wanted.len()); let mut empty = Vec::new(); - for want in &fetch.wanted { - let read = match self.place(want, probes) { + for ((want, position), first) in fetch.wanted.iter().zip(positions).zip(&mut first_polls) { + let read = match position { Position::Refused(code) => Read::Done(refused(want.partition, code)), Position::CaughtUp { at, probe } => { self.placed(want, at); Read::Done(served(want.partition, probe.high_watermark, Bytes::new())) } - Position::Behind { at, offset, .. } - | Position::Unsure { at, offset, .. } - | Position::Loading { at, offset, .. } - if held(stuck, want.topic, at.index, offset, Instant::now()) => - { - Read::Done(refused(want.partition, ERROR_UNKNOWN_SERVER_ERROR)) - } Position::Behind { at, offset, probe } => { let polled = Polled::new(offset, probe.high_watermark); - self.read(want, at, offset, probe, polled).await + self.read(want, at, offset, probe, polled, first.take()) + .await } Position::Unsure { at, offset, probe } | Position::Loading { at, offset, probe } => { - self.settle(want, at, offset, probe).await + self.settle(want, at, offset, probe, first.take()).await } }; let answer = match read { @@ -812,8 +830,11 @@ 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 @@ -821,7 +842,96 @@ impl<'a> Reader<'a> { { 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)); + } + } + first_polls } /// Ends this connection's spell at `at`: it just read there in range. @@ -830,6 +940,7 @@ impl<'a> Reader<'a> { } /// Polls one partition that has records waiting, and answers with what its budget holds. + /// `first` stands in for its first poll. async fn read( &mut self, want: &Wanted<'_>, @@ -837,16 +948,23 @@ impl<'a> Reader<'a> { offset: u64, probe: PartitionProbe, mut polled: Polled, + mut first: Option<PollOutcome>, ) -> Read { let budget = want.max_bytes.min(self.bytes_left); if worth_polling(self.owes_first_record, budget, probe.average_size) { while let Some(count) = next_page(&polled, budget, probe.average_size) { - let asked = Instant::now(); let left = budget.saturating_sub(polled.bytes); - let Some(page) = self - .poll(want.topic, at.index, polled.next, count, left) - .await - else { + let outcome = self + .poll_or( + first.take(), + want.topic, + at.index, + polled.next, + page_options(count, left), + ) + .await; + let asked = outcome.asked; + let Some(page) = outcome.page else { break; }; match page { @@ -905,13 +1023,15 @@ impl<'a> Reader<'a> { /// - Nothing, and Iggy's commit ends just before `offset`: caught up. /// - Nothing else: a probe taken after the poll decides. See [`Self::decide`]. /// - /// No poll when nothing that fits could come back and no record is owed. + /// No poll when nothing that fits could come back and no record is owed. `first` stands in + /// for the poll. async fn settle( &mut self, want: &Wanted<'_>, at: ProbeAt, offset: u64, probe: PartitionProbe, + first: Option<PollOutcome>, ) -> Read { // Sized as a read's first page, since an empty answer costs the same round trip. let budget = want.max_bytes.min(self.bytes_left); @@ -925,14 +1045,22 @@ impl<'a> Reader<'a> { return Read::Done(unread); } let count = poll_count(budget, probe.average_size, u64::from(MAX_POLL_COUNT)); - let asked = Instant::now(); - match self.poll(want.topic, at.index, offset, count, budget).await { + let PollOutcome { asked, page } = self + .poll_or( + first, + want.topic, + at.index, + offset, + page_options(count, budget), + ) + .await; + match page { None => Read::Done(unread), Some(Err(error)) => Read::Done(self.refuse(want, at, offset, &error)), Some(Ok(page)) if !page.messages.is_empty() => { let mut polled = Polled::new(offset, probe.high_watermark); polled.add(page); - self.read(want, at, offset, probe, polled).await + self.read(want, at, offset, probe, polled, None).await } Some(Ok(page)) if caught_up(offset, page.current_offset) => { self.placed(want, at); @@ -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), } + Some(page) } /// Whether an empty poll at `offset` proves the consumer is at the end: Iggy's commit counter, @@ -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, + ); + if before.is_none_or(|until| now.saturating_duration_since(until) >= UNPLACED_GRACE) { + tracing::warn!( + kafka_topic = topic, + partition, + poll_deadline_ms = POLL_DEADLINE.as_millis(), + "Iggy did not answer a Fetch poll in time. Fetch answers 6 for the partition until a \ + poll answers" + ); + } else { + tracing::debug!( + kafka_topic = topic, + partition, + "Iggy still does not answer Fetch polls in time" + ); + } +} + +/// Ends [`mark_unreachable`] for `partition` of `topic`: a poll of it answered. +fn mark_reachable(map: &Mutex<Unreachable>, topic: &str, partition: u32) { + map.lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(topic, partition); +} + /// Per Kafka topic, then partition, the offset that one connection's Fetches cannot place. A read /// in range ends it. pub(crate) type Unplaced = PartitionMap<UnplacedOffset>; @@ -1613,6 +1837,15 @@ fn poll_count(budget: usize, average_size: u64, waiting: u64) -> u32 { u32::try_from(count).unwrap_or(MAX_POLL_COUNT) } +/// A poll of one page: at most `count` messages, and [`page_max_bytes`] bytes. +fn page_options(count: u32, budget: usize) -> PollOptions { + PollOptions { + count, + max_bytes: page_max_bytes(count, budget), + ..PollOptions::default() + } +} + /// The byte cap of a poll of `count` messages with `budget` bytes left. /// /// The budget, but never below Iggy's default cap, since the encoder trims the page to the @@ -2425,6 +2658,30 @@ mod tests { ); } + #[test] + fn given_a_poll_that_ran_out_of_time_when_placed_again_should_refuse_the_partition_for_a_while() + { + let map = Mutex::default(); + let start = Instant::now(); + assert!(!unreachable(&map, "a", 0, start), "never timed out"); + mark_unreachable(&map, "a", 0, start); + assert!(unreachable(&map, "a", 0, start)); + assert!(unreachable(&map, "a", 0, start + UNREACHABLE_FOR / 2)); + assert!(!unreachable(&map, "a", 1, start), "per partition"); + assert!(!unreachable(&map, "b", 0, start), "per topic"); + assert!( + !unreachable(&map, "a", 0, start + UNREACHABLE_FOR), + "the next poll then decides" + ); + mark_unreachable(&map, "a", 0, start + UNREACHABLE_FOR); + assert!(unreachable(&map, "a", 0, start + UNREACHABLE_FOR)); + mark_reachable(&map, "a", 0); + assert!( + !unreachable(&map, "a", 0, start + UNREACHABLE_FOR), + "a poll that answers ends it" + ); + } + #[test] fn given_repeated_sightings_when_counted_should_report_only_the_first_of_each_spell() { let spells = Mutex::default(); diff --git a/gateways/kafka/tests/bridge_iggy_integration_tests.rs b/gateways/kafka/tests/bridge_iggy_integration_tests.rs index 5eefe5d32..aeed9ea1f 100644 --- a/gateways/kafka/tests/bridge_iggy_integration_tests.rs +++ b/gateways/kafka/tests/bridge_iggy_integration_tests.rs @@ -40,9 +40,9 @@ use tokio::time::Instant; use iggy_gateway_kafka::auth::{IggyAuthenticator, SaslAuthenticator}; use iggy_gateway_kafka::bridge::{ - BridgeError, DEFAULT_MAX_MESSAGE_SIZE, IggyBridge, IggyBridgeConfig, IggyTls, KnownNodes, - OffsetCalls, PartitionBatch, ProduceSessionLimits, Producer, TopicCreationOutcome, - TopicMapping, TopicOverride, + BridgeError, DEFAULT_FETCH_SLOTS, DEFAULT_MAX_MESSAGE_SIZE, IggyBridge, IggyBridgeConfig, + IggyTls, KnownNodes, OffsetCalls, PartitionBatch, ProduceSessionLimits, Producer, + TopicCreationOutcome, TopicMapping, TopicOverride, }; use iggy_gateway_kafka::protocol::api::ERROR_REQUEST_TIMED_OUT; use iggy_gateway_kafka::protocol::sasl::PlainCredentials; @@ -620,6 +620,7 @@ async fn connect_to_unreachable_iggy_returns_err_not_panic() { topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new()) .expect("valid mapping for this test's fixture data"), max_message_size: DEFAULT_MAX_MESSAGE_SIZE, + fetch_slots: DEFAULT_FETCH_SLOTS, tls: IggyTls::default(), produce_sessions: ProduceSessionLimits::default(), idempotent_producers: false, @@ -647,6 +648,7 @@ async fn connect_to_a_black_hole_address_times_out_instead_of_hanging() { topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new()) .expect("valid mapping for this test's fixture data"), max_message_size: DEFAULT_MAX_MESSAGE_SIZE, + fetch_slots: DEFAULT_FETCH_SLOTS, tls: IggyTls::default(), produce_sessions: ProduceSessionLimits::default(), idempotent_producers: false, diff --git a/gateways/kafka/tests/common/iggy_server.rs b/gateways/kafka/tests/common/iggy_server.rs index 1c9ff1c88..dddea6490 100644 --- a/gateways/kafka/tests/common/iggy_server.rs +++ b/gateways/kafka/tests/common/iggy_server.rs @@ -34,8 +34,8 @@ use iggy::prelude::{AutoLogin, Client, Credentials, IggyClient, IggyClientBuilde use secrecy::SecretString; use iggy_gateway_kafka::bridge::{ - DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig, IggyTls, KnownNodes, ProduceSessionLimits, - TopicMapping, + DEFAULT_FETCH_SLOTS, DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig, IggyTls, KnownNodes, + ProduceSessionLimits, TopicMapping, }; /// Desired slot count - the actual count `port_band()` computes may be smaller if the machine's @@ -401,6 +401,7 @@ impl TestServer { topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new()) .expect("valid mapping for this test's fixture data"), max_message_size: DEFAULT_MAX_MESSAGE_SIZE, + fetch_slots: DEFAULT_FETCH_SLOTS, tls: IggyTls::default(), produce_sessions: ProduceSessionLimits::default(), idempotent_producers: true, @@ -609,6 +610,7 @@ impl TestCluster { topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new()) .expect("valid mapping for this test's fixture data"), max_message_size: DEFAULT_MAX_MESSAGE_SIZE, + fetch_slots: DEFAULT_FETCH_SLOTS, tls: IggyTls::default(), produce_sessions: ProduceSessionLimits::default(), idempotent_producers: true, diff --git a/gateways/kafka/tests/fetch_real_bridge_tests.rs b/gateways/kafka/tests/fetch_real_bridge_tests.rs index 231520505..9d5a5f1cd 100644 --- a/gateways/kafka/tests/fetch_real_bridge_tests.rs +++ b/gateways/kafka/tests/fetch_real_bridge_tests.rs @@ -21,13 +21,19 @@ //! `kafka_protocol`'s client-side decoder reads, the way a consumer would. use std::collections::BTreeMap; +use std::mem::offset_of; use std::sync::Arc; use std::time::Duration; use bytes::{Bytes, BytesMut}; +use futures::future::join_all; use iggy::prelude::{ HeaderKey, HeaderValue, Identifier, IggyMessage, MessageClient, Partitioning, TopicClient, }; +use iggy_binary_protocol::WireDecode; +use iggy_binary_protocol::codes::{POLL_MESSAGES_CODE, POLL_MESSAGES_DEFERRED_CODE}; +use iggy_binary_protocol::consensus::{HEADER_SIZE, Operation, RequestHeader}; +use iggy_binary_protocol::requests::messages::{DeferredPollMessagesRequest, PollMessagesRequest}; use kafka_protocol::indexmap::IndexMap; use kafka_protocol::messages::fetch_request::{FetchPartition, FetchTopic}; use kafka_protocol::messages::fetch_response::PartitionData; @@ -41,10 +47,14 @@ use kafka_protocol::records::{ RecordBatchDecoder, RecordBatchEncoder, RecordEncodeOptions, TimestampType, }; use serial_test::serial; +use tokio::io::{AsyncReadExt, AsyncWriteExt, copy}; +use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::watch; use tokio::time::Instant; use tokio_util::sync::CancellationToken; -use iggy_gateway_kafka::bridge::IggyBridge; +use iggy_gateway_kafka::bridge::{IggyBridge, IggyBridgeConfig}; use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig}; use iggy_gateway_kafka::protocol::api::{ API_KEY_FETCH, API_KEY_PRODUCE, ConnectionState, GatewayState, handle_connection_request, @@ -68,6 +78,11 @@ const VERSION: i16 = 12; const PARTITION_MAX_BYTES: i32 = 1024 * 1024; /// A Fetch that finds records or an error answers well inside this. const PROMPT: Duration = Duration::from_secs(5); +/// The handler's limit on one Iggy poll, and how long a partition then answers 6 without a poll. +const POLL_DEADLINE: Duration = Duration::from_secs(2); +const UNREACHABLE_FOR: Duration = Duration::from_secs(2); +/// Room for a request that makes no Iggy call that waits. +const AT_ONCE: Duration = Duration::from_millis(500); /// Spelled out rather than imported: a test reading its expected code from the same constant the /// handler answers with cannot catch that constant changing. @@ -81,7 +96,12 @@ const UNKNOWN_OFFSET: i64 = -1; /// A live gateway with the topic in place, since Fetch creates nothing. async fn gateway_with_topic(server: &TestServer, partitions: u32) -> GatewayState { - let bridge = IggyBridge::connect(server.test_config()) + gateway_on(server.test_config(), partitions).await +} + +/// [`gateway_with_topic`], with a bridge built from `config`. +async fn gateway_on(config: IggyBridgeConfig, partitions: u32) -> GatewayState { + let bridge = IggyBridge::connect(config) .await .expect("bridge should connect to a ready server"); bridge @@ -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"); + tokio::time::sleep(UNREACHABLE_FOR).await; + let response = fetch(&state, &stalled_one).await; + assert_eq!( + values(only(&response)), + vec![Some(Bytes::from_static(b"b"))] + ); +} diff --git a/gateways/kafka/tests/kafka_client_e2e_tests.rs b/gateways/kafka/tests/kafka_client_e2e_tests.rs index 29d8c0052..3fdcaf3ef 100644 --- a/gateways/kafka/tests/kafka_client_e2e_tests.rs +++ b/gateways/kafka/tests/kafka_client_e2e_tests.rs @@ -56,8 +56,8 @@ use tokio::task::JoinHandle; use iggy_gateway_kafka::auth::IggyAuthenticator; use iggy_gateway_kafka::bridge::{ - DEFAULT_MAX_MESSAGE_SIZE, IggyBridge, IggyBridgeConfig, IggyTls, KnownNodes, - ProduceSessionLimits, TopicMapping, + DEFAULT_FETCH_SLOTS, DEFAULT_MAX_MESSAGE_SIZE, IggyBridge, IggyBridgeConfig, IggyTls, + KnownNodes, ProduceSessionLimits, TopicMapping, }; use iggy_gateway_kafka::listener::{ListenerConfig, SecurityProtocol, parse_brokers}; use iggy_gateway_kafka::protocol::auto_topic_creation::AutoTopicCreationConfig; @@ -432,6 +432,7 @@ async fn root_bridge(server: &TestServer) -> IggyBridge { topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new()) .expect("the default mapping is valid"), max_message_size: DEFAULT_MAX_MESSAGE_SIZE, + fetch_slots: DEFAULT_FETCH_SLOTS, tls: IggyTls::default(), produce_sessions: ProduceSessionLimits::default(), idempotent_producers: true, @@ -2326,6 +2327,7 @@ async fn tls_stack() -> Option<TlsStack> { topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new()) .expect("the default mapping is valid"), max_message_size: DEFAULT_MAX_MESSAGE_SIZE, + fetch_slots: DEFAULT_FETCH_SLOTS, produce_sessions: ProduceSessionLimits::default(), tls: IggyTls::default(), idempotent_producers: true,
