numinnex commented on code in PR #4502:
URL: https://github.com/apache/iggy/pull/4502#discussion_r4256853854
##########
gateways/kafka/src/protocol/handlers/produce.rs:
##########
@@ -156,17 +162,14 @@ pub async fn handle(state: &GatewayState, api_version:
i16, body: Bytes) -> Hand
// Before planning, not after: the client waits through decompression too.
let deadline = Deadline::new(request.timeout_ms);
- let Ok(Ok(_slot)) = tokio::time::timeout_at(deadline.at,
state.produce_slots.acquire()).await
- else {
- tracing::debug!("Produce deadline passed while waiting for a slot");
- return if request.acks == ACKS_NONE {
- HandleOutcome::Close
- } else {
- respond_or_close(
- encode_uniform_response(api_version, &request,
ERROR_REQUEST_TIMED_OUT),
- "Produce",
- )
- };
+ let cost = decode_cost(&request, state.max_frame_size);
+ let Ok(Ok(_admission)) = tokio::time::timeout_at(deadline.at, admit(state,
cost)).await else {
Review Comment:
`_admission` is released when the handler returns at the deadline, but a
timed-out send keeps running in the task spawned by `send_in_slot`
(`iggy_bridge/mod.rs:101`) and still owns the decoded `messages` for up to
`SLOT_LIMIT` (45s).
Before, the single `send_slot` meant at most one such batch. Now it is one
per session, so up to 256 partition caps (~2 GiB at the default frame) sit
outside `produce_memory`, and that happens exactly when Iggy is slow.
Could `produce_memory` become an `Arc<Semaphore>` and the owned permit move
into the send future?
##########
gateways/kafka/src/protocol/handlers/produce.rs:
##########
@@ -156,17 +162,14 @@ pub async fn handle(state: &GatewayState, api_version:
i16, body: Bytes) -> Hand
// Before planning, not after: the client waits through decompression too.
let deadline = Deadline::new(request.timeout_ms);
- let Ok(Ok(_slot)) = tokio::time::timeout_at(deadline.at,
state.produce_slots.acquire()).await
- else {
- tracing::debug!("Produce deadline passed while waiting for a slot");
- return if request.acks == ACKS_NONE {
- HandleOutcome::Close
- } else {
- respond_or_close(
- encode_uniform_response(api_version, &request,
ERROR_REQUEST_TIMED_OUT),
- "Produce",
- )
- };
+ let cost = decode_cost(&request, state.max_frame_size);
+ let Ok(Ok(_admission)) = tokio::time::timeout_at(deadline.at, admit(state,
cost)).await else {
+ tracing::debug!("Produce deadline passed while waiting to decode");
+ return timed_out(api_version, &request);
+ };
+ let Some(session) = bridge.produce_session(producer_of(&request,
connection)) else {
Review Comment:
Admission is taken before the session, so a retry queued behind its
producer's stuck send holds a large slot plus its memory while it waits on
`turn`, up to the 20s deadline.
Four such producers pin every `PRODUCE_LARGE` slot and all other large
requests answer 7, which is the head-of-line blocking this PR removes elsewhere.
Taking the session and checking its turn before `admit` would avoid that.
##########
gateways/kafka/src/protocol/handlers/produce.rs:
##########
@@ -206,6 +209,71 @@ fn stub_outcome(api_version: i16, request:
&ProduceRequest) -> HandleOutcome {
respond_or_close(encode_response(api_version, request), "Produce")
}
+/// Answers 7 for every partition. `acks=0` has no response to carry it, so
that closes instead.
+fn timed_out(api_version: i16, request: &ProduceRequest) -> HandleOutcome {
+ if request.acks == ACKS_NONE {
+ HandleOutcome::Close
+ } else {
+ respond_or_close(
+ encode_uniform_response(api_version, request,
ERROR_REQUEST_TIMED_OUT),
+ "Produce",
+ )
+ }
+}
+
+/// What the request may hold decoded, in frame bytes, at most one partition
cap.
+///
+/// Partitions decode one at a time, so the largest counts. Uncompressed, a
partition holds its own
+/// bytes or a record slot per declared record, whichever is more. Compressed,
it may fill the cap.
+fn decode_cost(request: &ProduceRequest, max_frame_size: usize) -> usize {
+ let largest = request
+ .topic_data
+ .iter()
+ .flat_map(|topic| &topic.partition_data)
+ .filter_map(|partition| partition.records.as_deref())
+ .map(|blob| {
+ if is_compressed(blob) {
+ max_frame_size
+ } else {
+ let slots = declared_records(blob).unwrap_or(0);
+ blob.len().max(slots.saturating_mul(FRAME_BYTES_PER_RECORD))
+ }
+ })
+ .max()
+ .unwrap_or(0);
+ largest.min(max_frame_size)
+}
+
+/// A large slot when the request is large, and its decode memory.
+type Admission<'a> = (Option<SemaphorePermit<'a>>, SemaphorePermit<'a>);
+
+/// Waits for the request's decode memory. A large request,
`CODEC_OFF_WORKER_BYTES` or more,
+/// first waits for a large slot, so while it waits its turn it holds no
memory.
+async fn admit(
+ state: &GatewayState,
+ cost: usize,
+) -> std::result::Result<Admission<'_>, AcquireError> {
+ let large = if cost >= CODEC_OFF_WORKER_BYTES {
+ Some(state.produce_large.acquire().await?)
+ } else {
+ None
+ };
+ let permits = u32::try_from(cost).unwrap_or(u32::MAX);
+ let memory = state.produce_memory.acquire_many(permits).await?;
+ Ok((large, memory))
+}
+
+/// The producer id of the request's first batch, or the connection when that
batch has none.
+fn producer_of(request: &ProduceRequest, connection: &ConnectionState) ->
Producer {
+ request
+ .topic_data
+ .iter()
+ .flat_map(|topic| &topic.partition_data)
+ .find_map(|partition| partition.records.as_deref())
+ .and_then(producer_id)
+ .map_or_else(|| Producer::Connection(connection.id()), Producer::Id)
Review Comment:
For producers without an id, ordering does not survive a reconnect. Batch A
times out and its send keeps running in the background. The client reconnects,
gets a new `ConnectionId` and so a new session with its own turn, and the retry
of A or the next batch B lands before A.
The old global `send_slot` kept B behind A, and the README's
`max.in.flight=1` / `retries=0` advice relies on that.
Either keep the old connection's turn reachable until its send finishes, or
drop the ordering claim for these producers.
##########
gateways/kafka/src/protocol/api.rs:
##########
@@ -343,9 +347,15 @@ pub struct GatewayState {
pub(crate) probe_board: ProbeBoard,
}
-/// Each holds one decoded partition at a time, so about 160 MB at the default
8 MiB frame. Sends
-/// run one at a time anyway.
-const PRODUCE_SLOTS: usize = 4;
+/// Partition caps Produce may hold decoded at once, about 160 MB at the
default 8 MiB frame.
+///
+/// Counted in bytes, not requests: a small request takes a small share, so it
does not queue
+/// behind a few large ones.
+const PRODUCE_CAPS: usize = 4;
+
+/// Large Produce requests that decode and send at once. More at once only
contend for the
+/// gateway's workers and for Iggy: at full load they cost CPU and gave no
throughput.
+const PRODUCE_LARGE: usize = 4;
Review Comment:
With `PRODUCE_LARGE == PRODUCE_CAPS` and a compressed partition costing a
full cap, four compressed requests (even tiny lz4 ones) take all of
`produce_memory`, and FIFO `acquire_many` queues every small request behind
them. The doc at :352 says the opposite.
It also means compressed and >=64 KiB traffic is still capped at 4 in
flight, same as before this PR.
Could `PRODUCE_LARGE` be below `PRODUCE_CAPS`, with a test that admits 4
full-frame requests and then a small one?
##########
gateways/kafka/src/protocol/handlers/produce.rs:
##########
@@ -206,6 +209,71 @@ fn stub_outcome(api_version: i16, request:
&ProduceRequest) -> HandleOutcome {
respond_or_close(encode_response(api_version, request), "Produce")
}
+/// Answers 7 for every partition. `acks=0` has no response to carry it, so
that closes instead.
+fn timed_out(api_version: i16, request: &ProduceRequest) -> HandleOutcome {
+ if request.acks == ACKS_NONE {
+ HandleOutcome::Close
+ } else {
+ respond_or_close(
+ encode_uniform_response(api_version, request,
ERROR_REQUEST_TIMED_OUT),
+ "Produce",
+ )
+ }
+}
+
+/// What the request may hold decoded, in frame bytes, at most one partition
cap.
+///
+/// Partitions decode one at a time, so the largest counts. Uncompressed, a
partition holds its own
+/// bytes or a record slot per declared record, whichever is more. Compressed,
it may fill the cap.
+fn decode_cost(request: &ProduceRequest, max_frame_size: usize) -> usize {
+ let largest = request
+ .topic_data
+ .iter()
+ .flat_map(|topic| &topic.partition_data)
+ .filter_map(|partition| partition.records.as_deref())
+ .map(|blob| {
+ if is_compressed(blob) {
+ max_frame_size
+ } else {
+ let slots = declared_records(blob).unwrap_or(0);
+ blob.len().max(slots.saturating_mul(FRAME_BYTES_PER_RECORD))
+ }
+ })
+ .max()
+ .unwrap_or(0);
+ largest.min(max_frame_size)
+}
+
+/// A large slot when the request is large, and its decode memory.
+type Admission<'a> = (Option<SemaphorePermit<'a>>, SemaphorePermit<'a>);
+
+/// Waits for the request's decode memory. A large request,
`CODEC_OFF_WORKER_BYTES` or more,
+/// first waits for a large slot, so while it waits its turn it holds no
memory.
+async fn admit(
+ state: &GatewayState,
+ cost: usize,
+) -> std::result::Result<Admission<'_>, AcquireError> {
+ let large = if cost >= CODEC_OFF_WORKER_BYTES {
+ Some(state.produce_large.acquire().await?)
+ } else {
+ None
+ };
+ let permits = u32::try_from(cost).unwrap_or(u32::MAX);
+ let memory = state.produce_memory.acquire_many(permits).await?;
+ Ok((large, memory))
+}
+
+/// The producer id of the request's first batch, or the connection when that
batch has none.
+fn producer_of(request: &ProduceRequest, connection: &ConnectionState) ->
Producer {
Review Comment:
The pool key is the producer id straight from the client's batch header,
never checked against `ProducerIdAllocator` or scoped by principal.
A client that rotates ids forces a new Iggy connection plus an argon2 login
per request and evicts legitimate free sessions. Reusing someone else's id puts
the victim behind the attacker's stuck send (7 for up to 45s).
Suggest keying `Producer::Id` only for ids this gateway issued, as
(principal, id), falling back to the connection otherwise. Keying by connection
alone would bring the reconnect reorder above to idempotent producers.
##########
gateways/kafka/README.md:
##########
@@ -507,13 +512,16 @@ cut before that threshold is reached - worth knowing
rather than discovering lat
### Concurrency ceiling
-- One `IggyClient` serves Produce, Metadata, CreateTopics, DescribeConfigs and
AlterConfigs for
- every Kafka connection, one Iggy request at a time. Fetch polls use 4 more,
one per read slot.
- Topic probes use 1 more. Offset calls use 8 more: a commit client and a
read client per offset slot.
+- Each producer sends on an `IggyClient` of its own, one send at a time, so
its batches land in
+ order. A producer is its producer id, or its connection when its batches
carry none. Up to 256
+ of these stay open. Opening one closes the free ones idle for 10 minutes,
then the least
+ recently used free one if 256 are open. A Produce request that finds all 256
busy answers 7.
+- Metadata, CreateTopics, DeleteTopics, DescribeConfigs and AlterConfigs share
1 more client, one
+ Iggy request at a time. Fetch polls use 4 more, one per read slot. Topic
probes use 1 more.
+ Offset calls use 8 more: a commit client and a read client per offset slot.
- OffsetCommit and OffsetFetch cost one Iggy call per partition, spread over
the offset slots. In a
cluster, an offset commit client can also open one connection per node for
stores.
-- `IGGY_KAFKA_MAX_CONNECTIONS` does not change that. A Produce client pool is
a TODO in
- [docs/SCOPE.md](docs/SCOPE.md).
+- `IGGY_KAFKA_MAX_CONNECTIONS` does not change that.
Review Comment:
A few docs no longer match the code:
- this line: the session cap now interacts with `IGGY_KAFKA_MAX_CONNECTIONS`.
- README.md:209 says Produce ignores the producer id, but it is now the pool
key.
- README.md:457-459, "one send runs in the SDK at a time" and "ListOffsets
waits behind a stuck send", are stale. The 7 row at :353 also misses the
busy-pool case.
- docs/IDEMPOTENCE.md:138-142, "harmless while no code path reads the
producer id": this PR adds that reader without the safeguards listed there.
##########
gateways/kafka/src/bridge/iggy_bridge/produce.rs:
##########
@@ -19,58 +19,171 @@
//!
//! Separate from `fetch.rs` so the two never edit one file.
+use std::collections::HashMap;
use std::future::Future;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex, PoisonError};
+use std::time::Duration;
use iggy::prelude::{IggyError, IggyMessage, MessageClient, Partitioning};
use tokio::sync::Semaphore;
-use tokio::time::{Instant, timeout};
+use tokio::time::{Instant, timeout, timeout_at};
+use super::fetch::LazyClient;
use super::{IggyBridge, SLOT_LIMIT, TopicTarget, in_slot, permit_by};
use crate::bridge::error::BridgeError;
/// Frame bytes besides the messages: request header, batch header and ids.
const SEND_FRAME_ROOM: u64 = 4 * 1024;
+/// Produce sessions open at once, each an Iggy connection.
+const PRODUCE_SESSIONS: usize = 256;
+
+/// A free session unused this long closes when a new producer opens one.
+const SESSION_IDLE: Duration = Duration::from_secs(600);
+
+/// Who sends a Produce request. Each producer gets an Iggy session of its own.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum Producer {
+ /// The producer id of the request's record batches.
+ Id(i64),
+ /// The Kafka connection, for a producer without an id.
+ Connection(u64),
+}
+
+/// One producer's Iggy client, and one send at a time on it, so its sends
reach Iggy in order.
+pub struct ProduceSession {
+ client: LazyClient,
+ turn: Arc<Semaphore>,
+}
+
+/// The Produce sessions, one per producer, so different producers send at
once.
+pub(super) struct ProducePool {
+ sessions: Mutex<HashMap<Producer, Held>>,
+ capacity: usize,
+}
+
+/// A session, and when a request last took it.
+struct Held {
+ session: Arc<ProduceSession>,
+ used: Instant,
+}
+
+impl ProducePool {
+ pub(super) fn new() -> Self {
+ Self::with_capacity(PRODUCE_SESSIONS)
+ }
+
+ fn with_capacity(capacity: usize) -> Self {
+ Self {
+ sessions: Mutex::new(HashMap::new()),
+ capacity,
+ }
+ }
+
+ /// The session of `producer`, opened if it has none. `None` when the pool
is full and every
+ /// session is busy.
+ ///
+ /// Opening one first drops the free sessions idle past [`SESSION_IDLE`],
then the least
+ /// recently used free one if the pool is still full.
+ fn take(&self, producer: Producer) -> Option<Arc<ProduceSession>> {
+ let now = Instant::now();
+ let mut sessions =
self.sessions.lock().unwrap_or_else(PoisonError::into_inner);
+ if let Some(held) = sessions.get_mut(&producer) {
+ held.used = now;
+ return Some(Arc::clone(&held.session));
+ }
+ sessions
+ .retain(|_, held| !(held.is_free() &&
now.duration_since(held.used) >= SESSION_IDLE));
Review Comment:
Sessions are only swept here, when a new producer opens one. A
`Producer::Connection` session outlives its Kafka connection, and its heartbeat
keeps both the Iggy TCP connection and the server-side registration alive.
Short-lived non-idempotent clients (librdkafka default, kcat) hit this
constantly.
Can the session be removed and logged out on connection teardown, plus a
periodic sweep?
##########
gateways/kafka/src/bridge/iggy_bridge/produce.rs:
##########
@@ -191,4 +304,91 @@ mod tests {
Err(BridgeError::SendLost(IggyError::Disconnected))
));
}
+
+ #[tokio::test]
+ async fn given_one_producer_when_taken_twice_should_get_one_session() {
Review Comment:
These tests cover pool bookkeeping only. Nothing checks two producers
sending concurrently through `send_records`, busy pool -> 7 (and `acks=0` ->
close) in the handler, or eviction / `close()` actually shutting a client down.
`IggyBridge::close` also has no production caller, so the pool is never
closed on shutdown.
##########
gateways/kafka/src/bridge/iggy_bridge/produce.rs:
##########
@@ -19,58 +19,171 @@
//!
//! Separate from `fetch.rs` so the two never edit one file.
+use std::collections::HashMap;
use std::future::Future;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex, PoisonError};
+use std::time::Duration;
use iggy::prelude::{IggyError, IggyMessage, MessageClient, Partitioning};
use tokio::sync::Semaphore;
-use tokio::time::{Instant, timeout};
+use tokio::time::{Instant, timeout, timeout_at};
+use super::fetch::LazyClient;
use super::{IggyBridge, SLOT_LIMIT, TopicTarget, in_slot, permit_by};
use crate::bridge::error::BridgeError;
/// Frame bytes besides the messages: request header, batch header and ids.
const SEND_FRAME_ROOM: u64 = 4 * 1024;
+/// Produce sessions open at once, each an Iggy connection.
+const PRODUCE_SESSIONS: usize = 256;
+
+/// A free session unused this long closes when a new producer opens one.
+const SESSION_IDLE: Duration = Duration::from_secs(600);
+
+/// Who sends a Produce request. Each producer gets an Iggy session of its own.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum Producer {
+ /// The producer id of the request's record batches.
+ Id(i64),
+ /// The Kafka connection, for a producer without an id.
+ Connection(u64),
+}
+
+/// One producer's Iggy client, and one send at a time on it, so its sends
reach Iggy in order.
+pub struct ProduceSession {
+ client: LazyClient,
+ turn: Arc<Semaphore>,
+}
+
+/// The Produce sessions, one per producer, so different producers send at
once.
+pub(super) struct ProducePool {
+ sessions: Mutex<HashMap<Producer, Held>>,
+ capacity: usize,
+}
+
+/// A session, and when a request last took it.
+struct Held {
+ session: Arc<ProduceSession>,
+ used: Instant,
+}
+
+impl ProducePool {
+ pub(super) fn new() -> Self {
+ Self::with_capacity(PRODUCE_SESSIONS)
+ }
+
+ fn with_capacity(capacity: usize) -> Self {
+ Self {
+ sessions: Mutex::new(HashMap::new()),
+ capacity,
+ }
+ }
+
+ /// The session of `producer`, opened if it has none. `None` when the pool
is full and every
+ /// session is busy.
+ ///
+ /// Opening one first drops the free sessions idle past [`SESSION_IDLE`],
then the least
+ /// recently used free one if the pool is still full.
+ fn take(&self, producer: Producer) -> Option<Arc<ProduceSession>> {
+ let now = Instant::now();
+ let mut sessions =
self.sessions.lock().unwrap_or_else(PoisonError::into_inner);
+ if let Some(held) = sessions.get_mut(&producer) {
+ held.used = now;
+ return Some(Arc::clone(&held.session));
+ }
+ sessions
+ .retain(|_, held| !(held.is_free() &&
now.duration_since(held.used) >= SESSION_IDLE));
+ if sessions.len() >= self.capacity {
+ let oldest = sessions
+ .iter()
+ .filter(|(_, held)| held.is_free())
+ .min_by_key(|(_, held)| held.used)
+ .map(|(producer, _)| *producer)?;
+ sessions.remove(&oldest);
Review Comment:
Evicted clients are dropped with no logout (`TcpClient::shutdown` does not
log out either), so on the VSR server each one holds a client-table slot until
the 30s session timeout. `clients_table_max` is 8192 and never evicts live
entries.
With more than 256 active producers (the connection limit defaults to 1024)
the pool thrashes into connect + login per request, and above ~270 new
sessions/s logins are refused cluster-wide, for every client.
A shared overflow client when the pool is full, explicit logout on eviction
and a cap on the open rate would bound this. 256 / 600s are also hardcoded.
--
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]
