This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new edb418bbe perf(gateways): give each Kafka producer its own Iggy client
(#4502)
edb418bbe is described below
commit edb418bbe395b7f05e8b32e9be2e7f9e4281402a
Author: Krishna Vishal <[email protected]>
AuthorDate: Wed Oct 14 13:29:56 2026 +0530
perf(gateways): give each Kafka producer its own Iggy client (#4502)
Relates to #3560
Produce through the gateway was slow because of three things in the
gateway itself, not in Iggy. This fixes them.
---
gateways/kafka/README.md | 65 +-
gateways/kafka/docs/IDEMPOTENCE.md | 37 +-
gateways/kafka/docs/SCOPE.md | 7 +-
gateways/kafka/src/bridge/config.rs | 88 ++-
gateways/kafka/src/bridge/iggy_bridge/fetch.rs | 34 +-
.../kafka/src/bridge/iggy_bridge/group_offsets.rs | 18 +-
gateways/kafka/src/bridge/iggy_bridge/mod.rs | 103 ++-
gateways/kafka/src/bridge/iggy_bridge/produce.rs | 819 ++++++++++++++++++---
gateways/kafka/src/bridge/mod.rs | 6 +-
gateways/kafka/src/main.rs | 43 +-
gateways/kafka/src/protocol/api.rs | 52 +-
.../src/protocol/handlers/init_producer_id.rs | 46 +-
gateways/kafka/src/protocol/handlers/mod.rs | 2 +-
gateways/kafka/src/protocol/handlers/produce.rs | 482 +++++++++++-
gateways/kafka/src/records.rs | 149 +++-
gateways/kafka/src/server.rs | 33 +-
.../kafka/tests/bridge_iggy_integration_tests.rs | 6 +-
gateways/kafka/tests/common/iggy_server.rs | 21 +-
gateways/kafka/tests/kafka_client_e2e_tests.rs | 3 +-
gateways/kafka/tests/produce_real_bridge_tests.rs | 190 ++++-
20 files changed, 1931 insertions(+), 273 deletions(-)
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index 3b88850b7..3c83aa194 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -53,7 +53,7 @@ Default bind: `127.0.0.1:9093`. Environment variables:
| `IGGY_KAFKA_IDLE_TIMEOUT_SECS` | `600` | Seconds a connection may sit idle
before the next frame's length prefix arrives |
| `IGGY_KAFKA_READ_TIMEOUT_SECS` | `15` | Seconds allowed to read a frame body
once its length prefix arrives |
| `IGGY_KAFKA_WRITE_TIMEOUT_SECS` | `10` | Seconds allowed to write a response
frame |
-| `IGGY_KAFKA_SHUTDOWN_DRAIN_TIMEOUT_SECS` | `25` | Seconds graceful shutdown
waits for in-flight connections before abandoning them |
+| `IGGY_KAFKA_SHUTDOWN_DRAIN_TIMEOUT_SECS` | `25` | Seconds graceful shutdown
waits for in-flight connections before abandoning them. The bridge then signs
its Iggy clients out, 5 s at most |
| `IGGY_KAFKA_INSTANCE_ID` | `0` | This gateway's number among the gateways
fronting one Iggy cluster. It is the high half of every producer id
`InitProducerId` hands out, and Kafka requires those to be cluster-unique, so
give every gateway its own value. A single gateway can leave it at `0`. |
| `IGGY_KAFKA_BRIDGE_ENABLED` | `false` | Connect the Iggy bridge at startup.
While false every API answers with its stub, and the `IGGY_KAFKA_IGGY_*`
variables below are read by nothing. A failed connection is fatal, not a
downgrade to stubs. |
| `IGGY_KAFKA_SASL_ENABLED` | `false` | Require SASL/PLAIN authentication
before serving any other API (`true` or `false`, nothing else) |
@@ -205,9 +205,9 @@ See [docs/SCOPE.md](docs/SCOPE.md) for
[#3421](https://github.com/apache/iggy/is
Delivery through this gateway is **at-least-once**, and stays at-least-once
across a gateway
restart. Transactions are not supported, and will not be. An idempotent Kafka
producer is given
-a producer id so that it starts, but its retries are not deduplicated: Produce
stores idempotent
-batches but ignores their producer id, epoch and sequence, so a retry after a
network timeout
-writes the record twice, and both copies reach the stream with their own
offsets.
+a producer id so that it starts, but its retries are not deduplicated: Produce
uses the producer
+id only to pick the producer's Iggy session, and ignores the epoch and
sequence, so a retry after a
+network timeout writes the record twice, and both copies reach the stream with
their own offsets.
Iggy deduplicates writes on its own partition plane, and that does not close
this gap, because it
guards the hop from the gateway to Iggy rather than the hop from the producer
to the gateway.
@@ -334,7 +334,8 @@ One partition, one Iggy send. Each partition answers for
itself.
| Topics | Creates none. Missing: 3. Bad name: 17. |
| `timeout_ms` | Honored, max 20 s. Past it: 7. |
| Compression | gzip, snappy, lz4. zstd from v7, else 76. |
-| Producer id, epoch, sequence | Ignored, so a retry writes twice. |
+| Producer id | Picks the producer's Iggy session. See [Concurrency
ceiling](#concurrency-ceiling). |
+| Producer epoch, sequence | Ignored, so a retry writes twice. |
| Timestamps over about 71 min apart | Clamped into one send. `kafka.ts` keeps
the real one. |
| Several batches in one partition | 87, as Kafka. |
| Bytes after the batch | 87. |
@@ -350,13 +351,22 @@ One partition, one Iggy send. Each partition answers for
itself.
| 87 | Record the gateway cannot map. Reason in `error_message` from v8. |
Fails. |
| 35 | Transactional or control batch | Fails. |
| 6 | Request budget ran out. Nothing written. | Retries. |
-| 7 | Deadline passed, or connection lost mid-send. May be written. | Retries.
Can duplicate. |
+| 7 | Deadline passed: in the send, waiting for decode memory, or behind the
producer's own earlier send. Or connection lost mid-send. May be written. |
Retries. Can duplicate. |
Partition cap: `max_frame_size` decompressed bytes, `max_frame_size / 64`
record slots, 3 headers
per slot. Past it: 10.
Request budget: 8 partition caps, refused partitions included. Past it: 6 for
the rest, not
-decoded. 4 requests decode at once.
+decoded.
+
+Decode memory: 5 partition caps across all requests, counted in bytes. A
request holds what its
+largest partition may decode: its bytes or 64 per record slot, whichever is
more, or a whole cap
+when compressed. One holding 64 KiB or more first takes one of 4 large slots,
and holds no memory
+while it waits for one. So large requests hold 4 caps at most, and leave 1 to
the small ones.
+
+A request waits for its producer's turn first, and holds no slot and no memory
while it does.
+Past its deadline waiting for any of the three: 7. A send that outlives its
request keeps its slot
+and its memory until it ends, 45 s at most.
[docs/BRIDGE_MAPPING.md](docs/BRIDGE_MAPPING.md) describes what a record
becomes once it is stored.
@@ -437,6 +447,8 @@ 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_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 |
The initial connect retries a fixed, bounded number of times
(`RECONNECTION_RETRIES = 3`, not the
Iggy SDK client's own default of unlimited retries, one dial per second,
forever), and the whole
@@ -450,9 +462,13 @@ Iggy becomes unreachable later.
`send_records` uses the Produce deadline instead:
-- One send runs in the SDK at a time.
-- A send past its deadline keeps that slot until it ends, up to 45 s. Other
sends answer 7 meanwhile.
-- ListOffsets needs no slot, but it waits behind a stuck send in the SDK. Past
`REQUEST_TIMEOUT` it answers 7.
+- One request of a producer runs at a time, on that producer's client.
+- A send past its deadline keeps the producer's turn until it ends. That
producer's requests answer
+ 7 meanwhile. Other producers send on.
+- After 45 s the gateway gives up on the send. It then closes that connection
before the turn
+ frees, so the send cannot land after a later one. The SDK ends its write
within 30 s more,
+ unless Iggy stopped reading: the producer then waits until that connection
breaks.
+- No other API waits behind a stuck send. Each uses another client.
### Topic mapping
@@ -507,15 +523,34 @@ 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 request at a time, so
its batches land
+ in order. Different producers send at once.
+- A producer is its producer id, if this gateway gave that id out since it
started, plus its user
+ when SASL is on. Every other producer is its Kafka connection: no id, an id
from before a
+ gateway restart, or an id from another gateway.
+- Up to `IGGY_KAFKA_PRODUCE_SESSIONS` (256) of these clients are open. A full
pool closes its
+ least recently used free client for a new producer, if that client sat
unused for 30 s.
+ Otherwise the producer shares 1 more client, one request at a time, and
moves to its own at a
+ later request, once there is room and none of its requests runs.
+- `IGGY_KAFKA_MAX_CONNECTIONS` (1024) allows more producers than sessions, so
raise the session
+ limit to give each its own.
+- Each client costs Iggy a connection, a slot of `metadata.clients_table_max`
(8192), and a slot
+ of `partition.dedup_clients_max` (4096) in each partition it writes.
+- A client whose producer is a connection closes, signed out, when that
connection ends. Any
+ other closes after `IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS` (600) unused,
checked every 30 s.
+- 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, 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).
- Order: set `max.in.flight.requests.per.connection=1`, or `retries=0`.
Otherwise a retried batch
can land after later ones.
+- Order after a 7: the send can still run. A producer with a producer id waits
behind it on any
+ connection. A producer that is its connection does not once it reconnects,
so its retry or its
+ next batch can land first. To keep that order use an idempotent producer,
the Java default, or
+ `IGGY_KAFKA_PRODUCE_SESSIONS=0`.
Fetches share topic probes: one `get_topic` per topic at a time, for all
consumers together, on
the probe client. A caught-up consumer of a busy topic can start one per
Fetch. A waiting Fetch
diff --git a/gateways/kafka/docs/IDEMPOTENCE.md
b/gateways/kafka/docs/IDEMPOTENCE.md
index 11e9d57b8..baae8497b 100644
--- a/gateways/kafka/docs/IDEMPOTENCE.md
+++ b/gateways/kafka/docs/IDEMPOTENCE.md
@@ -51,12 +51,9 @@ Kafka producer id therefore needs one bound session per
producer.
The request id is different. The transmute copies the header and replaces only
`group`, `client`,
`session` and `user_id`, so a caller-supplied request id survives to the
`ClientTable`.
-One session per producer is a connection pool keyed by producer id. That is
the pool the
-README's "Concurrency ceiling" section already owes before
-[#3535](https://github.com/apache/iggy/issues/3535) and
-[#3536](https://github.com/apache/iggy/issues/3536).
-
-Two SDK seams are missing for it:
+One session per producer is a connection pool keyed by producer id. The
gateway has that pool, for
+throughput: see the README's "Concurrency ceiling". Retries are still not
deduplicated, because two
+SDK seams are missing:
- a caller-chosen client id at build time. `ConsensusSession::with_client_id`
is public, but both
construction sites build with `ConsensusSession::new()`
@@ -103,11 +100,11 @@ refused with `RequestTooOld`.
## Decision
-Allocate only, and defer the pool.
+Allocate only, and defer deduplication.
-Rejecting the stock producer to avoid the pool trades away the one requirement
the maintainers
+Rejecting the stock producer to avoid that work trades away the one
requirement the maintainers
named. It buys a guarantee nobody is asking for yet. Allocating costs about a
day, leaves
-delivery where it already is, and blocks nothing the pool later needs.
+delivery where it already is, and blocks nothing deduplication later needs.
## Behavior
@@ -130,7 +127,7 @@ a birthday collision, not a remote one.
The id is a pool key, not a dedup identity. Under the design above, the dedup
identity is the
session's own random client id, minted at register. The producer id only
decides which connection
serves a producer. Kafka still requires it to be unique across the cluster,
which is what the
-instance number buys. It also has to stay unique across a restart once the
pool lands or
+instance number buys. It also has to stay unique across a restart once dedup
lands or
Produce persists, because `producer_epoch` is always 0, so a replayed id is a
replayed
`(producer_id, producer_epoch)` pair a live producer may still hold. The
counter therefore starts
at the wall clock in milliseconds rather than at 0. A restarted gateway starts
above every id its
@@ -139,11 +136,13 @@ uptime, or the clock stepped back across the restart. A
clock that reads before
seeds 0 and replays from the bottom. 2^47 milliseconds leaves the counter
space thousands of years
from running out.
-Nothing is persisted for this, so the clock seed narrows the replay window
without closing it. That
-is harmless while no code path reads the producer id. It stops being harmless
once one does:
-[#3535](https://github.com/apache/iggy/issues/3535) must first either persist
a high-water mark
-and seed the counter above it, or bump `producer_epoch` on every gateway
start. Either one makes a
-replayed `(producer_id, producer_epoch)` pair impossible rather than unlikely.
+Nothing is persisted for this, so the clock seed narrows the replay window
without closing it.
+Produce reads the id as a pool key, and only an id this run handed out: any
other id keys by
+connection. A replayed pair therefore makes two producers of one user share a
session and send one
+at a time, and nothing more. It stops being harmless once deduplication reads
the id. That work
+must first either persist a high-water mark and seed the counter above it, or
bump
+`producer_epoch` on every gateway start. Either one makes a replayed
+`(producer_id, producer_epoch)` pair impossible rather than unlikely.
`IGGY_KAFKA_INSTANCE_ID` defaults to 0 so a single gateway needs no
configuration. Two gateways
left on the default draw from the same 47-bit space and can hand out identical
ids, on the first
@@ -179,14 +178,15 @@ for consumer groups and has to refuse a `TXN`-type
coordinator lookup explicitly
Produce:
-- accept `producer_id`, `producer_epoch` and `base_sequence` on the request
and ignore them
+- accept `producer_id`, `producer_epoch` and `base_sequence` on the request.
The id picks the
+ producer's session when this run handed it out. The epoch and the sequence
are ignored
- never answer `OUT_OF_ORDER_SEQUENCE_NUMBER` (45) or
`DUPLICATE_SEQUENCE_NUMBER` (46)
- reject a non-empty `transactional_id` (v3+) with `UNSUPPORTED_VERSION` (35),
at the partition
level, keeping the connection open. Under `acks=0` there is no response to
carry 35, so the
gateway closes the connection instead, as a Kafka broker does on any
`acks=0` produce error.
The check runs before the `acks=0` branch, so no write path can see a
transactional batch
-Those first two codes stay unsent even once the pool lands. The watermark
accepts any request
+Those first two codes stay unsent even once dedup lands. The watermark accepts
any request
above it without noticing a gap, so a gap cannot be told apart from ordinary
traffic. Sending
either code claims a detection the gateway does not have.
@@ -235,8 +235,7 @@ A producer that holds an id believes its retries are
deduplicated. They are not.
network timeout writes the record twice, and both copies reach the stream with
their own offsets.
Iggy's own deduplication does not help, because it guards the other hop.
Delivery through the
-gateway is at-least-once until the pool lands, and at-least-once across a
gateway restart after
-that.
+gateway is at-least-once until dedup lands, and at-least-once across a gateway
restart after that.
The README states that limitation under "Delivery guarantees", so a user does
not have to infer
it from the presence of key 22.
diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md
index 91d2e7e36..b8cd2d962 100644
--- a/gateways/kafka/docs/SCOPE.md
+++ b/gateways/kafka/docs/SCOPE.md
@@ -193,7 +193,8 @@ below it are still open for the issues that build on top of
it.
- [x] Produce → `send_messages`
([#3535](https://github.com/apache/iggy/issues/3535)) - one call
per partition, base offset from the send confirmation, one error code
per partition. See
[README.md](../README.md#produce-3535).
-- [ ] Produce: `IggyClient` pool. Pin each partition to one client, so order
holds.
+- [x] Produce: `IggyClient` pool, one client per producer, so a producer's
batches land in order.
+ See [README.md](../README.md#concurrency-ceiling) and
[IDEMPOTENCE.md](IDEMPOTENCE.md).
- [ ] Produce: write keyed records' header TLVs into one buffer
(`records::to_iggy`). Benchmark
first. Keep Iggy's TLV layout and the 100 KB header check.
- [ ] Produce: skip the second batch CRC pass (`records::preflight`) only if
profiling shows it
@@ -407,8 +408,8 @@ InitProducerId and idempotent producers
([#3545](https://github.com/apache/iggy/issues/3545)):
[`IDEMPOTENCE.md`](IDEMPOTENCE.md).
- [x] InitProducerId (22) allocates a producer id so a stock idempotent
producer starts; a
- transactional request is refused. The producer-id-keyed connection pool
that would make
- retries deduplicated is deferred - delivery stays at-least-once
+ transactional request is refused
+- [ ] Deduplicate an idempotent producer's retries. Delivery stays
at-least-once until then
Authentication design ([#3549](https://github.com/apache/iggy/issues/3549)):
[`AUTHENTICATION.md`](AUTHENTICATION.md).
diff --git a/gateways/kafka/src/bridge/config.rs
b/gateways/kafka/src/bridge/config.rs
index 8e8de7af1..4f77661b1 100644
--- a/gateways/kafka/src/bridge/config.rs
+++ b/gateways/kafka/src/bridge/config.rs
@@ -16,6 +16,7 @@
// under the License.
use std::path::Path;
+use std::time::Duration;
use iggy::prelude::IggyByteSize;
use secrecy::SecretString;
@@ -37,6 +38,25 @@ const DEFAULT_IGGY_USERNAME: &str = "iggy";
/// Iggy's default `message_bus.max_message_size`.
pub const DEFAULT_MAX_MESSAGE_SIZE: u64 = 64 * 1024 * 1024;
+/// How many producers get an Iggy session of their own, and how long one
stays open unused.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ProduceSessionLimits {
+ /// Sessions open at once, each an Iggy connection. Producers past it
share one more. With 0
+ /// every producer shares that one, and sends run one at a time.
+ pub max: usize,
+ /// A session unused this long closes.
+ pub idle: Duration,
+}
+
+impl Default for ProduceSessionLimits {
+ fn default() -> Self {
+ Self {
+ max: 256,
+ idle: Duration::from_mins(10),
+ }
+ }
+}
+
/// 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
@@ -50,6 +70,7 @@ 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,
+ pub produce_sessions: ProduceSessionLimits,
}
impl IggyBridgeConfig {
@@ -68,6 +89,8 @@ impl IggyBridgeConfig {
"IGGY_KAFKA_IGGY_STREAM",
"IGGY_KAFKA_TOPIC_MAP_PATH",
"IGGY_KAFKA_IGGY_MAX_MESSAGE_SIZE",
+ "IGGY_KAFKA_PRODUCE_SESSIONS",
+ "IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS",
];
/// Builds config from `IGGY_KAFKA_*` env vars, defaulting to the Iggy
server's own
@@ -93,7 +116,8 @@ impl IggyBridgeConfig {
/// Returns [`BridgeError::InvalidConfig`] if `IGGY_KAFKA_IGGY_PASSWORD`
is unset or empty, if
/// `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`).
+ /// an empty `default_stream`), or if `IGGY_KAFKA_IGGY_MAX_MESSAGE_SIZE`,
+ /// `IGGY_KAFKA_PRODUCE_SESSIONS` or
`IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS` 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());
@@ -162,12 +186,32 @@ impl IggyBridgeConfig {
})?,
};
+ 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"
+ ))
+ })?;
+ }
+
Ok(Self {
address,
username,
password: SecretString::from(password),
topic_mapping,
max_message_size,
+ produce_sessions,
})
}
}
@@ -186,6 +230,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,
+ produce_sessions: ProduceSessionLimits::default(),
}
}
@@ -220,6 +265,8 @@ mod tests {
std::env::remove_var("IGGY_KAFKA_IGGY_STREAM");
std::env::remove_var("IGGY_KAFKA_TOPIC_MAP_PATH");
std::env::remove_var("IGGY_KAFKA_IGGY_MAX_MESSAGE_SIZE");
+ std::env::remove_var("IGGY_KAFKA_PRODUCE_SESSIONS");
+ std::env::remove_var("IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS");
}
let result = IggyBridgeConfig::from_env();
unsafe {
@@ -228,6 +275,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.produce_sessions, ProduceSessionLimits::default());
assert_eq!(config.address, "127.0.0.1:8090");
assert_eq!(config.username, "iggy");
assert_eq!(config.topic_mapping.default_stream(), "kafka");
@@ -262,6 +310,44 @@ mod tests {
assert!(matches!(garbage, Err(BridgeError::InvalidConfig(_))));
}
+ #[test]
+ #[serial]
+ fn from_env_reads_the_produce_session_limits() {
+ // 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_PRODUCE_SESSIONS", "0");
+ std::env::set_var("IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS", "30");
+ }
+ let set = IggyBridgeConfig::from_env();
+ unsafe {
+ std::env::set_var("IGGY_KAFKA_PRODUCE_SESSIONS", "many");
+ }
+ let bad_count = IggyBridgeConfig::from_env();
+ unsafe {
+ std::env::set_var("IGGY_KAFKA_PRODUCE_SESSIONS", "8");
+ std::env::set_var("IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS", "0");
+ }
+ let no_idle = IggyBridgeConfig::from_env();
+ unsafe {
+ std::env::remove_var("IGGY_KAFKA_IGGY_PASSWORD");
+ std::env::remove_var("IGGY_KAFKA_PRODUCE_SESSIONS");
+ std::env::remove_var("IGGY_KAFKA_PRODUCE_SESSION_IDLE_SECS");
+ }
+
+ assert_eq!(
+ set.unwrap().produce_sessions,
+ ProduceSessionLimits {
+ max: 0,
+ idle: Duration::from_secs(30),
+ },
+ "0 sessions is a choice: every producer on one client"
+ );
+ assert!(matches!(bad_count, Err(BridgeError::InvalidConfig(_))));
+ assert!(matches!(no_idle, Err(BridgeError::InvalidConfig(_))));
+ }
+
#[test]
#[serial]
fn from_env_rejects_missing_password() {
diff --git a/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
index 9ce900bbb..fd42f5030 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
@@ -31,7 +31,7 @@ use tokio::time::{Instant, timeout_at};
use super::offsets::high_watermark;
use super::{
- IggyBridge, REQUEST_TIMEOUT, TopicTarget, connect_client, in_slot,
with_request_timeout,
+ IggyBridge, REQUEST_TIMEOUT, TopicTarget, connect_client, in_slot,
sign_out_and_shut_down,
};
use crate::bridge::config::IggyBridgeConfig;
use crate::bridge::error::BridgeError;
@@ -77,7 +77,7 @@ impl FetchPool {
})
}
- /// Shuts down each client that connected. See [`LazyClient::close_all`].
+ /// Closes each client that connected. See [`LazyClient::close_all`].
pub(super) async fn close(&self) -> Result<(), BridgeError> {
LazyClient::close_all(&self.all).await
}
@@ -99,6 +99,12 @@ impl LazyClient {
Self(Arc::new(AsyncMutex::new(Some(client))))
}
+ /// Puts `client` in hand, so a test needs no server.
+ #[cfg(test)]
+ pub(super) async fn hold(&self, client: Arc<IggyClient>) {
+ *self.0.lock().await = Some(client);
+ }
+
/// The client, connected at the first call. Callers that come during a
connect wait for it.
pub(super) async fn connected(
&self,
@@ -114,6 +120,24 @@ impl LazyClient {
Ok(connected)
}
+ /// The client, if it connected.
+ pub(super) async fn held(&self) -> Option<Arc<IggyClient>> {
+ self.0.lock().await.clone()
+ }
+
+ /// Forgets `client` and closes its connection once the call running on it
ends. For a send
+ /// that was given up on: the SDK goes on writing it, and it must not land
after a later send
+ /// on a new connection.
+ ///
+ /// No timeout. The SDK ends a call by its 30 s read deadline, and a new
connection opened
+ /// before that could send ahead of it.
+ pub(super) async fn retire(&self, client: &IggyClient) {
+ self.0.lock().await.take();
+ if let Err(error) = client.shutdown().await {
+ tracing::debug!(%error, "failed to shut down a retired Iggy
client");
+ }
+ }
+
/// Runs `call` until `deadline`. A call that times out drops the client.
pub(super) async fn until<T>(
&self,
@@ -129,15 +153,15 @@ impl LazyClient {
result
}
- /// Shuts the client down, if it connected.
+ /// Signs the client out and shuts it down, if it connected.
pub(super) async fn close(&self) -> Result<(), BridgeError> {
let Some(client) = self.0.lock().await.take() else {
return Ok(());
};
- with_request_timeout(client.shutdown()).await
+ sign_out_and_shut_down(&client).await
}
- /// Shuts down each client that connected, and returns the first error. A
client that fails to
+ /// Closes each client that connected, and returns the first error. A
client that fails to
/// shut down stops none of the others.
pub(super) async fn close_all(
clients: impl IntoIterator<Item = &Self>,
diff --git a/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
index c455bce1e..3f71a0949 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
@@ -281,7 +281,7 @@ impl OffsetPool {
}
}
- /// Shuts down each client that connected. See [`LazyClient::close_all`].
+ /// Closes each client that connected. See [`LazyClient::close_all`].
pub(super) async fn close(&self) -> Result<(), BridgeError> {
let clients = self
.slots
@@ -402,18 +402,14 @@ async fn create_offset_group(
#[cfg(test)]
mod tests {
- use std::collections::HashMap;
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use iggy::prelude::{ConsumerKind, IggyClientBuilder};
- use secrecy::SecretString;
use tokio::sync::oneshot;
use super::*;
- use crate::bridge::iggy_bridge::FetchPool;
- use crate::bridge::{DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig,
TopicMapping};
fn target(topic: &str) -> TopicTarget {
TopicTarget {
@@ -450,17 +446,6 @@ mod tests {
read_clients: [Arc<IggyClient>; OFFSET_SLOTS],
) -> IggyBridge {
IggyBridge {
- client: offline_client(),
- config: IggyBridgeConfig {
- address: "127.0.0.1:1".to_string(),
- username: "iggy".to_string(),
- password: SecretString::from("iggy"),
- topic_mapping: TopicMapping::new("kafka".to_string(),
HashMap::new()).unwrap(),
- max_message_size: DEFAULT_MAX_MESSAGE_SIZE,
- },
- send_slot: Arc::new(Semaphore::new(1)),
- fetch_pool: Arc::new(FetchPool::new()),
- probe_client: LazyClient::default(),
offset_pool: OffsetPool {
slots: commit_clients
.into_iter()
@@ -471,6 +456,7 @@ mod tests {
})
.collect(),
},
+ ..IggyBridge::offline()
}
}
diff --git a/gateways/kafka/src/bridge/iggy_bridge/mod.rs
b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
index b0c8032ef..fc068e8ec 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
@@ -21,10 +21,11 @@ use std::time::Duration;
use iggy::prelude::{
AutoLogin, Client, Credentials, Identifier, IggyClient, IggyClientBuilder,
IggyError,
+ UserClient,
};
use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
use tokio::time::{Instant, timeout_at};
-use tracing::info;
+use tracing::{debug, info};
use crate::bridge::config::IggyBridgeConfig;
use crate::bridge::error::BridgeError;
@@ -40,6 +41,8 @@ use fetch::{FetchPool, LazyClient};
pub(crate) use fetch::{FetchSlot, PartitionProbe, TopicProbe};
use group_offsets::OffsetPool;
pub use group_offsets::{OFFSET_GROUP_PREFIX, OffsetCalls};
+use produce::ProducePool;
+pub use produce::{ProduceTurn, Producer};
pub use topics::{KafkaTopicMetadata, StreamTopicCache, TopicCreationOutcome,
TopicLoad};
/// Passes attempted, after the first, before [`IggyBridge::connect`] gives up
and returns `Err`.
@@ -80,7 +83,19 @@ async fn with_request_timeout<T>(
.map_err(BridgeError::Iggy)
}
-/// How long a send may hold its slot: the SDK read deadline (30 s) plus a
reconnect (15 s).
+/// Signs `client` out, then shuts it down, each within [`REQUEST_TIMEOUT`].
+///
+/// A shutdown alone leaves the session in Iggy's client table until its lease
runs out, 30 s by
+/// default.
+async fn sign_out_and_shut_down(client: &IggyClient) -> Result<(),
BridgeError> {
+ // Fails on a client that is not signed in. That one holds no session to
end.
+ if let Err(error) = with_request_timeout(client.logout_user()).await {
+ debug!(%error, "Iggy client did not sign out before its shutdown");
+ }
+ with_request_timeout(client.shutdown()).await
+}
+
+/// How long a call may hold its slot: the SDK read deadline (30 s) plus a
reconnect (15 s).
const SLOT_LIMIT: Duration = Duration::from_secs(45);
/// Runs `call` in a task that holds `slot` until `call` ends, then hands both
back. `None` once
@@ -89,7 +104,8 @@ const SLOT_LIMIT: Duration = Duration::from_secs(45);
/// A call given up on keeps its slot until it ends, so the next call waits
for the slot, not in
/// the SDK queue. A `call` that stops itself sooner frees the slot while the
SDK can still hold
/// its connection. A Fetch poll then drops that connection
(`LazyClient::until`). An offset
-/// commit keeps it, so the next commit waits behind it in the SDK and commits
stay in order.
+/// commit keeps it, so the next commit waits behind it in the SDK and commits
stay in order. A
+/// Produce send closes it before the slot frees (`LazyClient::retire`).
async fn in_slot<S: Send + 'static, T: Send + 'static>(
slot: S,
deadline: Instant,
@@ -114,17 +130,20 @@ async fn permit_by(turn: &Arc<Semaphore>, deadline:
Instant) -> Option<OwnedSema
/// Owns the connected `IggyClient`s and resolves Kafka topics against them.
///
-/// One lockstep client serves Produce, Metadata and `CreateTopics` for every
Kafka connection, so
-/// those Iggy calls run one at a time. Topic probes use a client of their
own, Fetch polls one per
-/// read slot, and offset calls two per offset slot. A Produce pool is a TODO
in `docs/SCOPE.md`.
+/// Each producer sends on a client of its own, one request at a time.
Producers past the session
+/// limit share one more. Metadata, `CreateTopics`, `DeleteTopics`,
`DescribeConfigs` and
+/// `AlterConfigs` share another. Topic probes use a client of their own,
Fetch polls one per read
+/// slot, and offset calls two per offset slot.
///
/// Each client signs in at the metadata leader, and a view change or a
refused call can move it to
/// another node. So in a cluster, two clients can read replicas that are not
at the same offset.
pub struct IggyBridge {
+ /// The client of Metadata, `CreateTopics`, `DeleteTopics`,
`DescribeConfigs` and
+ /// `AlterConfigs`.
client: Arc<IggyClient>,
config: IggyBridgeConfig,
- /// One Produce send inside the SDK at a time. See `send_records`.
- send_slot: Arc<Semaphore>,
+ /// One Iggy session per producer. See `produce_turn`.
+ produce_pool: Arc<ProducePool>,
/// Fetch read slots, each with its own client. See `fetch_slot`.
fetch_pool: Arc<FetchPool>,
/// The client of every topic probe, Fetch's and `ListOffsets`'. See
`probe`.
@@ -161,11 +180,13 @@ impl IggyBridge {
pub async fn connect(config: IggyBridgeConfig) -> Result<Self,
BridgeError> {
let client = connect_client(&config).await?;
info!("Iggy bridge connected to {}", config.address);
+ let produce_pool = Arc::new(ProducePool::new(config.produce_sessions));
+ produce_pool.sweep_in_background();
Ok(Self {
client: Arc::new(client),
config,
- send_slot: Arc::new(Semaphore::new(1)),
+ produce_pool,
fetch_pool: Arc::new(FetchPool::new()),
probe_client: LazyClient::default(),
offset_pool: OffsetPool::new(),
@@ -199,7 +220,7 @@ impl IggyBridge {
(stream.to_string(), topic.to_string())
}
- /// Tears down the underlying Iggy client, including its background
heartbeat task.
+ /// Signs out and tears down every Iggy client, including its background
heartbeat task.
///
/// Not `IggyClient::disconnect`: that only tears down the transport
/// (`TcpClient::disconnect_transport`) and never touches
`heartbeat_handle` - only
@@ -211,28 +232,67 @@ impl IggyBridge {
/// `IggyError::ClientShutdown` and self-terminates on, and which
`sign_in_credentials` never
/// dials past.
///
- /// Takes `self` by value: `shutdown` is terminal (no reconnect is coming
back from it), so
- /// nothing legitimate is left to call on this bridge afterward. A bridge
shared via `Arc`
- /// needs `Arc::try_unwrap` first.
+ /// Call it once nothing more is sent. `shutdown` is terminal for a client
(no reconnect is
+ /// coming back from it), but a later Produce, Fetch or offset call opens
a new one. By
+ /// reference, because the gateway shares the bridge through an `Arc` that
a connection task
+ /// can still hold at shutdown.
///
/// # Errors
///
- /// Returns [`BridgeError::Timeout`] if it takes longer than
`REQUEST_TIMEOUT`. Returns
+ /// Returns [`BridgeError::Timeout`] if a client takes longer than
`REQUEST_TIMEOUT`. Returns
/// [`BridgeError::Iggy`] if the underlying client reports a shutdown
failure (e.g. the socket
/// was already in a state that rejects a clean shutdown). Every client is
shut down even when
/// one fails, and the error is the first one.
- pub async fn close(self) -> Result<(), BridgeError> {
- let fetch_clients = self.fetch_pool.close().await;
- let probe_client = self.probe_client.close().await;
- let offset_clients = self.offset_pool.close().await;
- let shared_client = with_request_timeout(self.client.shutdown()).await;
- fetch_clients
+ pub async fn close(&self) -> Result<(), BridgeError> {
+ // At once, so a client that is slow to close does not use up the time
of the others.
+ let (produce_clients, fetch_clients, probe_client, offset_clients,
shared_client) = tokio::join!(
+ self.produce_pool.close(),
+ self.fetch_pool.close(),
+ self.probe_client.close(),
+ self.offset_pool.close(),
+ sign_out_and_shut_down(&self.client),
+ );
+ produce_clients
+ .and(fetch_clients)
.and(probe_client)
.and(offset_clients)
.and(shared_client)
}
}
+#[cfg(test)]
+impl IggyBridge {
+ /// A bridge that never dialed, so a test needs no server. Each Iggy call
on it fails at once.
+ pub(crate) fn offline() -> Self {
+ let address = "127.0.0.1:1".to_string();
+ let client = IggyClientBuilder::new()
+ .with_tcp()
+ .with_server_address(address.clone())
+ .build()
+ .expect("building a client dials nothing");
+ let config = IggyBridgeConfig {
+ address,
+ username: "iggy".to_string(),
+ password: secrecy::SecretString::from("iggy"),
+ topic_mapping: crate::bridge::TopicMapping::new(
+ "kafka".to_string(),
+ std::collections::HashMap::new(),
+ )
+ .expect("a valid mapping"),
+ max_message_size: crate::bridge::DEFAULT_MAX_MESSAGE_SIZE,
+ produce_sessions: crate::bridge::ProduceSessionLimits::default(),
+ };
+ Self {
+ client: Arc::new(client),
+ produce_pool: Arc::new(ProducePool::new(config.produce_sessions)),
+ config,
+ fetch_pool: Arc::new(FetchPool::new()),
+ probe_client: LazyClient::default(),
+ offset_pool: OffsetPool::new(),
+ }
+ }
+}
+
/// A client for `config`, connected and signed in. See
[`IggyBridge::connect`].
async fn connect_client(config: &IggyBridgeConfig) -> Result<IggyClient,
BridgeError> {
if config.address.trim().is_empty() {
@@ -243,9 +303,12 @@ async fn connect_client(config: &IggyBridgeConfig) ->
Result<IggyClient, BridgeE
let credentials =
Credentials::UsernamePassword(config.username.clone(),
config.password.clone());
+ // A send over the SDK's 8 KiB write buffer leaves as two writes, header
then payload. With
+ // Nagle on, the payload waits for the server's delayed ACK of the header:
about 40 ms a send.
let client = IggyClientBuilder::new()
.with_tcp()
.with_server_address(config.address.clone())
+ .with_no_delay()
.with_auto_sign_in(AutoLogin::Enabled(credentials))
.with_reconnection_max_retries(Some(RECONNECTION_RETRIES))
.build()
diff --git a/gateways/kafka/src/bridge/iggy_bridge/produce.rs
b/gateways/kafka/src/bridge/iggy_bridge/produce.rs
index b4ee4d09e..d95a511d0 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/produce.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/produce.rs
@@ -19,19 +19,326 @@
//!
//! 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::atomic::{AtomicBool, Ordering};
+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 futures::future::join_all;
+use iggy::prelude::{IggyClient, IggyError, IggyMessage, MessageClient,
Partitioning};
+use tokio::runtime::Handle;
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, MissedTickBehavior, interval, sleep_until, timeout,
timeout_at};
+use super::fetch::LazyClient;
use super::{IggyBridge, SLOT_LIMIT, TopicTarget, in_slot, permit_by};
+use crate::bridge::config::{IggyBridgeConfig, ProduceSessionLimits};
use crate::bridge::error::BridgeError;
/// Frame bytes besides the messages: request header, batch header and ids.
const SEND_FRAME_ROOM: u64 = 4 * 1024;
+/// Sessions that connect and sign in at once. A sign-in costs Iggy a password
hash on a shard
+/// thread, so a burst of new producers must not start them all together.
+const SESSION_OPENS: usize = 4;
+
+/// Each of those slots opens one session in this long, so at most 80 sign in
a second. Iggy keeps
+/// a session that ended without a sign-out for 30 s, and refuses new ones
once it holds 8192.
+const OPEN_SPACING: Duration = Duration::from_millis(50);
+
+/// How often the free sessions past their idle limit close.
+const SWEEP_EVERY: Duration = Duration::from_secs(30);
+
+/// A full pool closes its least recently used free session to make room, once
that session sat
+/// unused this long. A producer that sends more often keeps its session, so
more producers than
+/// sessions do not mean a sign-in for each request.
+const MAKE_ROOM_AFTER: Duration = Duration::from_secs(30);
+
+/// Who sends a Produce request. Each producer gets an Iggy session of its own.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub enum Producer {
+ /// A producer id this gateway gave out, and the user who sends with it
when SASL is on. One
+ /// user's id never reaches another user's session.
+ Id { id: i64, user: Option<String> },
+ /// The Kafka connection, for a producer without such an id.
+ Connection(u64),
+}
+
+/// One Iggy client. One request holds its turn at a time, so the sends on it
reach Iggy in order.
+struct ProduceSession {
+ client: LazyClient,
+ turn: Arc<Semaphore>,
+}
+
+impl ProduceSession {
+ fn new() -> Self {
+ Self {
+ client: LazyClient::default(),
+ turn: Arc::new(Semaphore::new(1)),
+ }
+ }
+}
+
+/// A producer's place in the pool: a session of its own, or the one that the
producers past the
+/// limit share.
+///
+/// Each request of the producer holds a clone, taken under the pool lock, and
a send that
+/// outlives a request keeps it. So a count of 1 means that nothing of the
producer runs.
+struct Seat {
+ session: Arc<ProduceSession>,
+ own: bool,
+}
+
+/// A producer's turn on its session. Dropping it frees the turn.
+pub struct ProduceTurn {
+ seat: Arc<Seat>,
+ _permit: OwnedSemaphorePermit,
+}
+
+/// The Produce sessions, one per producer, so different producers send at
once.
+pub(super) struct ProducePool {
+ seats: Mutex<Seats>,
+ /// The session of every producer that found the pool full.
+ overflow: Arc<ProduceSession>,
+ limits: ProduceSessionLimits,
+ /// [`SESSION_OPENS`] slots. See [`Self::opening`].
+ opens: Arc<Semaphore>,
+ /// Whether the log already says that the pool filled up.
+ full_logged: AtomicBool,
+}
+
+/// Every producer's seat, and how many of them have a session of their own.
+#[derive(Default)]
+struct Seats {
+ held: HashMap<Producer, Held>,
+ own: usize,
+}
+
+/// A seat, and when a request last took it.
+struct Held {
+ seat: Arc<Seat>,
+ used: Instant,
+}
+
+impl ProducePool {
+ pub(super) fn new(limits: ProduceSessionLimits) -> Self {
+ Self {
+ seats: Mutex::default(),
+ overflow: Arc::new(ProduceSession::new()),
+ limits,
+ opens: Arc::new(Semaphore::new(SESSION_OPENS)),
+ full_logged: AtomicBool::new(false),
+ }
+ }
+
+ /// Closes idle sessions every [`SWEEP_EVERY`], until the pool is dropped.
+ pub(super) fn sweep_in_background(self: &Arc<Self>) {
+ let pool = Arc::downgrade(self);
+ tokio::spawn(async move {
+ let mut tick = interval(SWEEP_EVERY);
+ tick.set_missed_tick_behavior(MissedTickBehavior::Delay);
+ loop {
+ tick.tick().await;
+ let Some(pool) = pool.upgrade() else {
+ return;
+ };
+ pool.sweep();
+ }
+ });
+ }
+
+ /// The seat of `producer`.
+ ///
+ /// A producer without a session of its own gets one while the pool has
room. A full pool
+ /// first closes its least recently used free session, if that one sat
unused for
+ /// [`MAKE_ROOM_AFTER`]. Failing that, the producer shares the overflow
session, and asks again
+ /// at each request that finds none of its earlier ones running.
+ fn take(&self, producer: Producer) -> Arc<Seat> {
+ let now = Instant::now();
+ let mut guard =
self.seats.lock().unwrap_or_else(PoisonError::into_inner);
+ let seats = &mut *guard;
+ if let Some(held) = seats.held.get_mut(&producer) {
+ held.used = now;
+ // A request of the producer still runs on the shared session. A
send on another
+ // session could land ahead of it.
+ if held.seat.own || !held.is_free() {
+ return Arc::clone(&held.seat);
+ }
+ }
+ let mut closed = seats.take_idle(now, self.limits.idle);
+ if seats.own >= self.limits.max {
+ let unused = self.limits.idle.min(MAKE_ROOM_AFTER);
+ closed.extend(seats.take_least_used(now, unused));
+ }
+ let seat = if seats.own < self.limits.max {
+ seats.own += 1;
+ Seat {
+ session: Arc::new(ProduceSession::new()),
+ own: true,
+ }
+ } else {
+ // Once, not per request. 0 sessions is the operator's choice, so
it says nothing.
+ if self.limits.max != 0 && !self.full_logged.swap(true,
Ordering::Relaxed) {
+ tracing::warn!(
+ sessions = self.limits.max,
+ "every Produce session is taken, so more producers share
one Iggy client: \
+ raise IGGY_KAFKA_PRODUCE_SESSIONS to give each its own"
+ );
+ }
+ Seat {
+ session: Arc::clone(&self.overflow),
+ own: false,
+ }
+ };
+ let seat = Arc::new(seat);
+ let held = Held {
+ seat: Arc::clone(&seat),
+ used: now,
+ };
+ seats.held.insert(producer, held);
+ drop(guard);
+ closed.into_iter().for_each(retire);
+ seat
+ }
+
+ /// The client of `session`, connected and signed in at its first send.
+ async fn client_of(
+ &self,
+ session: &Arc<ProduceSession>,
+ config: &IggyBridgeConfig,
+ ) -> Result<Arc<IggyClient>, BridgeError> {
+ if let Some(client) = session.client.held().await {
+ return Ok(client);
+ }
+ let (session, config) = (Arc::clone(session), config.clone());
+ let open = async move { session.client.connected(&config).await };
+ // `None` if the open's task died. Whether it signed in is then
unknown.
+ self.opening(open)
+ .await
+ .unwrap_or(Err(BridgeError::Timeout))
+ }
+
+ /// Runs `open` in one of the [`SESSION_OPENS`] slots, in a task of its
own, and waits for it.
+ ///
+ /// A caller that gives up while it waits for a slot started nothing. One
that gives up later
+ /// leaves the open to end, for the producer's next request: a sign-in cut
short leaves Iggy a
+ /// session that nobody signs out, at a rate no slot counts.
+ ///
+ /// The slot stays taken until [`OPEN_SPACING`] after the open started,
which caps the
+ /// sign-ins a second.
+ async fn opening<T: Send + 'static>(
+ &self,
+ open: impl Future<Output = T> + Send + 'static,
+ ) -> Option<T> {
+ // Never closed, so this waits and does not fail.
+ let slot = Arc::clone(&self.opens).acquire_owned().await;
+ let (sender, receiver) = oneshot::channel();
+ tokio::spawn(async move {
+ let free_at = Instant::now() + OPEN_SPACING;
+ // The caller may have stopped waiting.
+ let _ = sender.send(open.await);
+ sleep_until(free_at).await;
+ drop(slot);
+ });
+ receiver.await.ok()
+ }
+
+ /// Closes the free sessions unused for the idle limit.
+ fn sweep(&self) {
+ let closed = self
+ .seats
+ .lock()
+ .unwrap_or_else(PoisonError::into_inner)
+ .take_idle(Instant::now(), self.limits.idle);
+ closed.into_iter().for_each(retire);
+ }
+
+ /// Forgets the seat of `producer`, and closes its session once no send
runs on it.
+ fn forget(&self, producer: &Producer) {
+ let mut seats =
self.seats.lock().unwrap_or_else(PoisonError::into_inner);
+ let Some(held) = seats.held.remove(producer) else {
+ return;
+ };
+ if held.seat.own {
+ seats.own -= 1;
+ drop(seats);
+ retire(Arc::clone(&held.seat.session));
+ }
+ }
+
+ /// Signs out and shuts down every session's client that connected, all at
once. Returns the
+ /// first error. A client that fails to shut down stops none of the others.
+ pub(super) async fn close(&self) -> Result<(), BridgeError> {
+ let seats = std::mem::take(&mut
*self.seats.lock().unwrap_or_else(PoisonError::into_inner));
+ let sessions: Vec<&Arc<ProduceSession>> = seats
+ .held
+ .values()
+ .filter(|held| held.seat.own)
+ .map(|held| &held.seat.session)
+ .chain([&self.overflow])
+ .collect();
+ join_all(sessions.iter().map(|session| session.client.close()))
+ .await
+ .into_iter()
+ .fold(Ok(()), Result::and)
+ }
+}
+
+impl Seats {
+ /// Takes out the free seats on the shared session, and the free ones
unused for `idle`.
+ /// Returns the sessions to close.
+ fn take_idle(&mut self, now: Instant, idle: Duration) ->
Vec<Arc<ProduceSession>> {
+ // A free seat on the shared session holds nothing: its producer gets
a new one.
+ let closed: Vec<Arc<ProduceSession>> = self
+ .held
+ .extract_if(|_, held| {
+ held.is_free() && (!held.seat.own ||
now.duration_since(held.used) >= idle)
+ })
+ .filter(|(_, held)| held.seat.own)
+ .map(|(_, held)| Arc::clone(&held.seat.session))
+ .collect();
+ self.own -= closed.len();
+ closed
+ }
+
+ /// Takes out the least recently used free session, if it sat unused for
`unused`.
+ fn take_least_used(&mut self, now: Instant, unused: Duration) ->
Option<Arc<ProduceSession>> {
+ let producer = self
+ .held
+ .iter()
+ .filter(|(_, held)| held.seat.own && held.is_free())
+ .filter(|(_, held)| now.duration_since(held.used) >= unused)
+ .min_by_key(|(_, held)| held.used)
+ .map(|(producer, _)| producer.clone())?;
+ let held = self.held.remove(&producer)?;
+ self.own -= 1;
+ Some(Arc::clone(&held.seat.session))
+ }
+}
+
+impl Held {
+ /// None of the producer's requests runs, and no send that outlived one.
+ fn is_free(&self) -> bool {
+ Arc::strong_count(&self.seat) == 1
+ }
+}
+
+/// Closes a session the pool forgot, in a task of its own, once no send runs
on it.
+fn retire(session: Arc<ProduceSession>) {
+ // No runtime means the process is going down, and its sockets with it.
+ let Ok(runtime) = Handle::try_current() else {
+ return;
+ };
+ runtime.spawn(async move {
+ // A send that outlived its request still holds the turn. It ends
first.
+ let _turn = session.turn.acquire().await;
+ if let Err(error) = session.client.close().await {
+ tracing::debug!(%error, "failed to close a Produce session");
+ }
+ });
+}
+
impl IggyBridge {
/// Largest partition, in message bytes, that one Iggy send carries.
#[must_use]
@@ -39,7 +346,27 @@ impl IggyBridge {
self.config.max_message_size.saturating_sub(SEND_FRAME_ROOM)
}
- /// Appends `messages` to `partition` of `target` and returns the base
offset.
+ /// The turn of `producer` on its session. `None` if it does not free by
`deadline`: an
+ /// earlier request of the producer, or a send that outlived one, still
holds it.
+ ///
+ /// Take it before anything the request must not hold while it waits here.
+ pub async fn produce_turn(&self, producer: Producer, deadline: Instant) ->
Option<ProduceTurn> {
+ let seat = self.produce_pool.take(producer);
+ let permit = permit_by(&seat.session.turn, deadline).await?;
+ Some(ProduceTurn {
+ seat,
+ _permit: permit,
+ })
+ }
+
+ /// Closes the Produce session that `connection` keyed, once no send runs
on it. Call when the
+ /// Kafka connection ends.
+ pub fn end_connection(&self, connection: u64) {
+ self.produce_pool.forget(&Producer::Connection(connection));
+ }
+
+ /// Appends `messages` to `partition` of `target` on the session of
`turn`, and returns the
+ /// base offset. Hands `turn` and `kept` back with it.
///
/// `partition` passes through unchanged: both systems count from 0. The
server rejects a
/// partition the topic lacks, so no round trip checks first.
@@ -47,60 +374,90 @@ impl IggyBridge {
/// `Ok(None)`: committed, but the server named no offset (a duplicate
request, or a journal
/// entry that is gone). Do not retry it.
///
- /// A stuck send holds the slot up to `SLOT_LIMIT`. Other sends answer 7
meanwhile.
+ /// A send still running at `deadline` keeps `turn` and `kept`, and hands
back `None` for
+ /// them. So the producer's later requests answer 7 meanwhile and cannot
overtake it, and what
+ /// `kept` reserves for `messages` stays reserved. Other producers send on.
+ ///
+ /// Such a send drops `messages` and `kept` when it ends, at `SLOT_LIMIT`
at the latest. Cut
+ /// off there, it keeps `turn` until its connection is closed.
///
/// # Errors
///
/// [`BridgeError::Timeout`] if `deadline` passes. The send may have
landed.
/// [`BridgeError::SendLost`] if the connection broke during the send. It
may have landed.
- /// [`BridgeError::Iggy`] if the stream, topic or partition is missing, or
the send is too
- /// large.
- pub async fn send_records(
+ /// [`BridgeError::Iggy`] if the session cannot connect, if the stream,
topic or partition is
+ /// missing, or if the send is too large.
+ pub async fn send_records<K: Send + 'static>(
&self,
+ turn: ProduceTurn,
+ kept: K,
target: &TopicTarget,
partition: u32,
mut messages: Vec<IggyMessage>,
deadline: Instant,
- ) -> Result<Option<u64>, BridgeError> {
- let client = Arc::clone(&self.client);
+ ) -> (Result<Option<u64>, BridgeError>, Option<(ProduceTurn, K)>) {
+ let session = Arc::clone(&turn.seat.session);
+ let connect = self.produce_pool.client_of(&session, &self.config);
+ let client = match timeout_at(deadline, connect).await {
+ Ok(Ok(client)) => client,
+ Ok(Err(error)) => return (Err(error), Some((turn, kept))),
+ Err(_elapsed) => return (Err(BridgeError::Timeout), Some((turn,
kept))),
+ };
+ // Decoding can run past the deadline. A send must not start late.
+ if Instant::now() >= deadline {
+ return (Err(BridgeError::Timeout), Some((turn, kept)));
+ }
let (stream_id, topic_id) = (target.stream_id.clone(),
target.topic_id.clone());
let send = async move {
- let partitioning = Partitioning::partition_id(partition);
- client
- .send_messages(&stream_id, &topic_id, &partitioning, &mut
messages)
- .await
+ // Owns the messages, so a send cut off at the limit drops them.
+ let send = {
+ let client = Arc::clone(&client);
+ async move {
+ let partitioning = Partitioning::partition_id(partition);
+ client
+ .send_messages(&stream_id, &topic_id, &partitioning,
&mut messages)
+ .await
+ }
+ };
+ within_slot_limit(&session, &client, kept, send).await
};
- let response = send_in_slot(&self.send_slot, deadline, send).await?;
- Ok(response
- .confirmations
- .iter()
- .find(|confirmation| confirmation.partition_id == partition)
- .map(|confirmation| confirmation.base_offset))
+ // The SDK cannot cancel a send. One that outlives this call keeps the
turn in its task,
+ // so the next one waits for it here instead of queueing its batch
inside the SDK.
+ let Some(((response, kept), turn)) = in_slot(turn, deadline,
send).await else {
+ return (Err(BridgeError::Timeout), None);
+ };
+ let base_offset = response.map(|response| {
+ response
+ .confirmations
+ .iter()
+ .find(|confirmation| confirmation.partition_id == partition)
+ .map(|confirmation| confirmation.base_offset)
+ });
+ (base_offset, kept.map(|kept| (turn, kept)))
}
}
-/// Runs `send` in a task that holds `slot` until the send ends, and waits for
it until `deadline`.
+/// Runs `send` for at most [`SLOT_LIMIT`], and hands `kept` back when it ends
by then.
///
-/// The SDK cannot cancel a send. A send that outlives its caller keeps the
slot, so the next one
-/// waits here instead of queueing its batch inside the SDK.
-async fn send_in_slot<T: Send + 'static>(
- slot: &Arc<Semaphore>,
- deadline: Instant,
- send: impl Future<Output = Result<T, IggyError>> + Send + 'static,
-) -> Result<T, BridgeError> {
- let Some(permit) = permit_by(slot, deadline).await else {
- return Err(BridgeError::Timeout);
- };
- let send = async move {
- match timeout(SLOT_LIMIT, send).await {
- Ok(result) => result.map_err(send_error),
- Err(_elapsed) => Err(BridgeError::Timeout),
+/// A send cut off there drops `kept` with the messages it gave up. The SDK
goes on writing that
+/// send, so its connection closes before this returns, and so before the turn
frees: a later
+/// send, on a new connection, cannot land ahead of it.
+async fn within_slot_limit<T, K>(
+ session: &ProduceSession,
+ client: &IggyClient,
+ kept: K,
+ send: impl Future<Output = Result<T, IggyError>>,
+) -> (Result<T, BridgeError>, Option<K>) {
+ match timeout(SLOT_LIMIT, send).await {
+ Ok(result) => (result.map_err(send_error), Some(kept)),
+ Err(_elapsed) => {
+ // Not across the close: that waits for the SDK's write, which a
server that stopped
+ // reading can hold up without end.
+ drop(kept);
+ session.client.retire(client).await;
+ (Err(BridgeError::Timeout), None)
}
- };
- // The permit comes back only to be dropped. A send that outlives the
caller frees it in its task.
- in_slot(permit, deadline, send)
- .await
- .map_or(Err(BridgeError::Timeout), |(result, _permit)| result)
+ }
}
/// The SDK does not replay a send after these, so it may have landed.
@@ -116,79 +473,377 @@ const fn send_error(error: IggyError) -> BridgeError {
#[cfg(test)]
mod tests {
- use std::sync::atomic::{AtomicBool, Ordering};
- use std::time::Duration;
+ use std::sync::Weak;
- use tokio::sync::oneshot;
+ use iggy::prelude::{IggyClientBuilder, SystemClient};
use super::*;
const WAIT: Duration = Duration::from_millis(20);
+ const IDLE: Duration = Duration::from_secs(60);
+
+ fn pool_of(max: usize) -> ProducePool {
+ ProducePool::new(ProduceSessionLimits { max, idle: IDLE })
+ }
+
+ const fn id(id: i64) -> Producer {
+ Producer::Id { id, user: None }
+ }
+
+ /// A client that never dialed and holds no credentials, so each call
fails at once.
+ fn unconnected() -> Arc<IggyClient> {
+ let client = IggyClientBuilder::new()
+ .with_tcp()
+ .with_server_address("127.0.0.1:1".to_string())
+ .build()
+ .expect("building a client dials nothing");
+ Arc::new(client)
+ }
+
+ /// Gives `session` a client, and returns it to check on later.
+ async fn with_client(session: &ProduceSession) -> Arc<IggyClient> {
+ let client = unconnected();
+ session.client.hold(Arc::clone(&client)).await;
+ client
+ }
- /// A send that records that it ran.
- async fn flagged(started: Arc<AtomicBool>) -> Result<u64, IggyError> {
- started.store(true, Ordering::SeqCst);
- tokio::task::yield_now().await;
- Ok(2)
+ /// A call on a client that never dialed fails as not connected, until its
shutdown.
+ async fn is_shut_down(client: &IggyClient) -> bool {
+ matches!(client.ping().await, Err(IggyError::ClientShutdown))
+ }
+
+ /// Waits for the task that closes a session the pool forgot. Time is
paused in these tests.
+ async fn closed(session: &Weak<ProduceSession>) {
+ let gone = async {
+ while session.upgrade().is_some() {
+ tokio::time::sleep(Duration::from_millis(1)).await;
+ }
+ };
+ timeout(Duration::from_secs(1), gone)
+ .await
+ .expect("a forgotten session closes");
+ }
+
+ fn shares(pool: &ProducePool, seat: &Seat) -> bool {
+ Arc::ptr_eq(&seat.session, &pool.overflow)
}
#[tokio::test]
async fn
given_a_send_past_its_deadline_when_the_next_one_waits_should_not_start_it() {
- let slot = Arc::new(Semaphore::new(1));
+ let turn = Arc::new(Semaphore::new(1));
let (release, released) = oneshot::channel::<()>();
let stuck = async move {
let _ = released.await;
- Ok::<u64, IggyError>(1)
};
+ let sent = in_slot(
+ Arc::clone(&turn).acquire_owned().await.unwrap(),
+ Instant::now() + WAIT,
+ stuck,
+ )
+ .await;
+ assert!(sent.is_none(), "the send keeps the turn");
+ assert!(
+ permit_by(&turn, Instant::now() + WAIT).await.is_none(),
+ "so the next one never reaches the SDK"
+ );
+
+ release.send(()).unwrap();
+ assert!(
+ permit_by(&turn, Instant::now() + WAIT * 10).await.is_some(),
+ "the turn frees when the stuck send ends"
+ );
+ }
+
+ #[tokio::test]
+ async fn given_a_passed_deadline_when_sending_should_not_start_the_send() {
+ let bridge = IggyBridge::offline();
+ let target = bridge.topic_target("orders").expect("a valid topic");
+ with_client(&bridge.produce_pool.take(id(1)).session).await;
+
+ let (sent, held) = bridge
+ .send_records(
+ bridge
+ .produce_turn(id(1), Instant::now() + WAIT)
+ .await
+ .expect("a free turn"),
+ 7,
+ &target,
+ 0,
+ Vec::new(),
+ Instant::now(),
+ )
+ .await;
+ assert!(
+ matches!(sent, Err(BridgeError::Timeout)),
+ "a send that started would fail as not connected"
+ );
+ assert!(
+ matches!(held, Some((_, 7))),
+ "nothing runs, so nothing is kept"
+ );
+ }
+
+ #[tokio::test]
+ async fn
given_a_lost_connection_when_sending_should_say_the_send_may_have_landed() {
+ let session = ProduceSession::new();
+ let lost = async { Err::<u64, IggyError>(IggyError::Disconnected) };
+
assert!(matches!(
- send_in_slot(&slot, Instant::now() + WAIT, stuck).await,
- Err(BridgeError::Timeout)
+ within_slot_limit(&session, &unconnected(), 7, lost).await,
+ (Err(BridgeError::SendLost(IggyError::Disconnected)), Some(7))
));
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn
given_a_send_past_the_slot_limit_when_cut_off_should_close_its_connection_first()
{
+ let session = ProduceSession::new();
+ let client = with_client(&session).await;
+ let stuck = std::future::pending::<Result<u64, IggyError>>();
- let started = Arc::new(AtomicBool::new(false));
assert!(
matches!(
- send_in_slot(&slot, Instant::now() + WAIT,
flagged(Arc::clone(&started))).await,
- Err(BridgeError::Timeout)
+ within_slot_limit(&session, &client, 7, stuck).await,
+ (Err(BridgeError::Timeout), None)
),
- "the stuck send still holds the slot"
+ "what it kept for its messages goes with them"
+ );
+ assert!(
+ is_shut_down(&client).await,
+ "the SDK's write cannot land after a later send"
);
assert!(
- !started.load(Ordering::SeqCst),
- "a send that never got the slot never reaches the SDK"
+ session.client.held().await.is_none(),
+ "the next send connects a new client"
+ );
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn
given_many_new_sessions_when_they_open_should_cap_how_many_sign_in_a_second() {
+ let pool = pool_of(4);
+ let started = Instant::now();
+
+ for _ in 0..SESSION_OPENS {
+ pool.opening(async {}).await;
+ }
+ assert_eq!(started.elapsed(), Duration::ZERO, "a slot for each");
+ pool.opening(async {}).await;
+ assert_eq!(
+ started.elapsed(),
+ OPEN_SPACING,
+ "the next one waits for a slot, even after every open ended"
);
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn
given_an_open_its_caller_gave_up_on_when_it_can_go_on_should_run_to_its_end() {
+ let pool = pool_of(4);
+ let (release, released) = oneshot::channel::<()>();
+ let ended = Arc::new(AtomicBool::new(false));
+ let open = {
+ let ended = Arc::clone(&ended);
+ async move {
+ let _ = released.await;
+ ended.store(true, Ordering::SeqCst);
+ }
+ };
+ assert!(timeout(WAIT, pool.opening(open)).await.is_err());
release.send(()).unwrap();
- let after = send_in_slot(
- &slot,
- Instant::now() + WAIT * 10,
- flagged(Arc::clone(&started)),
- )
- .await;
- assert_eq!(after.unwrap(), 2, "the slot frees when the stuck send
ends");
+ tokio::time::sleep(WAIT).await;
+ assert!(
+ ended.load(Ordering::SeqCst),
+ "cut short, a sign-in leaves Iggy a session that nobody signs out"
+ );
}
#[tokio::test]
- async fn given_a_passed_deadline_when_sending_should_not_start_the_send() {
- let slot = Arc::new(Semaphore::new(1));
- let started = Arc::new(AtomicBool::new(false));
+ async fn given_one_producer_when_taken_twice_should_get_one_session() {
+ let pool = pool_of(4);
+ let first = pool.take(id(7));
- assert!(matches!(
- send_in_slot(&slot, Instant::now(),
flagged(Arc::clone(&started))).await,
- Err(BridgeError::Timeout)
- ));
- tokio::task::yield_now().await;
- assert!(!started.load(Ordering::SeqCst));
+ assert!(Arc::ptr_eq(&first.session, &pool.take(id(7)).session));
+ assert!(
+ !Arc::ptr_eq(&first.session,
&pool.take(Producer::Connection(7)).session),
+ "an id and a connection number never share a session"
+ );
+ let other_user = Producer::Id {
+ id: 7,
+ user: Some("eve".to_string()),
+ };
+ assert!(
+ !Arc::ptr_eq(&first.session, &pool.take(other_user).session),
+ "one user's id never reaches the session of another"
+ );
}
#[tokio::test]
- async fn
given_a_lost_connection_when_sending_should_say_the_send_may_have_landed() {
- let slot = Arc::new(Semaphore::new(1));
- let lost = async { Err::<u64, IggyError>(IggyError::Disconnected) };
+ async fn
given_a_stuck_send_when_another_producer_sends_should_not_wait_for_it() {
+ let pool = pool_of(4);
+ let (stuck, other) = (pool.take(id(1)), pool.take(id(2)));
+ let _sending = Arc::clone(&stuck.session.turn)
+ .acquire_owned()
+ .await
+ .unwrap();
- assert!(matches!(
- send_in_slot(&slot, Instant::now() + WAIT * 10, lost).await,
- Err(BridgeError::SendLost(IggyError::Disconnected))
- ));
+ assert!(
+ permit_by(&stuck.session.turn, Instant::now() + WAIT)
+ .await
+ .is_none()
+ );
+ assert!(
+ permit_by(&other.session.turn, Instant::now() + WAIT)
+ .await
+ .is_some()
+ );
+ }
+
+ #[tokio::test]
+ async fn
given_a_full_pool_when_more_producers_come_should_share_the_overflow_session() {
+ let pool = pool_of(1);
+ let own = Arc::downgrade(&pool.take(id(1)).session);
+
+ assert!(shares(&pool, &pool.take(id(2))));
+ assert!(shares(&pool, &pool.take(id(3))));
+ assert!(
+ own.upgrade()
+ .is_some_and(|own| Arc::ptr_eq(&own,
&pool.take(id(1)).session)),
+ "a session in use stays, so no producer signs in again for each
request"
+ );
+ }
+
+ #[tokio::test]
+ async fn
given_a_limit_of_no_sessions_when_producers_come_should_share_one_client() {
+ let pool = pool_of(0);
+
+ assert!(shares(&pool, &pool.take(id(1))));
+ assert!(shares(&pool, &pool.take(Producer::Connection(2))));
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn
given_a_full_pool_when_a_new_producer_comes_should_close_only_a_stale_session()
{
+ let pool = pool_of(2);
+ let stale = Arc::downgrade(&pool.take(id(1)).session);
+ tokio::time::advance(MAKE_ROOM_AFTER / 2).await;
+ let recent = Arc::downgrade(&pool.take(id(2)).session);
+ assert!(
+ shares(&pool, &pool.take(id(3))),
+ "both sessions were used a moment ago"
+ );
+
+ tokio::time::advance(MAKE_ROOM_AFTER / 2).await;
+ assert!(
+ !shares(&pool, &pool.take(id(3))),
+ "the least recently used one made room"
+ );
+ closed(&stale).await;
+ assert!(recent.upgrade().is_some());
+ }
+
+ #[tokio::test]
+ async fn
given_a_producer_on_the_overflow_session_when_room_comes_should_move_once_none_of_its_requests_runs()
+ {
+ let pool = pool_of(1);
+ drop(pool.take(Producer::Connection(1)));
+ let running = pool.take(id(2));
+ assert!(shares(&pool, &running));
+ pool.forget(&Producer::Connection(1));
+
+ assert!(
+ shares(&pool, &pool.take(id(2))),
+ "its earlier request still runs there, and must not be overtaken"
+ );
+ drop(running);
+ assert!(!shares(&pool, &pool.take(id(2))));
+ }
+
+ #[tokio::test]
+ async fn
given_room_in_the_pool_when_a_new_producer_comes_should_not_join_a_busy_overflow_session()
+ {
+ let pool = pool_of(1);
+ drop(pool.take(Producer::Connection(1)));
+ let _running = pool.take(id(2));
+ pool.forget(&Producer::Connection(1));
+
+ assert!(!shares(&pool, &pool.take(id(3))));
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn
given_a_session_idle_past_the_limit_when_swept_should_close_its_client() {
+ let pool = pool_of(4);
+ let seat = pool.take(Producer::Connection(1));
+ let client = with_client(&seat.session).await;
+ let idle = Arc::downgrade(&seat.session);
+ drop(seat);
+
+ tokio::time::advance(IDLE / 2).await;
+ pool.sweep();
+ assert!(idle.upgrade().is_some(), "not idle long enough yet");
+
+ tokio::time::advance(IDLE / 2).await;
+ pool.sweep();
+ closed(&idle).await;
+ assert!(is_shut_down(&client).await);
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn
given_a_session_idle_past_the_limit_when_a_new_producer_comes_should_close_it()
{
+ let pool = pool_of(4);
+ let idle = Arc::downgrade(&pool.take(id(1)).session);
+ tokio::time::advance(IDLE).await;
+
+ drop(pool.take(id(2)));
+ closed(&idle).await;
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn given_a_send_in_flight_when_its_session_is_idle_should_keep_it() {
+ let pool = pool_of(1);
+ let seat = pool.take(id(1));
+ tokio::time::advance(IDLE).await;
+
+ pool.sweep();
+ assert!(
+ shares(&pool, &pool.take(id(2))),
+ "a full pool does not close it to make room either"
+ );
+ assert!(
+ Arc::ptr_eq(&seat.session, &pool.take(id(1)).session),
+ "the producer's next send queues behind this one, not on a new
session"
+ );
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn
given_a_connection_that_ends_when_forgotten_should_close_its_session_after_its_send()
{
+ let pool = pool_of(4);
+ let seat = pool.take(Producer::Connection(1));
+ let client = with_client(&seat.session).await;
+ let sending = Arc::clone(&seat.session.turn)
+ .acquire_owned()
+ .await
+ .unwrap();
+ let ended = Arc::downgrade(&seat.session);
+ drop(seat);
+
+ pool.forget(&Producer::Connection(1));
+ tokio::time::sleep(WAIT).await;
+ assert!(ended.upgrade().is_some(), "its send still runs");
+ assert!(!is_shut_down(&client).await);
+
+ drop(sending);
+ closed(&ended).await;
+ assert!(is_shut_down(&client).await);
+ }
+
+ #[tokio::test]
+ async fn
given_open_sessions_when_the_pool_closes_should_shut_every_client_down() {
+ let pool = pool_of(1);
+ let own = with_client(&pool.take(id(1)).session).await;
+ let shared = with_client(&pool.take(id(2)).session).await;
+
+ pool.close()
+ .await
+ .expect("clients that never dialed close cleanly");
+ assert!(is_shut_down(&own).await);
+ assert!(is_shut_down(&shared).await, "the overflow session too");
}
}
diff --git a/gateways/kafka/src/bridge/mod.rs b/gateways/kafka/src/bridge/mod.rs
index 6083448f4..8b25a8ae7 100644
--- a/gateways/kafka/src/bridge/mod.rs
+++ b/gateways/kafka/src/bridge/mod.rs
@@ -26,10 +26,10 @@ pub mod error;
pub mod iggy_bridge;
pub mod topic_map;
-pub use config::{DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig};
+pub use config::{DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig,
ProduceSessionLimits};
pub use error::BridgeError;
pub use iggy_bridge::{
- IggyBridge, KafkaTopicMetadata, OFFSET_GROUP_PREFIX, OffsetCalls,
StreamTopicCache,
- TopicCreationOutcome, TopicLoad, TopicTarget,
+ IggyBridge, KafkaTopicMetadata, OFFSET_GROUP_PREFIX, OffsetCalls,
ProduceTurn, Producer,
+ StreamTopicCache, TopicCreationOutcome, TopicLoad, TopicTarget,
};
pub use topic_map::{TopicMapping, TopicOverride};
diff --git a/gateways/kafka/src/main.rs b/gateways/kafka/src/main.rs
index 11233607e..5eaf8fcf5 100644
--- a/gateways/kafka/src/main.rs
+++ b/gateways/kafka/src/main.rs
@@ -43,7 +43,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let listener = bind_listener(&config.bind_addr)
.map_err(|e| format!("failed to bind {}: {e}", config.bind_addr))?;
- let mut server = KafkaGateway::new(config).with_bridge(bridge);
+ let mut server = KafkaGateway::new(config).with_bridge(bridge.clone());
if sasl_enabled {
let authenticator = IggyAuthenticator::from_env()?;
// Unconditional, because the Kafka listener has no TLS at any
setting. This is the hop the
@@ -67,17 +67,38 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let (tx, rx) = broadcast::channel(1);
let mut server_task = tokio::spawn(async move { server.run(listener,
rx).await });
- tokio::select! {
- result = &mut server_task => {
- return Ok(result??);
- }
- () = shutdown_signal() => {
- let _ = tx.send(());
- }
- }
+ let stopped_alone = tokio::select! {
+ result = &mut server_task => Some(result),
+ () = shutdown_signal() => None,
+ };
+ let outcome = if let Some(outcome) = stopped_alone {
+ outcome
+ } else {
+ let _ = tx.send(());
+ server_task.await
+ };
+ close_bridge(bridge.as_deref()).await;
+ Ok(outcome??)
+}
- server_task.await??;
- Ok(())
+/// How long shutdown waits for the Iggy clients to sign out. The connection
drain before it can
+/// take 25 s of a 30 s grace period.
+const BRIDGE_CLOSE_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// Signs the bridge's Iggy clients out and shuts them down. Without it Iggy
keeps each client's
+/// session until its lease runs out.
+async fn close_bridge(bridge: Option<&IggyBridge>) {
+ let Some(bridge) = bridge else {
+ return;
+ };
+ match tokio::time::timeout(BRIDGE_CLOSE_TIMEOUT, bridge.close()).await {
+ Ok(Ok(())) => info!("Iggy bridge closed"),
+ Ok(Err(error)) => warn!("the Iggy bridge did not close cleanly:
{error}"),
+ Err(_elapsed) => warn!(
+ "the Iggy bridge did not close within {BRIDGE_CLOSE_TIMEOUT:?}:
Iggy ends the \
+ remaining sessions when their leases run out"
+ ),
+ }
}
/// The `IGGY_KAFKA_*` vars `load_config` itself reads. `IGGY_KAFKA_` is a
diff --git a/gateways/kafka/src/protocol/api.rs
b/gateways/kafka/src/protocol/api.rs
index 9539773b8..e6038a06d 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, PoisonError};
use std::time::Duration;
@@ -318,9 +319,9 @@ pub fn supported_api_ranges() -> &'static [ApiVersionRange]
{
/// `bridge` is `None` until `IGGY_KAFKA_BRIDGE_ENABLED` turns it on. A
handler that finds `None`
/// answers with its stub, so APIs can be wired one at a time.
///
-/// Every Iggy call but a Fetch poll, a topic probe and an offset call goes
through one lockstep
-/// `IggyClient`, so Kafka connections serialize behind whichever of those
calls is in flight. The
-/// `Arc` does not change that. See the README's "Concurrency ceiling".
+/// Produce sends on one Iggy client per producer. Metadata, `CreateTopics`,
`DeleteTopics`,
+/// `DescribeConfigs` and `AlterConfigs` share one lockstep `IggyClient`, so
those calls run one
+/// at a time across connections. See the README's "Concurrency ceiling".
pub struct GatewayState {
pub broker: BrokerAdvertise,
pub bridge: Option<Arc<IggyBridge>>,
@@ -331,8 +332,12 @@ pub struct GatewayState {
/// Shared across every connection this gateway serves: a producer id has
to be unique for
/// the process, not for the connection that asked for it.
pub producer_ids: ProducerIdAllocator,
- /// Produce requests that decode and send at once. Caps their memory.
- pub(crate) produce_slots: Semaphore,
+ /// `PRODUCE_CAPS` partition caps, in frame bytes, that Produce may hold
decoded at once. Caps
+ /// its memory. A request takes what its largest partition may decode, and
a send that
+ /// outlives the request keeps it while it owns the decoded partition.
+ pub(crate) produce_memory: Arc<Semaphore>,
+ /// `PRODUCE_LARGE` slots a large Produce request takes before its memory.
Small ones skip it.
+ pub(crate) produce_large: Arc<Semaphore>,
/// Consumer group membership. Process-wide and independent of the bridge:
a member outlives
/// the connection that created it, and group coordination needs no Iggy
call.
pub groups: GroupCoordinator,
@@ -343,9 +348,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;
+/// 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;
+
+/// Partition caps Produce may hold decoded at once, about 200 MB at the
default 8 MiB frame.
+///
+/// Counted in bytes, not requests, so a small request takes a small share.
One cap more than the
+/// large slots: large requests can hold a cap each, and the small ones still
have one.
+const PRODUCE_CAPS: usize = PRODUCE_LARGE + 1;
impl GatewayState {
#[must_use]
@@ -363,7 +374,12 @@ impl GatewayState {
max_frame_size,
sasl_enabled,
producer_ids: ProducerIdAllocator::new(instance_id),
- produce_slots: Semaphore::const_new(PRODUCE_SLOTS),
+ produce_memory: Arc::new(Semaphore::new(
+ PRODUCE_CAPS
+ .saturating_mul(max_frame_size)
+ .min(Semaphore::MAX_PERMITS),
+ )),
+ produce_large: Arc::new(Semaphore::new(PRODUCE_LARGE)),
groups,
loading: Mutex::default(),
probe_board: ProbeBoard::default(),
@@ -400,9 +416,27 @@ pub struct ConnectionState {
/// The member that last sent a Heartbeat here. An offset call caps its
hold by that member's
/// session, since its next heartbeat waits behind the call.
heartbeat_member: Mutex<Option<GroupMember>>,
+ /// Keys the Produce session of a producer without a producer id that this
gateway gave out.
+ id: ConnectionId,
+}
+
+/// A Kafka connection's number, unique in the process.
+#[derive(Debug, Clone, Copy)]
+struct ConnectionId(u64);
+
+static NEXT_CONNECTION_ID: AtomicU64 = AtomicU64::new(0);
+
+impl Default for ConnectionId {
+ fn default() -> Self {
+ Self(NEXT_CONNECTION_ID.fetch_add(1, Ordering::Relaxed))
+ }
}
impl ConnectionState {
+ pub(crate) const fn id(&self) -> u64 {
+ self.id.0
+ }
+
pub(crate) fn note_heartbeat(&self, member: GroupMember) {
*self
.heartbeat_member
diff --git a/gateways/kafka/src/protocol/handlers/init_producer_id.rs
b/gateways/kafka/src/protocol/handlers/init_producer_id.rs
index c9718f197..4f62d754a 100644
--- a/gateways/kafka/src/protocol/handlers/init_producer_id.rs
+++ b/gateways/kafka/src/protocol/handlers/init_producer_id.rs
@@ -19,8 +19,9 @@
//!
//! A Java producer sets `enable.idempotence=true` without being asked
(KIP-679, default since
//! Kafka 3.0) and sends this before its first record, so answering it is what
lets a stock
-//! producer start against this gateway at all. The id is handed out and then
ignored: delivery
-//! stays at-least-once, and no retry is deduplicated. See
`docs/IDEMPOTENCE.md`.
+//! producer start against this gateway at all. Produce uses the id only to
pick the producer's
+//! Iggy session: delivery stays at-least-once, and no retry is deduplicated.
See
+//! `docs/IDEMPOTENCE.md`.
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};
@@ -64,26 +65,42 @@ const PRODUCER_EPOCH: i16 = 0;
/// would hand a restarted gateway's producers the pairs its previous run gave
out. Seeding from
/// the clock keeps every new id above the old ones unless the previous run
averaged more than one
/// allocation per millisecond of its uptime, the clock stepped back across
the restart, or the
-/// clock reads before the Unix epoch and seeds 0. Nothing is persisted, and
nothing reads the id
-/// yet. Before Produce (#3535) keys anything on it, the allocator has to
persist a high-water mark
-/// or bump the epoch on every start.
+/// clock reads before the Unix epoch and seeds 0. Nothing is persisted.
+///
+/// Produce keys an Iggy session on the id, and only on an id of this run
(`issued`). A
+/// replayed pair then makes two producers share a session, and nothing more.
Before deduplication
+/// reads the id, the allocator has to persist a high-water mark or bump the
epoch on every start.
#[derive(Debug)]
pub struct ProducerIdAllocator {
instance_id: u16,
+ /// The first counter of this run.
+ first_counter: u64,
next_counter: AtomicU64,
}
impl ProducerIdAllocator {
#[must_use]
pub fn new(instance_id: u16) -> Self {
+ let first_counter = clock_counter();
Self {
instance_id,
- next_counter: AtomicU64::new(clock_counter()),
+ first_counter,
+ next_counter: AtomicU64::new(first_counter),
}
}
+ /// Whether this run gave out `producer_id`. An id a client made up, or
one from before a
+ /// restart or from another gateway, is not.
+ pub(crate) fn issued(&self, producer_id: i64) -> bool {
+ let Ok(id) = u64::try_from(producer_id) else {
+ return false;
+ };
+ let given_out =
self.first_counter..self.next_counter.load(Ordering::Relaxed);
+ id >> COUNTER_BITS == u64::from(self.instance_id) &&
given_out.contains(&(id & MAX_COUNTER))
+ }
+
/// The next id for this instance, or `None` once its counter space is
spent.
- fn allocate(&self) -> Option<i64> {
+ pub(crate) fn allocate(&self) -> Option<i64> {
let counter = self.next_counter.fetch_add(1, Ordering::Relaxed);
if counter > MAX_COUNTER {
return None;
@@ -196,6 +213,21 @@ mod tests {
);
}
+ #[test]
+ fn
given_an_id_when_checked_should_be_issued_only_if_this_run_gave_it_out() {
+ let allocator = ProducerIdAllocator::new(3);
+ let id = allocator.allocate().expect("id");
+
+ assert!(allocator.issued(id));
+ assert!(!allocator.issued(id + 1), "not given out yet");
+ assert!(!allocator.issued(id - 1), "from before this run");
+ assert!(!allocator.issued(-1), "no producer id");
+ assert!(
+ !ProducerIdAllocator::new(4).issued(id),
+ "another gateway's id"
+ );
+ }
+
#[test]
fn
given_a_spent_counter_when_allocating_should_refuse_instead_of_bleeding_into_the_instance()
{
let allocator = ProducerIdAllocator::new(1);
diff --git a/gateways/kafka/src/protocol/handlers/mod.rs
b/gateways/kafka/src/protocol/handlers/mod.rs
index 2e2282229..941fe8e74 100644
--- a/gateways/kafka/src/protocol/handlers/mod.rs
+++ b/gateways/kafka/src/protocol/handlers/mod.rs
@@ -111,7 +111,7 @@ pub async fn dispatch(
body: Bytes,
) -> HandleOutcome {
match api_key {
- API_KEY_PRODUCE => produce::handle(state, api_version, body).await,
+ API_KEY_PRODUCE => produce::handle(state, connection, principal,
api_version, body).await,
API_KEY_FETCH => fetch::handle(state, connection, api_version,
body).await,
API_KEY_LIST_OFFSETS => list_offsets::handle(state, api_version,
body).await,
API_KEY_METADATA => metadata::handle(state, api_version, body).await,
diff --git a/gateways/kafka/src/protocol/handlers/produce.rs
b/gateways/kafka/src/protocol/handlers/produce.rs
index a18b9c388..bb2e24c5d 100644
--- a/gateways/kafka/src/protocol/handlers/produce.rs
+++ b/gateways/kafka/src/protocol/handlers/produce.rs
@@ -17,6 +17,7 @@
//! Produce (API key 0).
+use std::sync::Arc;
use std::time::Duration;
use bytes::Bytes;
@@ -25,13 +26,15 @@ use
kafka_protocol::messages::produce_request::PartitionProduceData;
use kafka_protocol::messages::produce_response::{PartitionProduceResponse,
TopicProduceResponse};
use kafka_protocol::messages::{ProduceRequest, ProduceResponse};
use kafka_protocol::protocol::StrBytes;
+use tokio::sync::{AcquireError, OwnedSemaphorePermit};
-use crate::bridge::{BridgeError, IggyBridge, TopicTarget};
+use crate::auth::AuthenticatedPrincipal;
+use crate::bridge::{BridgeError, IggyBridge, ProduceTurn, Producer,
TopicTarget};
use crate::error::Result;
use crate::protocol::api::{
- API_KEY_PRODUCE, ApiVersionRange, ERROR_INVALID_RECORD,
ERROR_INVALID_REQUIRED_ACKS,
- ERROR_MESSAGE_TOO_LARGE, ERROR_NONE, ERROR_NOT_LEADER_OR_FOLLOWER,
ERROR_REQUEST_TIMED_OUT,
- ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ API_KEY_PRODUCE, ApiVersionRange, ConnectionState, ERROR_INVALID_RECORD,
+ ERROR_INVALID_REQUIRED_ACKS, ERROR_MESSAGE_TOO_LARGE, ERROR_NONE,
ERROR_NOT_LEADER_OR_FOLLOWER,
+ ERROR_REQUEST_TIMED_OUT, ERROR_UNKNOWN_SERVER_ERROR,
ERROR_UNKNOWN_TOPIC_OR_PARTITION,
ERROR_UNSUPPORTED_COMPRESSION_TYPE, ERROR_UNSUPPORTED_VERSION,
GatewayState, HandleOutcome,
REQUEST_DEADLINE, UNKNOWN_OFFSET, supported_max_version,
};
@@ -43,7 +46,7 @@ use crate::protocol::handlers::{
use crate::protocol::probe_board::ProbeBoard;
use crate::records::{
Allowance, DecompressionBudget, RecordCodecError, TimestampWindow, Zstd,
decode_batch,
- is_compressed, to_iggy,
+ is_compressed, producer_id, record_slots, to_iggy,
};
pub const RANGE: ApiVersionRange = ApiVersionRange {
@@ -63,7 +66,7 @@ const ZSTD_MIN_VERSION: i16 = 7;
/// Frame bytes per record slot. Caps one partition at 131,072 slots, about 40
MB, at the default
/// 8 MiB frame.
///
-/// `GatewayState` caps how many requests hold that at once.
+/// `GatewayState` caps how much of that requests hold at once, at the same
rate.
const FRAME_BYTES_PER_RECORD: usize = 64;
/// Partition allowances one request may decode, refused partitions included.
@@ -82,7 +85,13 @@ const RESPONSE_BASE_BYTES: usize = 512;
/// spec-compliant client can legitimately send Produce v0-2 with `acks=0`.
Rejecting those
/// versions before reading `acks` would send an error response the client
never expects,
/// desyncing the next correlation id it reads.
-pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) ->
HandleOutcome {
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
// Above the encoder max there is no response parseable at the client's
version, so close
// rather than decode (same policy the other APIs apply). Fail-closed on a
missing row
// (i16::MIN, not i16::MAX): dispatch routes Produce on its api key, not
from this table, so
@@ -156,17 +165,17 @@ 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",
- )
- };
+ // The turn before the memory: a request that waits behind its producer's
stuck send then
+ // holds none, and no large slot.
+ let producer = producer_of(state, connection, principal, &request);
+ let Some(turn) = bridge.produce_turn(producer, deadline.at).await else {
+ tracing::debug!("Produce deadline passed while waiting for the
producer's earlier send");
+ return timed_out(api_version, &request);
+ };
+ 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 partition = Allowance {
bytes: state.max_frame_size,
@@ -178,8 +187,8 @@ pub async fn handle(state: &GatewayState, api_version: i16,
body: Bytes) -> Hand
} else {
Zstd::Refused
};
- let responses =
- write_request(bridge, &state.probe_board, &request, budget, zstd,
deadline).await;
+ let (board, held) = (&state.probe_board, (turn, admission));
+ let responses = write_request(bridge, held, board, &request, budget, zstd,
deadline).await;
// Kafka closes an `acks=0` connection on any error, so the client
refreshes metadata.
if request.acks == ACKS_NONE {
@@ -206,6 +215,101 @@ 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 its record slots, whichever is more. Compressed, it may fill the
cap.
+fn decode_cost(request: &ProduceRequest, max_frame_size: usize) -> usize {
+ let blobs = || {
+ request
+ .topic_data
+ .iter()
+ .flat_map(|topic| &topic.partition_data)
+ .filter_map(|partition| partition.records.as_deref())
+ };
+ let cap_slots = max_frame_size / FRAME_BYTES_PER_RECORD;
+ // Only an uncompressed partition under the cap needs its slots counted.
+ let counted = |blob: &[u8]| !is_compressed(blob) && blob.len() <
max_frame_size;
+ let cost = |blob: &[u8]| {
+ if counted(blob) {
+ let slots = record_slots(blob, cap_slots);
+ blob.len().max(slots.saturating_mul(FRAME_BYTES_PER_RECORD))
+ } else {
+ max_frame_size
+ }
+ };
+ // Counting reads every record, so a large partition counts off the
worker, as it decodes.
+ let heavy = blobs().any(|blob| counted(blob) && blob.len() >=
CODEC_OFF_WORKER_BYTES);
+ off_worker(heavy, ||
blobs().map(cost).max().unwrap_or(0)).min(max_frame_size)
+}
+
+/// What a request holds from before its first decode: a large slot when it is
large, and its
+/// decode memory.
+///
+/// A send that outlives the request keeps it for as long as it owns the
decoded partition: until
+/// it ends, or until the bridge gives up on it.
+struct Admission {
+ _large: Option<OwnedSemaphorePermit>,
+ _memory: OwnedSemaphorePermit,
+}
+
+/// The producer's turn and the request's admission. They travel together
through every send.
+type Held = (ProduceTurn, Admission);
+
+/// Waits for the request's decode memory. A large request,
`CODEC_OFF_WORKER_BYTES` or more,
+/// first waits for a large slot, and holds no memory while it does.
+async fn admit(state: &GatewayState, cost: usize) ->
std::result::Result<Admission, AcquireError> {
+ let large = if cost >= CODEC_OFF_WORKER_BYTES {
+ Some(Arc::clone(&state.produce_large).acquire_owned().await?)
+ } else {
+ None
+ };
+ let permits = u32::try_from(cost).unwrap_or(u32::MAX);
+ let memory = Arc::clone(&state.produce_memory)
+ .acquire_many_owned(permits)
+ .await?;
+ Ok(Admission {
+ _large: large,
+ _memory: memory,
+ })
+}
+
+/// The producer id of the request's first batch, when this gateway gave that
id out. Otherwise
+/// the connection: an id a client made up must not pick a session, or open
one per id.
+fn producer_of(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
+ request: &ProduceRequest,
+) -> Producer {
+ request
+ .topic_data
+ .iter()
+ .flat_map(|topic| &topic.partition_data)
+ .find_map(|partition| partition.records.as_deref())
+ .and_then(producer_id)
+ .filter(|id| state.producer_ids.issued(*id))
+ .map_or_else(
+ || Producer::Connection(connection.id()),
+ |id| Producer::Id {
+ id,
+ user: principal.map(|principal| principal.username.clone()),
+ },
+ )
+}
+
/// When a request stops writing.
#[derive(Clone, Copy)]
struct Deadline {
@@ -258,10 +362,11 @@ impl From<RecordCodecError> for Refusal {
/// Only one partition's messages live at a time. The budget is owned here and
borrowed only in
/// sync calls: a borrow held across an await makes the connection task
`!Send`.
///
-/// Sequential: the one `IggyClient` behind the bridge is lockstep, so
concurrent sends would
-/// just queue on its stream mutex (see the README's "Concurrency ceiling").
+/// Sequential, and `held` carries the producer's turn, so its sends reach
Iggy in order. A send
+/// that outlives the deadline keeps `held`, and the partitions after it
answer 7.
async fn write_request(
bridge: &IggyBridge,
+ held: Held,
board: &ProbeBoard,
request: &ProduceRequest,
budget: DecompressionBudget,
@@ -269,6 +374,7 @@ async fn write_request(
deadline: Deadline,
) -> Vec<TopicProduceResponse> {
let max_send = bridge.max_send_bytes();
+ let mut held = Some(held);
let mut topics = Vec::with_capacity(request.topic_data.len());
for topic in &request.topic_data {
let name = topic.name.as_str();
@@ -276,20 +382,23 @@ async fn write_request(
let mut responses = Vec::with_capacity(topic.partition_data.len());
let mut wrote = false;
for partition in &topic.partition_data {
- let outcome = match &target {
- // Skip the decode. The answer is 7 either way.
- Ok(_) if deadline.passed() =>
Err(ERROR_REQUEST_TIMED_OUT.into()),
- Ok(target) => {
- let planned = plan(&budget, zstd, name, partition,
max_send);
- match planned {
- Ok((id, messages)) => send_by(bridge, target, name,
id, messages, deadline)
- .await
- .map_err(Refusal::from),
- Err(refusal) => Err(refusal),
+ let (outcome, kept) = match (&target, held.take()) {
+ (Ok(target), Some(kept)) if !deadline.passed() => {
+ match plan(&budget, zstd, name, partition, max_send) {
+ Ok((id, messages)) => {
+ let (sent, kept) =
+ send_by(bridge, kept, target, name, id,
messages, deadline).await;
+ (sent.map_err(Refusal::from), kept)
+ }
+ Err(refusal) => (Err(refusal), Some(kept)),
}
}
- Err(error) => Err(refusal_code(name, error).into()),
+ // Skip the decode. The answer is 7 either way: the deadline
passed, or a send
+ // that outlived it kept the memory this decode would need.
+ (Ok(_), kept) => (Err(ERROR_REQUEST_TIMED_OUT.into()), kept),
+ (Err(error), kept) => (Err(refusal_code(name, error).into()),
kept),
};
+ held = kept;
wrote |= outcome.is_ok();
responses.push(partition_outcome(partition.index, outcome));
// Planning is sync CPU work. Let other connections on this worker
run.
@@ -408,21 +517,23 @@ fn request_timeout(timeout_ms: i32) -> Duration {
.min(REQUEST_DEADLINE)
}
-/// Sends one partition's messages, or answers 7 once the deadline has passed.
+/// Sends one partition's messages, or answers 7 once the deadline has passed.
Hands `held` back,
+/// unless the send still runs and keeps it.
///
/// 7 means nothing was written, or the send may still land and a retry
duplicate it.
async fn send_by(
bridge: &IggyBridge,
+ (turn, admission): Held,
target: &TopicTarget,
kafka_topic: &str,
partition_id: u32,
messages: Vec<IggyMessage>,
deadline: Deadline,
-) -> std::result::Result<Option<u64>, i16> {
- match bridge
- .send_records(target, partition_id, messages, deadline.at)
- .await
- {
+) -> (std::result::Result<Option<u64>, i16>, Option<Held>) {
+ let (sent, held) = bridge
+ .send_records(turn, admission, target, partition_id, messages,
deadline.at)
+ .await;
+ let sent = match sent {
Ok(base_offset) => Ok(base_offset),
Err(BridgeError::Timeout) => {
// The client picks `timeout_ms`, so its expiry is no signal to
the operator.
@@ -442,7 +553,8 @@ async fn send_by(
Err(ERROR_REQUEST_TIMED_OUT)
}
Err(error) => Err(refusal_code(kafka_topic, &error)),
- }
+ };
+ (sent, held)
}
/// The Kafka code for a bridge refusal, logged at a level that matches who
can fix it.
@@ -571,11 +683,18 @@ mod tests {
use super::*;
use bytes::BytesMut;
use kafka_protocol::indexmap::IndexMap;
+ use kafka_protocol::messages::TopicName;
+ use kafka_protocol::messages::produce_request::TopicProduceData;
+ use kafka_protocol::protocol::{Decodable, Encodable};
use kafka_protocol::records::{
Compression, NO_PARTITION_LEADER_EPOCH, NO_PRODUCER_EPOCH,
NO_PRODUCER_ID, NO_SEQUENCE,
Record, RecordBatchEncoder, RecordEncodeOptions, TimestampType,
};
+ use tokio_util::sync::CancellationToken;
+ use crate::group::{GroupCoordinator, GroupCoordinatorConfig};
+ use crate::protocol::acl::PrincipalPermissions;
+ use crate::protocol::api::BrokerAdvertise;
use crate::records::encode_batch;
const CREATE_TIME: i64 = 1_700_000_000_123;
@@ -694,6 +813,289 @@ mod tests {
);
}
+ fn request_of(partitions: Vec<PartitionProduceData>) -> ProduceRequest {
+ ProduceRequest::default().with_topic_data(vec![
+ TopicProduceData::default()
+ .with_name(TopicName(StrBytes::from_static_str(TOPIC)))
+ .with_partition_data(partitions),
+ ])
+ }
+
+ /// The default frame.
+ const FRAME: usize = 8 * 1024 * 1024;
+ /// Long enough for an admission that is free to complete. Time is paused
in these tests.
+ const WAIT: Duration = Duration::from_millis(50);
+
+ /// A one-record batch that carries `producer_id`.
+ fn batch_of(producer_id: i64) -> Bytes {
+ let mut records = [record(0, b"v")];
+ records[0].producer_id = producer_id;
+ records[0].producer_epoch = 0;
+ records[0].sequence = 0;
+ compressed_batch(&records, Compression::None)
+ }
+
+ fn principal(username: &str) -> AuthenticatedPrincipal {
+ AuthenticatedPrincipal {
+ username: username.to_string(),
+ permissions: PrincipalPermissions::default(),
+ permissions_known: true,
+ }
+ }
+
+ #[test]
+ fn
given_a_producer_id_this_gateway_gave_out_when_keyed_should_take_it_with_its_user()
{
+ let state = GatewayState::stub(BrokerAdvertise::default(), FRAME);
+ let id = state.producer_ids.allocate().expect("an id");
+ let request = request_of(vec![entry(0, Some(batch_of(id)))]);
+ let connection = ConnectionState::default();
+
+ assert_eq!(
+ producer_of(&state, &connection, None, &request),
+ Producer::Id { id, user: None }
+ );
+ assert_eq!(
+ producer_of(&state, &connection, Some(&principal("ann")),
&request),
+ Producer::Id {
+ id,
+ user: Some("ann".to_string())
+ },
+ "another user's request with this id gets another session"
+ );
+ }
+
+ #[test]
+ fn
given_a_producer_id_a_client_made_up_when_keyed_should_take_the_connection() {
+ let state = GatewayState::stub(BrokerAdvertise::default(), FRAME);
+ let connection = ConnectionState::default();
+ let issued = state.producer_ids.allocate().expect("an id");
+
+ for made_up in [7, issued + 1] {
+ let request = request_of(vec![entry(0, Some(batch_of(made_up)))]);
+ assert_eq!(
+ producer_of(&state, &connection, None, &request),
+ Producer::Connection(connection.id()),
+ "{made_up} picks no session, and opens none of its own"
+ );
+ }
+ }
+
+ #[test]
+ fn
given_a_batch_without_a_producer_id_when_keyed_should_take_the_connection() {
+ let state = GatewayState::stub(BrokerAdvertise::default(), FRAME);
+ let batch = compressed_batch(&[record(0, b"v")], Compression::None);
+ let request = request_of(vec![entry(0, Some(batch))]);
+ let connection = ConnectionState::default();
+
+ assert_eq!(
+ producer_of(&state, &connection, None, &request),
+ Producer::Connection(connection.id())
+ );
+ assert_ne!(
+ ConnectionState::default().id(),
+ connection.id(),
+ "each connection keys a session of its own"
+ );
+ }
+
+ fn cost_of(partitions: Vec<PartitionProduceData>, max_frame_size: usize)
-> usize {
+ decode_cost(&request_of(partitions), max_frame_size)
+ }
+
+ #[test]
+ fn given_an_uncompressed_batch_when_costed_should_take_its_bytes() {
+ let batch = encode_batch(&mut [record(0, &[7; 1000]), record(1, &[7;
1000])]).unwrap();
+ assert_eq!(
+ cost_of(vec![entry(0, Some(batch.clone()))], FRAME),
+ batch.len()
+ );
+ }
+
+ #[test]
+ fn given_small_records_when_costed_should_take_a_record_slot_each() {
+ let mut records: Vec<Record> = (0..100).map(|offset| record(offset,
b"v")).collect();
+ let batch = encode_batch(&mut records).unwrap();
+ assert!(batch.len() < 100 * FRAME_BYTES_PER_RECORD);
+ assert_eq!(
+ cost_of(vec![entry(0, Some(batch))], FRAME),
+ 100 * FRAME_BYTES_PER_RECORD
+ );
+ }
+
+ #[test]
+ fn
given_records_with_many_headers_when_costed_should_take_the_slots_the_decode_charges()
{
+ let mut heavy = record(0, b"v");
+ heavy.headers = (0..300)
+ .map(|index| (StrBytes::from_string(format!("h{index}")), None))
+ .collect();
+ let batch = encode_batch(&mut [heavy]).unwrap();
+ let cost = cost_of(vec![entry(0, Some(batch.clone()))], FRAME);
+
+ // One slot for the record and one for each three headers.
+ assert_eq!(cost, 101 * FRAME_BYTES_PER_RECORD);
+ assert!(
+ cost > batch.len(),
+ "its bytes alone would admit it for less"
+ );
+ let slots = cost / FRAME_BYTES_PER_RECORD;
+ let entry = entry(0, Some(batch));
+ assert!(
+ plan_with(&partition_budget(FRAME, slots), Zstd::Allowed,
&entry).is_ok(),
+ "the decode takes no more than the request was admitted for"
+ );
+ assert_eq!(
+ plan_with(&partition_budget(FRAME, slots - 1), Zstd::Allowed,
&entry).unwrap_err(),
+ ERROR_MESSAGE_TOO_LARGE
+ );
+ }
+
+ #[test]
+ fn given_a_compressed_batch_when_costed_should_take_a_whole_cap() {
+ let batch = gzip_batch(&[record(0, b"v")]);
+ assert_eq!(
+ cost_of(vec![entry(0, Some(batch))], FRAME),
+ FRAME,
+ "it may inflate to the cap"
+ );
+ }
+
+ #[test]
+ fn given_several_partitions_when_costed_should_take_the_largest() {
+ let small = compressed_batch(&[record(0, &[7; 100])],
Compression::None);
+ let large = compressed_batch(&[record(0, &[7; 5000])],
Compression::None);
+ let partitions = vec![
+ entry(0, Some(small)),
+ entry(1, Some(large.clone())),
+ entry(2, None),
+ ];
+ assert_eq!(
+ cost_of(partitions, FRAME),
+ large.len(),
+ "partitions decode one at a time"
+ );
+ }
+
+ #[test]
+ fn given_a_partition_past_the_cap_when_costed_should_take_one_cap() {
+ let batch = compressed_batch(&[record(0, &[7; 5000])],
Compression::None);
+ assert_eq!(cost_of(vec![entry(0, Some(batch))], 4096), 4096);
+ }
+
+ /// Admits four requests of `cost`, which takes every large slot.
+ async fn large_slots_taken(state: &GatewayState, cost: usize) ->
Vec<Admission> {
+ let mut admitted = Vec::new();
+ for _ in 0..4 {
+ admitted.push(admit(state, cost).await.unwrap());
+ }
+ admitted
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn
given_the_large_slots_taken_when_a_small_request_comes_should_admit_it() {
+ let state = GatewayState::stub(BrokerAdvertise::default(), FRAME);
+ let _large = large_slots_taken(&state, FRAME).await;
+
+ let fifth = admit(&state, CODEC_OFF_WORKER_BYTES);
+ assert!(
+ tokio::time::timeout(WAIT, fifth).await.is_err(),
+ "a fifth large request waits its turn"
+ );
+ let small = admit(&state, CODEC_OFF_WORKER_BYTES - 1);
+ assert!(
+ tokio::time::timeout(WAIT, small).await.is_ok(),
+ "four requests of a whole cap each leave a small one its memory"
+ );
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn
given_the_large_slots_taken_when_a_large_request_waits_should_hold_no_memory() {
+ let state = GatewayState::stub(BrokerAdvertise::default(), FRAME);
+ let _large = large_slots_taken(&state, CODEC_OFF_WORKER_BYTES).await;
+ let free = state.produce_memory.available_permits();
+
+ let waiting = admit(&state, FRAME);
+ tokio::pin!(waiting);
+ assert!(tokio::time::timeout(WAIT, &mut waiting).await.is_err());
+ assert_eq!(
+ state.produce_memory.available_permits(),
+ free,
+ "small requests can still take that memory"
+ );
+ }
+
+ /// A gateway whose bridge never dialed. A request that stops before its
first send needs no
+ /// server.
+ fn offline_state() -> GatewayState {
+ GatewayState::new(
+ BrokerAdvertise::default(),
+ Some(Arc::new(IggyBridge::offline())),
+ FRAME,
+ false,
+ 0,
+ GroupCoordinator::new(GroupCoordinatorConfig::default(),
CancellationToken::new()),
+ )
+ }
+
+ /// One large Produce v9 request of `connection` for partition 0, through
the whole handler.
+ async fn handled(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ acks: i16,
+ ) -> HandleOutcome {
+ let value = vec![7; CODEC_OFF_WORKER_BYTES];
+ let batch = compressed_batch(&[record(0, &value)], Compression::None);
+ let request = request_of(vec![entry(0, Some(batch))])
+ .with_acks(acks)
+ .with_timeout_ms(50);
+ let mut body = BytesMut::new();
+ request.encode(&mut body, 9).expect("encode the request");
+ handle(state, connection, None, 9, body.freeze()).await
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn
given_the_producers_earlier_send_still_running_when_handled_should_answer_timed_out()
{
+ let state = offline_state();
+ let bridge = state.bridge.as_deref().expect("a bridge");
+ let connection = ConnectionState::default();
+ let far = tokio::time::Instant::now() + Duration::from_secs(60);
+ let _earlier_send = bridge
+ .produce_turn(Producer::Connection(connection.id()), far)
+ .await
+ .expect("a free turn");
+ let free = |state: &GatewayState| {
+ (
+ state.produce_large.available_permits(),
+ state.produce_memory.available_permits(),
+ )
+ };
+ let before = free(&state);
+
+ let waiting = handled(&state, &connection, ACKS_ALL);
+ tokio::pin!(waiting);
+ assert!(
+ tokio::time::timeout(Duration::from_millis(25), &mut waiting)
+ .await
+ .is_err(),
+ "it waits for the turn until its own deadline"
+ );
+ assert_eq!(
+ free(&state),
+ before,
+ "and holds no large slot and no memory while it waits"
+ );
+ let mut answer = waiting.await.expect_response("acks=-1 is answered");
+ let response = ProduceResponse::decode(&mut answer, 9).expect("decode
the response");
+ let code = response.responses[0].partition_responses[0].error_code;
+ assert_eq!(
+ code, ERROR_REQUEST_TIMED_OUT,
+ "it must not overtake that send"
+ );
+ assert!(
+ handled(&state, &connection, ACKS_NONE).await.is_close(),
+ "acks=0 has no response to carry the 7"
+ );
+ }
+
#[test]
fn given_an_idempotent_batch_when_planned_should_convert_it() {
let mut records = [record(0, b"v")];
diff --git a/gateways/kafka/src/records.rs b/gateways/kafka/src/records.rs
index 42c8fbd51..586ded2cc 100644
--- a/gateways/kafka/src/records.rs
+++ b/gateways/kafka/src/records.rs
@@ -867,6 +867,38 @@ pub fn is_compressed(blob: &[u8]) -> bool {
.is_some_and(|attributes| attributes & 0b111 != 0)
}
+/// The producer id of the first batch in `blob`, bytes 43 to 50. `None` when
it has none.
+#[must_use]
+pub fn producer_id(blob: &[u8]) -> Option<i64> {
+ let id = i64::from_be_bytes(blob.get(43..51)?.try_into().ok()?);
+ (id != NO_PRODUCER_ID).then_some(id)
+}
+
+/// The record count the first batch in `blob` declares, bytes 57 to 60.
`None` when it has none.
+#[must_use]
+pub fn declared_records(blob: &[u8]) -> Option<usize> {
+ let count =
i32::from_be_bytes(blob.get(57..BATCH_HEADER_BYTES)?.try_into().ok()?);
+ usize::try_from(count).ok()
+}
+
+/// Record slots the first batch in `blob` takes to decode, when it is not
compressed: one per
+/// declared record, plus its headers as `scan_records` charges them. Stops at
`most`.
+///
+/// Counts the records it can walk. The decode refuses a batch that does not
walk, so that batch
+/// holds nothing.
+#[must_use]
+pub fn record_slots(blob: &[u8], most: usize) -> usize {
+ let mut slots = declared_records(blob).unwrap_or(0);
+ let mut records = blob.get(BATCH_HEADER_BYTES..).unwrap_or_default();
+ while slots < most && !records.is_empty() {
+ let Ok((headers, _)) = take_record(&mut records) else {
+ break;
+ };
+ slots = slots.saturating_add(headers / HEADERS_PER_SLOT);
+ }
+ slots.min(most)
+}
+
/// Decodes the one record batch a Produce partition carries.
///
/// # Errors
@@ -1075,30 +1107,13 @@ const HEADERS_PER_SLOT: usize = 3;
/// (`kafka-protocol-0.18.0/src/records.rs:896`), and that memory is
resident.
/// - Charges headers to the record budget, [`HEADERS_PER_SLOT`] per slot.
/// - Returns the record count, which `decode_batch` checks against the batch
header.
-///
-/// The framing mirrors `Record::decode_new`, field for field.
fn scan_records(records: &Bytes, budget: &DecompressionBudget) ->
Result<usize> {
let mut blob = records.as_ref();
let mut walked = 0;
let mut slots = 0;
let mut names = Vec::new();
while !blob.is_empty() {
- let size = take_varint(&mut blob)?;
- let size = usize::try_from(size).map_err(|_|
RecordCodecError::RecordFieldLength(size))?;
- let mut record = take_bytes(&mut blob, size)?;
-
- take_bytes(&mut record, 1)?; // attributes
- skip_varint(&mut record, MAX_VARLONG_BYTES)?; // timestamp delta
- skip_varint(&mut record, MAX_VARINT_BYTES)?; // offset delta
- skip_field(&mut record)?; // key
- skip_field(&mut record)?; // value
-
- let count = take_varint(&mut record)?;
- let limit = record.len();
- let headers = usize::try_from(count)
- .ok()
- .filter(|headers| headers.saturating_mul(MIN_HEADER_BYTES) <=
limit)
- .ok_or(RecordCodecError::HeaderCountTooLarge { count, limit })?;
+ let (headers, mut record) = take_record(&mut blob)?;
if headers > 1 {
check_header_names(&mut record, headers, &mut names)?;
}
@@ -1109,6 +1124,30 @@ fn scan_records(records: &Bytes, budget:
&DecompressionBudget) -> Result<usize>
Ok(walked)
}
+/// Steps over one record of `blob`. Returns its header count and its header
bytes, and refuses a
+/// count those bytes cannot hold.
+///
+/// The framing mirrors `Record::decode_new`, field for field.
+fn take_record<'a>(blob: &mut &'a [u8]) -> Result<(usize, &'a [u8])> {
+ let size = take_varint(blob)?;
+ let size = usize::try_from(size).map_err(|_|
RecordCodecError::RecordFieldLength(size))?;
+ let mut record = take_bytes(blob, size)?;
+
+ take_bytes(&mut record, 1)?; // attributes
+ skip_varint(&mut record, MAX_VARLONG_BYTES)?; // timestamp delta
+ skip_varint(&mut record, MAX_VARINT_BYTES)?; // offset delta
+ skip_field(&mut record)?; // key
+ skip_field(&mut record)?; // value
+
+ let count = take_varint(&mut record)?;
+ let limit = record.len();
+ let headers = usize::try_from(count)
+ .ok()
+ .filter(|headers| headers.saturating_mul(MIN_HEADER_BYTES) <= limit)
+ .ok_or(RecordCodecError::HeaderCountTooLarge { count, limit })?;
+ Ok((headers, record))
+}
+
/// Refuses a record that repeats a header name. `kafka_protocol` keeps only
the last value.
///
/// `names` is scratch space, reused across records.
@@ -1135,24 +1174,25 @@ fn check_header_names<'a>(
/// Reads a zigzag varint, five bytes at most, the way `kafka_protocol` reads
one.
fn take_varint(buf: &mut &[u8]) -> Result<i32> {
let mut value = 0u32;
- for index in 0..MAX_VARINT_BYTES {
- let byte = take_bytes(buf, 1)?[0];
+ for (index, &byte) in buf.iter().enumerate().take(MAX_VARINT_BYTES) {
value |= u32::from(byte & 0x7f) << (index * 7);
- if byte < 0x80 {
- break;
+ if byte < 0x80 || index + 1 == MAX_VARINT_BYTES {
+ *buf = &buf[index + 1..];
+ return Ok((value >> 1).cast_signed() ^ -(value & 1).cast_signed());
}
}
- Ok((value >> 1).cast_signed() ^ -(value & 1).cast_signed())
+ Err(RecordCodecError::RecordTruncated)
}
/// Steps over a zigzag varint of up to `most` bytes, for the fields the scan
does not read.
fn skip_varint(buf: &mut &[u8], most: usize) -> Result<()> {
- for _ in 0..most {
- if take_bytes(buf, 1)?[0] < 0x80 {
- break;
+ for (index, &byte) in buf.iter().enumerate().take(most) {
+ if byte < 0x80 || index + 1 == most {
+ *buf = &buf[index + 1..];
+ return Ok(());
}
}
- Ok(())
+ Err(RecordCodecError::RecordTruncated)
}
fn take_bytes<'a>(buf: &mut &'a [u8], len: usize) -> Result<&'a [u8]> {
@@ -1839,6 +1879,22 @@ mod tests {
assert!(!is_compressed(&[]));
}
+ #[test]
+ fn
given_a_batch_when_its_records_are_counted_should_read_the_declared_count() {
+ let mut records = [record_at_offset(0, b"a"), record_at_offset(1,
b"b")];
+ assert_eq!(
+ declared_records(&encode_batch(&mut records).unwrap()),
+ Some(2)
+ );
+ // `encode_batch` set the sequences that keep both records in one
batch.
+ assert_eq!(
+ declared_records(&encode_with(&records, Compression::Gzip)),
+ Some(2),
+ "the header counts records before compression"
+ );
+ assert_eq!(declared_records(&[0; BATCH_HEADER_BYTES - 1]), None);
+ }
+
#[test]
fn given_a_gzip_batch_when_decoded_should_read_it() {
let records = vec![record_at_offset(0, b"compressed")];
@@ -2059,6 +2115,45 @@ mod tests {
assert!(decode_batch(&mut batch, &enough, Zstd::Allowed).is_ok());
}
+ /// A record at `offset` with `count` headers.
+ fn record_with_headers(offset: i64, count: usize) -> Record {
+ let names: Vec<String> = (0..count).map(|index|
format!("h{index}")).collect();
+ let headers: Vec<_> = names
+ .iter()
+ .map(|name| (name.as_str(), Some(&b"v"[..])))
+ .collect();
+ let mut record = record_with(None, Some(b"v"), &headers);
+ record.offset = offset;
+ record
+ }
+
+ #[test]
+ fn
given_headers_when_slots_are_counted_should_match_what_the_decode_charges() {
+ let mut records = [
+ record_with_headers(0, 6),
+ record_with_headers(1, 3),
+ record_with_headers(2, 2),
+ ];
+ let batch = encode_batch(&mut records).unwrap();
+
+ // Three records, two slots for six headers and one for three.
+ assert_eq!(record_slots(&batch, usize::MAX), 6);
+ let exact = partition_budget(4096, 6);
+ assert!(decode_batch(&mut batch.clone(), &exact,
Zstd::Allowed).is_ok());
+ let short = partition_budget(4096, 5);
+ assert!(matches!(
+ decode_batch(&mut batch.clone(), &short, Zstd::Allowed),
+ Err(RecordCodecError::RecordBudgetExceeded { count: 6, limit: 5 })
+ ));
+
+ assert_eq!(record_slots(&batch, 4), 4, "the walk stops at the cap");
+ assert_eq!(
+ record_slots(&batch[..BATCH_HEADER_BYTES + 3], usize::MAX),
+ 3,
+ "a batch that does not walk counts its declared records"
+ );
+ }
+
/// A one-record v2 batch whose record body declares `headers` headers and
carries none.
///
/// Built by hand because no encoder writes that, and it is the shape that
reaches
diff --git a/gateways/kafka/src/server.rs b/gateways/kafka/src/server.rs
index 70ee83e2e..65a82570e 100644
--- a/gateways/kafka/src/server.rs
+++ b/gateways/kafka/src/server.rs
@@ -630,7 +630,7 @@ async fn route_frame(
}
async fn handle_connection(
- mut stream: TcpStream,
+ stream: TcpStream,
config: Arc<GatewayConfig>,
peer: SocketAddr,
state: Arc<GatewayState>,
@@ -638,7 +638,35 @@ async fn handle_connection(
cancel: CancellationToken,
) -> Result<()> {
debug!(%peer, "connection accepted");
+ let connection = ConnectionState::default();
+ let bridge = state.bridge.clone();
+ let served = serve_connection(
+ stream,
+ config,
+ peer,
+ state,
+ shared_auth,
+ cancel,
+ &connection,
+ )
+ .await;
+ // A producer without a producer id has its Produce session by connection,
so it ends here.
+ if let Some(bridge) = bridge {
+ bridge.end_connection(connection.id());
+ }
+ served
+}
+/// Reads and answers the frames of `connection` until it closes.
+async fn serve_connection(
+ mut stream: TcpStream,
+ config: Arc<GatewayConfig>,
+ peer: SocketAddr,
+ state: Arc<GatewayState>,
+ shared_auth: Arc<SharedAuth>,
+ cancel: CancellationToken,
+ connection: &ConnectionState,
+) -> Result<()> {
// A gateway with SASL off starts every connection already authenticated,
so the dispatch path
// below has one question to ask rather than two.
let mut sasl_state = if config.sasl_enabled {
@@ -647,11 +675,10 @@ async fn handle_connection(
SaslState::Authenticated
};
let mut principal: Option<AuthenticatedPrincipal> = None;
- let connection = ConnectionState::default();
let ctx = ConnectionContext {
config: &config,
state: &state,
- connection: &connection,
+ connection,
authenticator: shared_auth.authenticator.as_deref(),
auth_slots: &shared_auth.slots,
failed_logins: &shared_auth.failed_logins,
diff --git a/gateways/kafka/tests/bridge_iggy_integration_tests.rs
b/gateways/kafka/tests/bridge_iggy_integration_tests.rs
index 51af8ba70..a61aacdf5 100644
--- a/gateways/kafka/tests/bridge_iggy_integration_tests.rs
+++ b/gateways/kafka/tests/bridge_iggy_integration_tests.rs
@@ -30,8 +30,8 @@ use secrecy::SecretString;
use serial_test::serial;
use iggy_gateway_kafka::bridge::{
- BridgeError, DEFAULT_MAX_MESSAGE_SIZE, IggyBridge, IggyBridgeConfig,
TopicMapping,
- TopicOverride,
+ BridgeError, DEFAULT_MAX_MESSAGE_SIZE, IggyBridge, IggyBridgeConfig,
ProduceSessionLimits,
+ TopicMapping, TopicOverride,
};
#[path = "common/iggy_server.rs"]
@@ -394,6 +394,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,
+ produce_sessions: ProduceSessionLimits::default(),
};
let result = IggyBridge::connect(config).await;
@@ -418,6 +419,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,
+ produce_sessions: ProduceSessionLimits::default(),
};
let start = tokio::time::Instant::now();
diff --git a/gateways/kafka/tests/common/iggy_server.rs
b/gateways/kafka/tests/common/iggy_server.rs
index 51cc9a746..414519ffb 100644
--- a/gateways/kafka/tests/common/iggy_server.rs
+++ b/gateways/kafka/tests/common/iggy_server.rs
@@ -33,7 +33,9 @@ use std::time::Duration;
use iggy::prelude::{AutoLogin, Client, Credentials, IggyClient,
IggyClientBuilder};
use secrecy::SecretString;
-use iggy_gateway_kafka::bridge::{DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig,
TopicMapping};
+use iggy_gateway_kafka::bridge::{
+ DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig, ProduceSessionLimits,
TopicMapping,
+};
/// Desired slot count - the actual count `port_band()` computes may be
smaller if the machine's
/// real ephemeral range leaves little room, but this crate never needs more
than a handful live
@@ -239,7 +241,8 @@ pub struct TestServer {
child: Child,
address: String,
password: String,
- _port_guard: PortGuard,
+ /// `None` once [`Self::stop`] handed it to the caller.
+ port_guard: Option<PortGuard>,
}
impl TestServer {
@@ -306,12 +309,23 @@ impl TestServer {
child,
address,
password: password.to_string(),
- _port_guard: port_guard,
+ port_guard: Some(port_guard),
};
server.wait_ready().await;
server
}
+ /// Stops the server and keeps its port until the returned guard drops.
Without it, another
+ /// test's server can take the port while a client of this one still
retries there.
+ pub fn stop(mut self) -> PortGuard {
+ let port_guard = self
+ .port_guard
+ .take()
+ .expect("a running server holds its port");
+ drop(self);
+ port_guard
+ }
+
/// Polls with a bare TCP connect, not a full `IggyBridge::connect`: the
latter carries
/// `RECONNECTION_RETRIES` (3 retries after the first attempt, 4 dials at
~1s apart total) on
/// every failed attempt, so a poll loop built
@@ -332,6 +346,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,
+ produce_sessions: ProduceSessionLimits::default(),
}
}
}
diff --git a/gateways/kafka/tests/kafka_client_e2e_tests.rs
b/gateways/kafka/tests/kafka_client_e2e_tests.rs
index ce835c239..1ed5d3054 100644
--- a/gateways/kafka/tests/kafka_client_e2e_tests.rs
+++ b/gateways/kafka/tests/kafka_client_e2e_tests.rs
@@ -51,7 +51,7 @@ use tokio::task::JoinHandle;
use iggy_gateway_kafka::auth::IggyAuthenticator;
use iggy_gateway_kafka::bridge::{
- DEFAULT_MAX_MESSAGE_SIZE, IggyBridge, IggyBridgeConfig, TopicMapping,
+ DEFAULT_MAX_MESSAGE_SIZE, IggyBridge, IggyBridgeConfig,
ProduceSessionLimits, TopicMapping,
};
use iggy_gateway_kafka::server::bind_listener;
use iggy_gateway_kafka::{GatewayConfig, KafkaGateway};
@@ -334,6 +334,7 @@ async fn start_bridged_gateway(server: &TestServer, topics:
&[(&str, u32)]) -> B
topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new())
.expect("the default mapping is valid"),
max_message_size: DEFAULT_MAX_MESSAGE_SIZE,
+ produce_sessions: ProduceSessionLimits::default(),
};
let bridge = IggyBridge::connect(config)
.await
diff --git a/gateways/kafka/tests/produce_real_bridge_tests.rs
b/gateways/kafka/tests/produce_real_bridge_tests.rs
index 1db4ee69f..6420aa155 100644
--- a/gateways/kafka/tests/produce_real_bridge_tests.rs
+++ b/gateways/kafka/tests/produce_real_bridge_tests.rs
@@ -21,10 +21,12 @@
//! the same way in both directions cannot pass.
use std::sync::Arc;
+use std::time::Duration;
use bytes::{Bytes, BytesMut};
use iggy::prelude::{
- Consumer, HeaderKey, Identifier, IggyMessage, MessageClient,
PollingStrategy, TopicClient,
+ Consumer, HeaderKey, Identifier, IggyClient, IggyMessage, MessageClient,
PollingStrategy,
+ SystemClient, TopicClient,
};
use kafka_protocol::indexmap::IndexMap;
use kafka_protocol::protocol::StrBytes;
@@ -35,11 +37,12 @@ use kafka_protocol::records::{
use serial_test::serial;
use tokio_util::sync::CancellationToken;
-use iggy_gateway_kafka::bridge::IggyBridge;
+use iggy_gateway_kafka::bridge::{IggyBridge, Producer};
use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig};
use iggy_gateway_kafka::protocol::api::{
- API_KEY_PRODUCE, BrokerAdvertise, ERROR_NONE, ERROR_NOT_LEADER_OR_FOLLOWER,
- ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, handle_request_bounded,
+ API_KEY_INIT_PRODUCER_ID, API_KEY_PRODUCE, BrokerAdvertise, ERROR_NONE,
+ ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_UNKNOWN_TOPIC_OR_PARTITION,
GatewayState,
+ handle_request_bounded,
};
use iggy_gateway_kafka::records::{KEY_HEADER, MAPPING_VERSION, VERSION_HEADER,
from_iggy};
@@ -651,7 +654,8 @@ async fn
given_a_stopped_server_when_handled_should_answer_retriable() {
let data_dir = tempfile::tempdir().expect("tempdir");
let server = TestServer::spawn(data_dir.path()).await;
let state = gateway_with_topic(&server, 1).await;
- drop(server);
+ // The Produce session connects at its first send, after this, so the port
stays ours.
+ let _port = server.stop();
let entries = [(0, batch(&[record(0, None, b"v", &[])],
Compression::None))];
let [(_, code, _)] = produce(&state, 3, 1, TOPIC, &entries).await[..] else
{
@@ -663,6 +667,182 @@ async fn
given_a_stopped_server_when_handled_should_answer_retriable() {
);
}
+/// Long enough for any one step against a local server.
+const STEP: Duration = Duration::from_secs(10);
+
+fn soon_enough() -> tokio::time::Instant {
+ tokio::time::Instant::now() + STEP
+}
+
+/// A producer id from the gateway's own `InitProducerId`. No other id keys a
Produce session.
+async fn issued_producer_id(state: &GatewayState) -> i64 {
+ let mut enc = Encoder::with_capacity(16);
+ enc.write_nullable_string(None).expect("null fits"); // transactional_id
+ enc.write_i32(60_000); // transaction_timeout_ms
+ let response = handle_request_bounded(state, API_KEY_INIT_PRODUCER_ID, 0,
enc.freeze())
+ .await
+ .expect_response("InitProducerId is answered");
+ let mut d = Decoder::new(response);
+ assert_eq!(d.read_i32().expect("throttle time"), 0);
+ assert_eq!(d.read_i16().expect("error code"), ERROR_NONE);
+ d.read_i64().expect("producer id")
+}
+
+/// A one-record batch that carries `producer_id`, as an idempotent producer
sends it.
+fn idempotent_batch(producer_id: i64, value: &[u8]) -> Bytes {
+ let mut records = [record(0, None, value, &[])];
+ records[0].producer_id = producer_id;
+ records[0].producer_epoch = 0;
+ records[0].sequence = 0;
+ batch(&records, Compression::None)
+}
+
+#[tokio::test]
+#[serial]
+async fn
given_a_producers_request_still_running_when_handled_should_hold_back_only_that_producer()
+{
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+ let bridge = state.bridge.as_deref().expect("a bridge");
+ let id = issued_producer_id(&state).await;
+ let running = bridge
+ .produce_turn(Producer::Id { id, user: None }, soon_enough())
+ .await
+ .expect("a free turn");
+
+ let other = [(
+ 0,
+ batch(&[record(0, None, b"other", &[])], Compression::None),
+ )];
+ assert_eq!(
+ produce(&state, 3, 1, TOPIC, &other).await,
+ vec![(0, ERROR_NONE, 0)],
+ "another producer sends on a session of its own"
+ );
+
+ let own = [(0, idempotent_batch(id, b"own"))];
+ let answer = |acks| {
+ let body = encode_timed_request(3, acks, 100, TOPIC, &own);
+ handle_request_bounded(&state, API_KEY_PRODUCE, 3, body)
+ };
+ assert_eq!(
+ partition_results(3, answer(1).await.expect_response("acks=1 is
answered")),
+ vec![(0, ERROR_REQUEST_TIMED_OUT, -1)],
+ "it must not overtake the request that still runs"
+ );
+ assert!(
+ answer(0).await.is_close(),
+ "acks=0 has no response to carry the 7"
+ );
+ assert_eq!(
+ stored(&server, 0, 10).await.len(),
+ 1,
+ "nothing of the held producer was written"
+ );
+
+ drop(running);
+ assert_eq!(
+ produce(&state, 3, 1, TOPIC, &own).await,
+ vec![(0, ERROR_NONE, 1)]
+ );
+}
+
+/// Sends `count` one-message batches of `producer` to partition 0, one
request after the other.
+async fn send_each(bridge: &IggyBridge, producer: Producer, name: &str, count:
usize) {
+ let target = bridge.topic_target(TOPIC).expect("a valid topic");
+ for index in 0..count {
+ let message = IggyMessage::builder()
+ .payload(Bytes::from(format!("{name}-{index}")))
+ .build()
+ .expect("a message");
+ let (sent, held) = bridge
+ .send_records(
+ bridge
+ .produce_turn(producer.clone(), soon_enough())
+ .await
+ .expect("the producer's own turn"),
+ (),
+ &target,
+ 0,
+ vec![message],
+ soon_enough(),
+ )
+ .await;
+ sent.expect("the send lands");
+ assert!(held.is_some(), "a send that ends hands its turn back");
+ }
+}
+
+/// How many clients Iggy counts as connected, `raw` among them.
+async fn connected_clients(raw: &IggyClient) -> usize {
+ raw.get_clients().await.expect("list the clients").len()
+}
+
+async fn wait_for_clients(raw: &IggyClient, expected: usize) {
+ let reached = async {
+ while connected_clients(raw).await != expected {
+ tokio::time::sleep(Duration::from_millis(20)).await;
+ }
+ };
+ tokio::time::timeout(STEP, reached)
+ .await
+ .unwrap_or_else(|_elapsed| panic!("Iggy never counted {expected}
clients"));
+}
+
+#[tokio::test]
+#[serial]
+async fn
given_two_producers_when_they_send_at_once_should_keep_the_order_of_each() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+ let bridge = state.bridge.as_deref().expect("a bridge");
+
+ tokio::join!(
+ send_each(bridge, Producer::Connection(1), "a", 20),
+ send_each(bridge, Producer::Connection(2), "b", 20),
+ );
+ // This client, the bridge's shared one, and one for each producer.
+ assert_eq!(connected_clients(&raw_client(&server).await).await, 4);
+
+ let payloads: Vec<String> = stored(&server, 0, 100)
+ .await
+ .iter()
+ .map(|message| String::from_utf8_lossy(&message.payload).into_owned())
+ .collect();
+ assert_eq!(payloads.len(), 40);
+ for name in ["a", "b"] {
+ let sent: Vec<String> = (0..20).map(|index|
format!("{name}-{index}")).collect();
+ let stored: Vec<String> = payloads
+ .iter()
+ .filter(|payload| payload.starts_with(name))
+ .cloned()
+ .collect();
+ assert_eq!(stored, sent, "producer {name} keeps its order");
+ }
+}
+
+#[tokio::test]
+#[serial]
+async fn
given_open_produce_sessions_when_a_connection_or_the_bridge_ends_should_close_them()
{
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+ let bridge = state.bridge.as_deref().expect("a bridge");
+ let raw = raw_client(&server).await;
+
+ send_each(bridge, Producer::Connection(1), "a", 1).await;
+ send_each(bridge, Producer::Connection(2), "b", 1).await;
+ // This client, the bridge's shared one, and one for each producer.
+ wait_for_clients(&raw, 4).await;
+
+ bridge.end_connection(1);
+ wait_for_clients(&raw, 3).await;
+
+ bridge.close().await.expect("the bridge closes");
+ wait_for_clients(&raw, 1).await;
+}
+
#[tokio::test]
#[serial]
async fn
given_an_unknown_acks_value_when_handled_should_answer_invalid_required_acks() {
