You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/10/20 15:59:12 UTC
(iggy) branch master updated: feat(gateways): auto-create Kafka topics on first use (#4582)
This is an automated email from the ASF dual-hosted git repository.
hubcio 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 64ce664bf feat(gateways): auto-create Kafka topics on first use (#4582)
64ce664bf is described below
commit 64ce664bfb0c781eb4314a81f5aa58b74d4817f1
Author: Grzegorz Koszyk <11...@users.noreply.github.com>
AuthorDate: Tue Oct 20 17:59:07 2026 +0200
feat(gateways): auto-create Kafka topics on first use (#4582)
---
gateways/kafka/README.md | 82 +++-
gateways/kafka/docs/ACL_MAPPING.md | 17 +-
gateways/kafka/docs/AUTHENTICATION.md | 6 +-
gateways/kafka/docs/SCOPE.md | 2 +-
gateways/kafka/docs/TEST_SUITE.md | 2 +-
gateways/kafka/src/bridge/iggy_bridge/fetch.rs | 122 +++--
.../kafka/src/bridge/iggy_bridge/group_offsets.rs | 3 +-
gateways/kafka/src/bridge/iggy_bridge/mod.rs | 67 ++-
gateways/kafka/src/bridge/iggy_bridge/topics.rs | 199 ++++++--
gateways/kafka/src/bridge/mod.rs | 5 +-
gateways/kafka/src/bridge/topic_map.rs | 152 +++++--
gateways/kafka/src/main.rs | 108 ++++-
gateways/kafka/src/protocol/api.rs | 35 ++
gateways/kafka/src/protocol/auto_topic_creation.rs | 503 +++++++++++++++++++++
.../kafka/src/protocol/handlers/create_topics.rs | 36 +-
gateways/kafka/src/protocol/handlers/metadata.rs | 166 ++++++-
gateways/kafka/src/protocol/handlers/mod.rs | 2 +-
gateways/kafka/src/protocol/mod.rs | 1 +
gateways/kafka/src/server.rs | 25 +-
.../kafka/tests/bridge_iggy_integration_tests.rs | 12 +-
gateways/kafka/tests/common/iggy_server.rs | 13 +-
gateways/kafka/tests/kafka_client_e2e_tests.rs | 95 +++-
gateways/kafka/tests/metadata_real_bridge_tests.rs | 483 +++++++++++++++++++-
23 files changed, 1912 insertions(+), 224 deletions(-)
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index d99faf3cc..459bfd617 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -10,11 +10,16 @@ Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/34
> DeleteTopics and AlterConfigs run as the bridge's own Iggy user**: with the bridge on and
> `IGGY_KAFKA_SASL_ENABLED` off (the default), any client that can reach this port can create
> topics (up to 1000 partitions each), delete topics and their data, or shorten a topic's
-> `retention.ms` low enough that the cleaner drops sealed segments, deleting data. With SASL on,
-> `DeleteTopics` and `AlterConfigs` additionally require the authenticated principal to hold
-> `manage_topics`; without it every resource/name in the request fails
-> `TOPIC_AUTHORIZATION_FAILED` (29) and nothing is written or deleted. `CreateTopics` has no such
-> gate yet - SASL on does not protect it. Every client reads and writes as the bridge's Iggy user.
+> `retention.ms` low enough that the cleaner drops sealed segments, deleting data. With
+> `IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED=true` and no listener that names principals, any client
+> that reaches this port can also create topics just by naming them in a `Metadata` request. When
+> the connection has a principal (SASL, or a client certificate mapped to an Iggy user),
+> `DeleteTopics`, `AlterConfigs` and `Metadata` auto-creation additionally require it to hold a
+> global `manage_topics` or `manage_streams`; without it every resource of a `DeleteTopics` or
+> `AlterConfigs` request, and every missing topic a `Metadata` request would create, fails
+> `TOPIC_AUTHORIZATION_FAILED` (29) and nothing is written, created or deleted. `CreateTopics` has
+> no such gate yet - SASL on does not protect it. Every client reads and writes as the bridge's
+> Iggy user.
> See [docs/SCOPE.md](docs/SCOPE.md).
>
> InitProducerId does real work too, with or without the bridge: it allocates a producer id, so a stock idempotent producer starts instead of failing at startup.
@@ -67,6 +72,8 @@ Default bind: a `PLAINTEXT` listener on `127.0.0.1:9093`. Environment variables:
| `IGGY_KAFKA_SASL_ENABLED` | `false` | Single-listener form: require SASL/PLAIN before serving any other API (`true` or `false`, nothing else). With `IGGY_KAFKA_LISTENERS`, a `SASL_PLAINTEXT` or `SASL_SSL` listener says the same |
| `IGGY_KAFKA_PRE_AUTH_TIMEOUT_SECS` | `15` | Seconds an unauthenticated connection may sit between frames. Waiting for an authentication slot and the verification itself each get this budget, the verification's starting once it holds a slot. Separate from the 10-minute idle timeout that applies once authenticated |
| `IGGY_KAFKA_MAX_CONCURRENT_AUTHENTICATIONS` | `4` | Credential verifications the gateway runs at once, across all connections. Each costs a password hash on an Iggy shard thread. This bounds the gateway's side only: a check that times out frees its slot while its hash keeps running inside Iggy. Size it below the Iggy node's shard count |
+| `IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED` | `false` | Kafka's `auto.create.topics.enable`: a `Metadata` lookup of a missing topic creates it ([Auto topic creation](#auto-topic-creation)). Needs `IGGY_KAFKA_BRIDGE_ENABLED=true`. Off by default, unlike Kafka |
+| `IGGY_KAFKA_AUTO_CREATE_TOPICS_PARTITIONS` | `1` | Partitions of an auto-created topic, at most 1000. Only read while auto-creation is on, and refused otherwise |
## Quick start (Docker Compose)
@@ -101,8 +108,8 @@ health check before it starts. From another terminal:
kcat -b 127.0.0.1:9093 -L
```
-A full create-topic-then-produce flow needs a client that can send `CreateTopics` (there is no
-auto-create on Produce). With Python's `kafka-python` (illustrative - not exercised by an
+A full create-topic-then-produce flow needs a client that can send `CreateTopics`, unless
+`IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED=true` (see [Auto topic creation](#auto-topic-creation)). With Python's `kafka-python` (illustrative - not exercised by an
automated test, so adjust if the library's API has moved since this was written):
```python
@@ -303,8 +310,8 @@ With `requested`, a client that presents no certificate is Kafka's `ANONYMOUS` p
holds no permissions. Once any listener names principals (a SASL listener, or an `SSL` listener
with client authentication), every connection that proved no identity is `ANONYMOUS`, on a
`PLAINTEXT` listener beside it too. The requests the gateway checks against a principal,
-`DeleteTopics` and `AlterConfigs`, need `manage_topics` and are refused for it; everything else is
-served as the bridge's user. A gateway whose listeners name nobody serves every request as the
+`DeleteTopics`, `AlterConfigs` and the auto-creation a `Metadata` request asks for, need
+`manage_topics` and are refused for it; everything else is served as the bridge's user. A gateway whose listeners name nobody serves every request as the
bridge's user.
The rules are Kafka's syntax, tried in order: `RULE:pattern/replacement/[LU]` applies when the
@@ -610,6 +617,60 @@ messages or 1 MiB of unflushed data, whichever comes first). Kafka's own default
posture, so this isn't a wrong choice, but on a single node both can lose an acked write to a power
cut before that threshold is reached - worth knowing rather than discovering later.
+### Auto topic creation
+
+With `IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED=true`, the gateway behaves like a Kafka broker with
+`auto.create.topics.enable=true`. Producers, Connect source tasks, Debezium and dead-letter
+topics that rely on creation by first use then work without a `CreateTopics` call.
+
+- A `Metadata` request that names a missing topic and allows auto-creation (always below v4, the
+ request's `allow_auto_topic_creation` from v4) starts creating it with
+ `IGGY_KAFKA_AUTO_CREATE_TOPICS_PARTITIONS` partitions and answers `UNKNOWN_TOPIC_OR_PARTITION`
+ (3), as Kafka does. The client finds the topic on its next refresh.
+- A `Metadata` request for all topics never creates anything, and neither does Produce or Fetch.
+- An invalid topic name is answered `INVALID_TOPIC_EXCEPTION` (17). Kafka's internal topics,
+ `__consumer_offsets`, `__transaction_state` and `__share_group_state`, are answered
+ `UNKNOWN_TOPIC_OR_PARTITION` and never created. Any other name starting with `__`, such as a
+ Debezium heartbeat topic, is created like any other.
+- When the connection has a principal (a SASL login, a client certificate mapped to an Iggy user,
+ or `ANONYMOUS` beside either), it needs a global `manage_topics` or `manage_streams`. Only global
+ grants count: `manage_topics` granted on one stream does not, even where Iggy would let the
+ principal create the topic. A principal without it is answered `TOPIC_AUTHORIZATION_FAILED` (29)
+ for each missing topic it names, and nothing is created; topics that exist are answered as usual.
+ Kafka clients treat 29 as fatal, so a consumer whose principal may not create topics should set
+ `allow.auto.create.topics=false`, which keeps its lookups of a missing topic answered
+ `UNKNOWN_TOPIC_OR_PARTITION` (3). A principal whose permissions could not be read is answered 3
+ and nothing is created, since 29 is fatal to Java producers, consumers and Streams while Produce
+ and Fetch still work for it. An explicit `CreateTopics` is not checked against `manage_topics`
+ today, so a principal refused auto-creation can still create the same topic with `CreateTopics`.
+- A topic already being created is not started twice. At most 100 creations run at once; a name
+ past that is started by a later refresh. A creation Iggy refuses, for example at
+ `metadata.partitions_max`, is not retried for 1 second. One Iggy refuses for the bridge user's
+ permissions is not retried for a minute: Iggy refuses those only after journaling the create, so
+ each attempt costs a replicated metadata write. A name waiting out either still counts toward the
+ 100. Its first failure is logged at `warn`, and repeats within a minute at `debug`.
+- Creations run as the bridge's Iggy user, like `CreateTopics`, on an Iggy client of their own.
+ Both create the topic first, and the backing Iggy stream, which the topic mapping names and the
+ client never does, only when Iggy reports it missing. The bridge user needs `manage_topics` on
+ that stream, or globally, to create a topic in a stream that exists, and the global
+ `manage_streams` to create a missing stream. A principal with `manage_topics` but no stream
+ rights still gets a new stream when the mapping names one that does not exist yet.
+- Like every topic the gateway creates, an auto-created topic has no message expiry (see
+ [Provisioning and idempotency](#provisioning-and-idempotency)): its messages are kept until a
+ client sets `retention.ms` with `AlterConfigs`.
+- Shutdown cancels the creations still running before the bridge signs out, and drops their
+ client if any was running, so the sign-out does not wait behind it. An embedder that builds its
+ own `GatewayState` calls `shutdown_auto_topic_creation` before closing the bridge.
+- `IGGY_KAFKA_AUTO_CREATE_TOPICS_PARTITIONS` applies to auto-creation only. A `CreateTopics` with
+ `num_partitions=-1` still creates 1 partition.
+
+As in Kafka, a producer or consumer still running recreates a topic right after `DeleteTopics`
+removes it, on its next `Metadata` refresh. A Java consumer with `allow.auto.create.topics=true`
+creates whatever topic it subscribes to, including one whose name is a typo.
+
+With auto-creation off, the default, nothing changes: a missing topic is answered
+`UNKNOWN_TOPIC_OR_PARTITION` and a client waits until something else creates it.
+
### Concurrency ceiling
- Each producer sends on an `IggyClient` of its own, one request at a time, so its batches land
@@ -631,7 +692,8 @@ cut before that threshold is reached - worth knowing rather than discovering lat
- `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, and a `ListOffsets` lookup
- by time borrows one of those slots for its request. Topic probes use 1 more.
+ by time borrows one of those slots for its request. Topic probes use 1 more. With auto topic
+ creation on, creations use 1 more, opened at the first one.
Offset calls use 8 more: a commit client and a read client per offset slot.
- 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.
diff --git a/gateways/kafka/docs/ACL_MAPPING.md b/gateways/kafka/docs/ACL_MAPPING.md
index 2fe7c455c..edc2c8542 100644
--- a/gateways/kafka/docs/ACL_MAPPING.md
+++ b/gateways/kafka/docs/ACL_MAPPING.md
@@ -18,12 +18,14 @@ fetches the principal's own record, and the connection keeps the result. No seco
stored password, one extra round trip on a path that already does one.
**Global permissions only, as wildcard bindings.** Iggy's per-stream and per-topic permissions are
-not rendered. See [What is not mapped](#what-is-not-mapped).
+not rendered, and the checks the gateway makes against a principal read the same global grants. See
+[What is not mapped](#what-is-not-mapped).
## Why this is worth doing before the data plane exists
-Authorization cannot be *enforced* yet. Produce and Fetch act as the bridge's Iggy user for every
-client, so no operation runs as the principal, and the verified identity is discarded.
+Authorization cannot be *enforced* on the data plane yet. Produce and Fetch act as the bridge's
+Iggy user for every client, so no read or write runs as the principal. The gateway checks a
+principal only before a few admin requests, listed under [What is not mapped](#what-is-not-mapped).
`DescribeAcls` is the one authorization surface that needs neither. It is a read: take the
authenticated principal, take the permissions Iggy already holds for it, and answer in Kafka's
@@ -132,8 +134,13 @@ needs `manage_users` on the *Kafka client's own* Iggy user. Coherent, but it tur
surface into one that mutates accounts, and it cannot be verified the way a read can. Left out
deliberately rather than half-built.
-**Enforcement.** Nothing here gates an operation. This describes what Iggy would allow; Iggy is
-still the thing that decides, once there is an operation for it to decide about.
+**Enforcement of per-stream grants.** The gateway refuses `DeleteTopics`, `AlterConfigs` and the
+auto topic creation a `Metadata` request asks for unless the principal holds `manage_topics` or
+`manage_streams` in the global projection above. A principal granted `manage_topics` on one stream
+alone is therefore refused, though Iggy would let it manage topics there. A refused auto-creation is
+answered `TOPIC_AUTHORIZATION_FAILED` (29), which Kafka clients treat as fatal (see the README's
+[Auto topic creation](../README.md#auto-topic-creation)). A request that passes still runs as the
+bridge's Iggy user, and Iggy decides it against that user's grants.
## One divergence from a real broker
diff --git a/gateways/kafka/docs/AUTHENTICATION.md b/gateways/kafka/docs/AUTHENTICATION.md
index 31ee2d072..739c1d019 100644
--- a/gateways/kafka/docs/AUTHENTICATION.md
+++ b/gateways/kafka/docs/AUTHENTICATION.md
@@ -272,9 +272,9 @@ first frame:
the name unchecked.
With `requested`, a client that presents no certificate is Kafka's `ANONYMOUS` principal, which holds no
-permissions: `DescribeAcls` describes it as holding nothing, and `DeleteTopics` and `AlterConfigs`, which
-need `manage_topics`, are refused. Everything else is served as the bridge's user. With `required`, rustls
-fails the handshake instead.
+permissions: `DescribeAcls` describes it as holding nothing, and `DeleteTopics`, `AlterConfigs` and the auto
+topic creation a `Metadata` request asks for, which need a global `manage_topics`, are refused. Everything
+else is served as the bridge's user. With `required`, rustls fails the handshake instead.
The same holds on every listener that names nobody once any listener names principals: a `PLAINTEXT`
listener beside a `SASL_SSL` or mTLS one serves `ANONYMOUS`, so it cannot do what an authenticated user
diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md
index 62fda57bc..2cd185bde 100644
--- a/gateways/kafka/docs/SCOPE.md
+++ b/gateways/kafka/docs/SCOPE.md
@@ -50,7 +50,7 @@ it knows the server supports flexible encoding.
| API key | Name | Min version | Max version | Valid versions | Behavior |
| --------- | ------ | ------------- | ------------- | ---------------- | ---------- |
| 18 | ApiVersions | 0 | 3 | 0, 1, 2, 3 | Advertise supported ranges; flexible encoding at v3+ |
-| 3 | Metadata | 0 | 9 | 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 | Decode topic list count; stub broker host from the listener's advertised address or its bound IP; flexible encoding at v9+ |
+| 3 | Metadata | 0 | 9 | 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 | Decode topic list count; stub broker host from the listener's advertised address or its bound IP; flexible encoding at v9+. With `IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED=true`, a named missing topic is created as in Kafka (README "Auto topic creation") |
| 0 | Produce | 3 | 9 | 3, 4, 5, 6, 7, 8, 9 | With a bridge: one `send_messages` per partition. Without one: stub returns `NOT_LEADER_OR_FOLLOWER` (6) |
| 1 | Fetch | 4 | 12 | 4, 5, 6, 7, 8, 9, 10, 11, 12 | With a bridge: shared topic probes, paged `poll_messages` per partition with records, wait up to `max_wait_ms` (max 18 s). Without one: stub returns `NOT_LEADER_OR_FOLLOWER` (6) |
| 2 | ListOffsets | 1 | 7 | 1, 2, 3, 4, 5, 6, 7 | With a bridge: LATEST, EARLIEST, a timestamp and, from v7, `MAX_TIMESTAMP` (see `BRIDGE_MAPPING.md`). Without one: stub response |
diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md
index be99855b7..c8e668852 100644
--- a/gateways/kafka/docs/TEST_SUITE.md
+++ b/gateways/kafka/docs/TEST_SUITE.md
@@ -79,7 +79,7 @@ file under `tests/` anymore.
| [`listener_robustness_tests.rs`](../tests/listener_robustness_tests.rs) | TCP listener robustness — framing, pipelining, concurrency, connection limits | No |
| [`sasl_tests.rs`](../tests/sasl_tests.rs) | SASL/PLAIN over a socket — full handshake, every refusal path, and the disabled default. Drives a stub verifier implementing `SaslAuthenticator`, so no Iggy server is needed | No |
| [`tls_listener_tests.rs`](../tests/tls_listener_tests.rs) | The `SSL` and `SASL_SSL` listeners over a real socket with certificates minted per test: each listener advertising its own port, a plaintext client refused on the TLS port, SASL inside TLS, client certificates mapped to users through a stub `PrincipalResolver` (`required`, `requested`, unknown user, a subject no rule maps, a failing or hung lookup, `SASL_SSL` asking for no certificate), a multi-MiB response to a slow reader, t [...]
-| [`kafka_client_e2e_tests.rs`](../tests/kafka_client_e2e_tests.rs) | **Real Kafka clients** against the whole stack: a spawned `iggy-server`, the gateway in-process with a real authenticator, and kcat / the Java tools from containers. With a bridge, kcat writes records, and kcat and the Java consumer read them back through Fetch. Java consumers in one group resume from their commits, across a gateway restart and a member leaving. Lookups by time resolve through `kafka-get-offsets` (a ti [...]
+| [`kafka_client_e2e_tests.rs`](../tests/kafka_client_e2e_tests.rs) | **Real Kafka clients** against the whole stack: a spawned `iggy-server`, the gateway in-process with a real authenticator, and kcat / the Java tools from containers. With a bridge, kcat writes records, and kcat and the Java consumer read them back through Fetch. Java consumers in one group resume from their commits, across a gateway restart and a member leaving. Lookups by time resolve through `kafka-get-offsets` (a ti [...]
| [`bridge_iggy_integration_tests.rs`](../tests/bridge_iggy_integration_tests.rs) | `IggyBridge` against a real, spawned `iggy-server` — provisioning idempotency, high watermark, credential/connection edge cases, and the bridge dialing a TLS-only server over TLS | No (needs the `iggy-server` binary - see Prerequisites) |
| [`list_offsets_real_bridge_tests.rs`](../tests/list_offsets_real_bridge_tests.rs) | ListOffsets (key 2) through the whole handler against a real, spawned `iggy-server`: LATEST, EARLIEST, lookups by time and `MAX_TIMESTAMP`, the `-1` answers, code 3, several partitions looked up by time in one request, and a stalled lookup leaving the later partitions their answers | No (needs the `iggy-server` binary - see Prerequisites) |
| [`fetch_real_bridge_tests.rs`](../tests/fetch_real_bridge_tests.rs) | Fetch (key 1) through the whole handler against a real, spawned `iggy-server` — records go in through the Iggy SDK or the Produce handler and come back through `kafka_protocol`'s client decoder, at every version, with codes 0, 1, 3, 6, -1 and 70, paged polls, and the wait | No (needs the `iggy-server` binary - see Prerequisites) |
diff --git a/gateways/kafka/src/bridge/iggy_bridge/fetch.rs b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
index 1ae0fab4b..2e3325888 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
@@ -137,38 +137,72 @@ impl LazyClient {
}
}
- /// Runs `call` until `deadline`. A call that times out drops the client.
- pub(super) async fn until<T>(
+ /// Connects, then runs `call` on the client until `deadline`. A call that times out drops
+ /// the client it ran on.
+ pub(super) async fn until<T, Call, Fut>(
&self,
+ config: &IggyBridgeConfig,
deadline: Instant,
- call: impl Future<Output = Result<T, BridgeError>>,
- ) -> Result<T, BridgeError> {
- let result = timeout_at(deadline, call)
+ call: Call,
+ ) -> Result<T, BridgeError>
+ where
+ Call: FnOnce(Arc<IggyClient>) -> Fut,
+ Fut: Future<Output = Result<T, BridgeError>>,
+ {
+ let mut used = None;
+ let run = async {
+ let client = self.connected(config).await?;
+ used = Some(Arc::clone(&client));
+ call(client).await
+ };
+ let result = timeout_at(deadline, run)
.await
.unwrap_or(Err(BridgeError::Timeout));
- if matches!(result, Err(BridgeError::Timeout)) {
- self.0.lock().await.take();
+ let timed_out = matches!(result, Err(BridgeError::Timeout));
+ if let Some(used) = used.filter(|_| timed_out) {
+ self.forget(&used).await;
}
result
}
+ /// Drops `client` from the slot, unless another call already replaced it with a new one.
+ async fn forget(&self, client: &Arc<IggyClient>) {
+ let mut held = self.0.lock().await;
+ if held
+ .as_ref()
+ .is_some_and(|current| Arc::ptr_eq(current, client))
+ {
+ held.take();
+ }
+ }
+
/// [`Self::until`] on a task of its own, so the deadline still drops a stuck client when the
/// caller stops waiting first.
- pub(super) async fn until_detached<T, Call>(
+ pub(super) async fn until_detached<T, Call, Fut>(
&self,
+ config: IggyBridgeConfig,
deadline: Instant,
call: Call,
) -> Result<T, BridgeError>
where
T: Send + 'static,
- Call: Future<Output = Result<T, BridgeError>> + Send + 'static,
+ Call: FnOnce(Arc<IggyClient>) -> Fut + Send + 'static,
+ Fut: Future<Output = Result<T, BridgeError>> + Send,
{
let client = self.clone();
- tokio::spawn(async move { client.until(deadline, call).await })
+ tokio::spawn(async move { client.until(&config, deadline, call).await })
.await
.unwrap_or(Err(BridgeError::Timeout))
}
+ /// Takes the client out of the slot without signing it out, so `close` does not wait behind
+ /// a call still running on it. Leaves a client that is connecting right now.
+ pub(super) fn discard(&self) {
+ if let Ok(mut held) = self.0.try_lock() {
+ held.take();
+ }
+ }
+
/// 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 {
@@ -307,17 +341,15 @@ impl IggyBridge {
}
async fn probe_target(&self, target: &TopicTarget) -> Result<TopicProbe, BridgeError> {
- let client = &self.probe_client;
- let get_topic = async {
+ let get_topic = |client: Arc<IggyClient>| async move {
client
- .connected(&self.config)
- .await?
.get_topic(&target.stream_id, &target.topic_id)
.await
.map_err(BridgeError::Iggy)
};
- let details = client
- .until(Instant::now() + REQUEST_TIMEOUT, get_topic)
+ let details = self
+ .probe_client
+ .until(&self.config, Instant::now() + REQUEST_TIMEOUT, get_topic)
.await?
// A missing stream reads as `None` too. Both answer the same Kafka code.
.ok_or_else(|| {
@@ -397,10 +429,8 @@ impl IggyBridge {
) -> Result<PolledMessages, BridgeError> {
let target = self.topic_target(kafka_topic)?;
let consumer = Consumer::default();
- let poll = async {
- client
- .connected(&self.config)
- .await?
+ let poll = |connected: Arc<IggyClient>| async move {
+ connected
.poll_messages(
&target.stream_id,
&target.topic_id,
@@ -415,7 +445,7 @@ impl IggyBridge {
};
// The SDK replays a poll that Iggy refuses for up to 30 s. Stop with the request, so the
// slot frees.
- client.until(deadline, poll).await
+ client.until(&self.config, deadline, poll).await
}
}
@@ -458,6 +488,14 @@ mod tests {
Arc::new(client)
}
+ fn config() -> IggyBridgeConfig {
+ IggyBridge::offline().config
+ }
+
+ fn stuck(_client: Arc<IggyClient>) -> std::future::Pending<Result<u8, BridgeError>> {
+ std::future::pending()
+ }
+
#[tokio::test]
async fn given_a_detached_call_whose_caller_gave_up_when_its_deadline_passes_should_drop_the_client()
{
@@ -467,7 +505,7 @@ mod tests {
let caller = tokio::time::timeout(
Duration::from_millis(10),
- client.until_detached(deadline, std::future::pending::<Result<u8, BridgeError>>()),
+ client.until_detached(config(), deadline, stuck),
)
.await;
assert!(
@@ -488,8 +526,9 @@ mod tests {
*client.0.lock().await = Some(Arc::clone(&held));
let soon = || Instant::now() + Duration::from_millis(20);
+ let config = config();
let ended = client
- .until(soon(), async { Ok::<_, BridgeError>(7) })
+ .until(&config, soon(), |_| async { Ok::<_, BridgeError>(7) })
.await;
assert_eq!(ended.expect("the call ended in time"), 7);
let kept = client
@@ -501,7 +540,7 @@ mod tests {
assert!(Arc::ptr_eq(&kept, &held));
let failed = client
- .until(soon(), async {
+ .until(&config, soon(), |_| async {
Err::<u8, _>(BridgeError::Iggy(IggyError::Disconnected))
})
.await;
@@ -511,16 +550,43 @@ mod tests {
"an error the SDK returned leaves no call behind"
);
- let stuck = client
- .until(soon(), std::future::pending::<Result<u8, BridgeError>>())
- .await;
- assert!(matches!(stuck, Err(BridgeError::Timeout)));
+ let timed_out = client.until(&config, soon(), stuck).await;
+ assert!(matches!(timed_out, Err(BridgeError::Timeout)));
assert!(
client.0.lock().await.is_none(),
"the next call connects a new client"
);
}
+ #[tokio::test]
+ async fn given_a_call_past_its_deadline_when_another_call_replaced_its_client_should_keep_the_new_one()
+ {
+ let client = LazyClient::default();
+ let (old, new) = (unconnected(), unconnected());
+ client.hold(Arc::clone(&old)).await;
+ let config = config();
+ let swap = client.clone();
+ let replaced = Arc::clone(&new);
+
+ let timed_out = client
+ .until(
+ &config,
+ Instant::now() + Duration::from_millis(20),
+ |used| {
+ assert!(Arc::ptr_eq(&used, &old));
+ async move {
+ swap.hold(replaced).await;
+ std::future::pending::<Result<u8, BridgeError>>().await
+ }
+ },
+ )
+ .await;
+
+ assert!(matches!(timed_out, Err(BridgeError::Timeout)));
+ let kept = client.held().await.expect("the new client stays");
+ assert!(Arc::ptr_eq(&kept, &new));
+ }
+
#[tokio::test]
async fn given_every_slot_taken_when_one_is_dropped_should_hand_its_client_to_the_next_read() {
let pool = Arc::new(FetchPool::new());
diff --git a/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
index b3f5bbebb..922a16a70 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
@@ -247,8 +247,7 @@ impl IggyBridge {
let Some((_queued, _turn)) = calls.take(slot, lane).await else {
return Err(BridgeError::Timeout);
};
- let read = async { call(lane.client.connected(&self.config).await?).await };
- lane.client.until(calls.deadline, read).await
+ lane.client.until(&self.config, calls.deadline, call).await
}
}
diff --git a/gateways/kafka/src/bridge/iggy_bridge/mod.rs b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
index d4e717b8f..9ddb87c9b 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
@@ -46,7 +46,10 @@ pub use group_offsets::{OFFSET_GROUP_PREFIX, OffsetCalls};
pub(crate) use offsets::{LocatedRecord, RecordLookup};
use produce::ProducePool;
pub use produce::{ProduceTurn, Producer};
-pub use topics::{KafkaTopicMetadata, StreamTopicCache, TopicCreationOutcome, TopicLoad};
+pub use topics::{
+ DEFAULT_TOPIC_PARTITIONS, KafkaTopicMetadata, MAX_TOPIC_PARTITIONS, StreamTopicCache,
+ TopicCreationOutcome, TopicLoad,
+};
/// Passes attempted, after the first, before [`IggyBridge::connect`] gives up and returns `Err`.
///
@@ -135,8 +138,9 @@ async fn permit_by(turn: &Arc<Semaphore>, deadline: Instant) -> Option<OwnedSema
///
/// 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 and certificate user lookups each use a client of
-/// their own, Fetch polls one per read slot, and offset calls two per offset slot.
+/// `AlterConfigs` share another. Topic probes, certificate user lookups and `Metadata`'s auto topic
+/// creations each 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.
@@ -153,6 +157,8 @@ pub struct IggyBridge {
probe_client: LazyClient,
/// The client of every certificate user lookup. See `resolve`.
lookup_client: LazyClient,
+ /// The client of every auto topic creation. See `auto_create_kafka_topic`.
+ auto_create_client: LazyClient,
/// Group offset slots, each with a commit client and a read client. See `OffsetPool`.
offset_pool: OffsetPool,
}
@@ -195,6 +201,7 @@ impl IggyBridge {
fetch_pool: Arc::new(FetchPool::new()),
probe_client: LazyClient::default(),
lookup_client: LazyClient::default(),
+ auto_create_client: LazyClient::default(),
offset_pool: OffsetPool::new(),
})
}
@@ -226,6 +233,25 @@ impl IggyBridge {
(stream.to_string(), topic.to_string())
}
+ /// Whether `kafka_topic` is no override's key yet resolves to an override's Iggy topic. See
+ /// [`TopicMapping::aliases_an_override`](crate::bridge::TopicMapping::aliases_an_override).
+ #[must_use]
+ pub(crate) fn aliases_an_override(&self, kafka_topic: &str) -> bool {
+ self.config.topic_mapping.aliases_an_override(kafka_topic)
+ }
+
+ /// Drops the auto topic creation client without signing it out, so [`Self::close`] does not
+ /// wait behind a creation still running on it. Iggy ends its session when the lease runs out.
+ pub(crate) fn discard_auto_create_client(&self) {
+ self.auto_create_client.discard();
+ }
+
+ /// Whether the auto topic creation client is in its slot.
+ #[cfg(test)]
+ pub(crate) async fn holds_auto_create_client(&self) -> bool {
+ self.auto_create_client.held().await.is_some()
+ }
+
/// Signs out and tears down every Iggy client, including its background heartbeat task.
///
/// Not `IggyClient::disconnect`: that only tears down the transport
@@ -256,6 +282,7 @@ impl IggyBridge {
fetch_clients,
probe_client,
lookup_client,
+ auto_create_client,
offset_clients,
shared_client,
) = tokio::join!(
@@ -263,6 +290,7 @@ impl IggyBridge {
self.fetch_pool.close(),
self.probe_client.close(),
self.lookup_client.close(),
+ self.auto_create_client.close(),
self.offset_pool.close(),
sign_out_and_shut_down(&self.client),
);
@@ -270,6 +298,7 @@ impl IggyBridge {
.and(fetch_clients)
.and(probe_client)
.and(lookup_client)
+ .and(auto_create_client)
.and(offset_clients)
.and(shared_client)
}
@@ -278,13 +307,22 @@ impl IggyBridge {
/// A timeout keeps the SDK's retained request and never retries it as a new mutation.
async fn metadata_write<T, F: Future<Output = Result<T, IggyError>>>(
&self,
- mut write: impl FnMut() -> F,
+ write: impl FnMut() -> F,
) -> Result<T, BridgeError> {
- let deadline = (Instant::now() + REQUEST_TIMEOUT).into_std();
- with_request_timeout(write_with_recovery(&self.client, deadline, &mut write)).await
+ metadata_write_on(&self.client, write).await
}
}
+/// [`IggyBridge::metadata_write`] on `client`, for the writes that run on the admin client or on
+/// the auto topic creation client.
+async fn metadata_write_on<T, F: Future<Output = Result<T, IggyError>>>(
+ client: &IggyClient,
+ mut write: impl FnMut() -> F,
+) -> Result<T, BridgeError> {
+ let deadline = (Instant::now() + REQUEST_TIMEOUT).into_std();
+ with_request_timeout(write_with_recovery(client, deadline, &mut write)).await
+}
+
async fn write_with_recovery<T, F: Future<Output = Result<T, IggyError>>>(
client: &IggyClient,
deadline: std::time::Instant,
@@ -337,8 +375,11 @@ impl IggyBridge {
produce_sessions: crate::bridge::ProduceSessionLimits::default(),
tls: crate::bridge::IggyTls::default(),
};
+ let client = Arc::new(client);
Self {
- client: Arc::new(client),
+ // Held rather than lazy, so an auto topic creation fails at once instead of dialing.
+ auto_create_client: LazyClient::holding(Arc::clone(&client)),
+ client,
produce_pool: Arc::new(ProducePool::new(config.produce_sessions)),
config,
fetch_pool: Arc::new(FetchPool::new()),
@@ -399,15 +440,17 @@ impl std::fmt::Debug for IggyBridge {
#[async_trait]
impl PrincipalResolver for IggyBridge {
async fn resolve(&self, username: &str) -> Result<AuthenticatedPrincipal, AuthError> {
- let client = self.lookup_client.clone();
- let config = self.config.clone();
let username = username.to_owned();
- let lookup = async move {
- let connected = client.connected(&config).await?;
+ let lookup = |connected: Arc<IggyClient>| async move {
Ok(resolve_principal(connected.as_ref(), &username).await)
};
let deadline = Instant::now() + REQUEST_TIMEOUT;
- match self.lookup_client.until_detached(deadline, lookup).await {
+ let config = self.config.clone();
+ match self
+ .lookup_client
+ .until_detached(config, deadline, lookup)
+ .await
+ {
Ok(resolved) => resolved,
Err(error) => {
warn!(%error, "could not reach Iggy to look up the certificate's user");
diff --git a/gateways/kafka/src/bridge/iggy_bridge/topics.rs b/gateways/kafka/src/bridge/iggy_bridge/topics.rs
index f5a8c13fc..6df32f3ff 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/topics.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/topics.rs
@@ -18,6 +18,7 @@
//! Stream and topic provisioning.
use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
use iggy::prelude::{
Identifier, IggyClient, IggyError, IggyExpiry, StreamClient, Topic, TopicClient,
@@ -27,11 +28,25 @@ use kafka_protocol::protocol::StrBytes;
use tokio::time::{Instant, timeout_at};
use tracing::{debug, info, warn};
-use super::{IggyBridge, with_request_timeout};
+use super::{IggyBridge, REQUEST_TIMEOUT, metadata_write_on, with_request_timeout};
use crate::bridge::error::BridgeError;
use crate::bridge::topic_map::validate_kafka_topic_name;
use crate::protocol::handlers::topic_config::message_expiry_is_explicit;
+/// Partitions of a topic whose creator leaves the count to the broker: Kafka's own
+/// `num.partitions` default.
+pub const DEFAULT_TOPIC_PARTITIONS: u32 = 1;
+
+/// Iggy's own server-side per-topic partition cap (`IggyError::TooManyPartitions`).
+///
+/// Not exported by `iggy::prelude`. Checked before a create so a count Iggy can never create is
+/// refused the same way on every path, including `CreateTopics`' `validate_only`.
+pub const MAX_TOPIC_PARTITIONS: u32 = 1000;
+
+/// How long one auto topic creation may take: a connect, `create_topic`, and for a missing stream
+/// `create_stream` and `create_topic` again, each bounded by [`REQUEST_TIMEOUT`] on its own.
+const AUTO_CREATE_LIMIT: std::time::Duration = REQUEST_TIMEOUT.saturating_mul(4);
+
/// Outcome of [`IggyBridge::create_kafka_topic`].
///
/// Distinguishes "this call is the one that created it" from "it already existed", so
@@ -84,7 +99,7 @@ impl IggyBridge {
/// already accepts that sentinel at the wire-validation layer). The broker default is for
/// the first caller with a real Kafka request (`Metadata` or `CreateTopics`) to decide.
///
- /// No caching: every call pays a stream-create attempt (`Self::ensure_stream`) and a
+ /// No caching: every call pays a stream-create attempt (`ensure_stream`) and a
/// `get_topic` (two round trips once both already exist), even for a topic this same bridge
/// already confirmed a moment ago. A cache keyed on `kafka_topic` would remove that cost, but
/// would also have to answer "how does a cache entry ever get invalidated" - the topic being
@@ -115,57 +130,16 @@ impl IggyBridge {
});
}
let (stream_name, topic_name) = self.config.topic_mapping.resolve(kafka_topic);
- let stream_id = self.ensure_stream(stream_name).await?;
+ let stream_id = ensure_stream(&self.client, stream_name).await?;
self.ensure_topic(&stream_id, topic_name, kafka_topic, partition_count)
.await?;
Ok(())
}
- /// Ensures the stream named `stream_name` exists, creating it if missing.
- ///
- /// `Identifier::named` - never `Identifier::try_from`/`FromStr` - because the latter parses
- /// an all-digit string as a numeric Iggy ID rather than a name. A stream or topic named e.g.
- /// `"42"` would otherwise resolve against the wrong resource on every call after the first:
- /// the first `ensure_stream_and_topic("42", ...)` creates a stream *named* `"42"`, but a
- /// second call would look it up *by ID* `42` instead, almost certainly finding nothing and
- /// breaking the "idempotent on repeated calls" guarantee.
- ///
- /// Returns the same *named* `Identifier` it was given, not the numeric id the SDK hands back
- /// from `get`/`create` - streams are backed by a recycled slab (`core/metadata`'s
- /// `stm/stream.rs`: freed keys are reused by the next created stream), so a numeric id
- /// captured here could point at a *different* stream by the time `ensure_topic` uses it, if
- /// this stream is deleted and recreated in between. The name has no such window.
- ///
- /// Attempts the create directly rather than probing existence first: `get_stream` answers
- /// with `StreamDetails`, which embeds every topic header in the stream (`iggy_common`'s
- /// `StreamDetails.topics: Vec<Topic>`), not just the stream's own metadata - a probe paid on
- /// every call, including the steady-state case where the stream (almost always) already
- /// exists, could cost one full per-stream topic listing per topic in a `CreateTopics` batch
- /// (up to 100 today). `StreamNameAlreadyExists` on the create attempt is exactly as
- /// informative as a prior `get_stream` would have been - the stream exists either way - so
- /// this pays one round trip in every case instead of up to two in the common one.
- async fn ensure_stream(&self, stream_name: &str) -> Result<Identifier, BridgeError> {
- let identifier = Identifier::named(stream_name).map_err(BridgeError::Iggy)?;
- match self
- .metadata_write(|| self.client.create_stream(stream_name))
- .await
- {
- Ok(_created) => {
- info!("created Iggy stream '{stream_name}'");
- Ok(identifier)
- }
- Err(BridgeError::Iggy(IggyError::StreamNameAlreadyExists(_))) => {
- debug!("Iggy stream '{stream_name}' already exists");
- Ok(identifier)
- }
- Err(err) => Err(err),
- }
- }
-
/// Looks up (or creates) the topic named `topic_name` under `stream_id`.
///
/// `Identifier::named`, not `Identifier::try_from` - the same numeric-name ambiguity
- /// [`Self::ensure_stream`]'s doc comment describes for stream names applies to topic names.
+ /// [`ensure_stream`]'s doc comment describes for stream names applies to topic names.
///
/// `partition_count == 0` is unreachable here: [`IggyBridge::ensure_stream_and_topic`] rejects
/// it with [`BridgeError::InvalidPartitionCount`] before calling this method. The server
@@ -449,6 +423,49 @@ impl IggyBridge {
&self,
kafka_topic: &str,
partition_count: u32,
+ ) -> Result<TopicCreationOutcome, BridgeError> {
+ self.create_kafka_topic_on(&self.client, kafka_topic, partition_count)
+ .await
+ }
+
+ /// [`Self::create_kafka_topic`] on a client of its own, for `Metadata`'s auto topic creation.
+ ///
+ /// Not the admin client: the SDK runs one request at a time per client, so creations queued
+ /// behind a slow Iggy would hold every Metadata lookup and `CreateTopics` behind them. A call
+ /// that times out drops this client, so the next creation dials a new one instead of queueing
+ /// behind the exchange that is still pending.
+ ///
+ /// # Errors
+ ///
+ /// Same as [`Self::create_kafka_topic`].
+ pub(crate) async fn auto_create_kafka_topic(
+ &self,
+ kafka_topic: &str,
+ partition_count: u32,
+ ) -> Result<TopicCreationOutcome, BridgeError> {
+ let create = |connected: Arc<IggyClient>| async move {
+ self.create_kafka_topic_on(&connected, kafka_topic, partition_count)
+ .await
+ };
+ self.auto_create_client
+ .until(&self.config, Instant::now() + AUTO_CREATE_LIMIT, create)
+ .await
+ }
+
+ /// [`Self::create_kafka_topic`] on `client`.
+ ///
+ /// Creates the topic first, and the stream only when Iggy reports it missing, which costs one
+ /// refused `create_topic` for each stream that has to be created. Iggy refuses a
+ /// `create_stream` of a stream that exists, or one the bridge user may not create, only after
+ /// journaling and replicating it, so creating the stream first would cost every creation, and
+ /// every retry of a refused one, a metadata write. It would also lock out a bridge user that
+ /// holds `manage_topics` on the stream alone: `create_topic` accepts that grant, while
+ /// `create_stream` needs the global `manage_streams`.
+ async fn create_kafka_topic_on(
+ &self,
+ client: &IggyClient,
+ kafka_topic: &str,
+ partition_count: u32,
) -> Result<TopicCreationOutcome, BridgeError> {
validate_kafka_topic_name("kafka_topic", kafka_topic)?;
if partition_count == 0 {
@@ -457,16 +474,27 @@ impl IggyBridge {
});
}
let (stream_name, topic_name) = self.config.topic_mapping.resolve(kafka_topic);
- let stream_id = self.ensure_stream(stream_name).await?;
+ let stream_id = Identifier::named(stream_name).map_err(BridgeError::Iggy)?;
let options = TopicCreateOptions {
partitions_count: Some(partition_count),
..TopicCreateOptions::default()
};
- match self
- .metadata_write(|| self.client.create_topic(&stream_id, topic_name, &options))
- .await
- {
+ let create_topic = || {
+ metadata_write_on(client, || {
+ client.create_topic(&stream_id, topic_name, &options)
+ })
+ };
+ let created = match create_topic().await {
+ Err(BridgeError::Iggy(
+ IggyError::StreamIdNotFound(_) | IggyError::StreamNameNotFound(_),
+ )) => {
+ ensure_stream(client, stream_name).await?;
+ create_topic().await
+ }
+ created => created,
+ };
+ match created {
Ok(created) => {
info!("created Iggy topic '{topic_name}' with {partition_count} partitions");
// Same postcondition check ensure_topic's own create branch makes: partitions_count
@@ -586,6 +614,59 @@ impl IggyBridge {
}
}
+/// Ensures the stream named `stream_name` exists, creating it if missing.
+///
+/// `Identifier::named` - never `Identifier::try_from`/`FromStr` - because the latter parses
+/// an all-digit string as a numeric Iggy ID rather than a name. A stream or topic named e.g.
+/// `"42"` would otherwise resolve against the wrong resource on every call after the first:
+/// the first `ensure_stream_and_topic("42", ...)` creates a stream *named* `"42"`, but a
+/// second call would look it up *by ID* `42` instead, almost certainly finding nothing and
+/// breaking the "idempotent on repeated calls" guarantee.
+///
+/// Returns the same *named* `Identifier` it was given, not the numeric id the SDK hands back
+/// from `get`/`create` - streams are backed by a recycled slab (`core/metadata`'s
+/// `stm/stream.rs`: freed keys are reused by the next created stream), so a numeric id
+/// captured here could point at a *different* stream by the time `ensure_topic` uses it, if
+/// this stream is deleted and recreated in between. The name has no such window.
+///
+/// Attempts the create directly rather than probing existence first: `get_stream` answers
+/// with `StreamDetails`, which embeds every topic header in the stream (`iggy_common`'s
+/// `StreamDetails.topics: Vec<Topic>`), not just the stream's own metadata - a probe paid on
+/// every call, including the steady-state case where the stream (almost always) already
+/// exists, could cost one full per-stream topic listing per topic in a `CreateTopics` batch
+/// (up to 100 today). `StreamNameAlreadyExists` on the create attempt is exactly as
+/// informative as a prior `get_stream` would have been - the stream exists either way - so
+/// this pays one round trip in every case instead of up to two in the common one.
+async fn ensure_stream(client: &IggyClient, stream_name: &str) -> Result<Identifier, BridgeError> {
+ let identifier = Identifier::named(stream_name).map_err(BridgeError::Iggy)?;
+ stream_ensured(
+ stream_name,
+ metadata_write_on(client, || client.create_stream(stream_name)).await,
+ )?;
+ Ok(identifier)
+}
+
+/// Whether a `create_stream` left the stream in place: it created it, another caller had, or the
+/// SDK replayed a create that had already landed (`RequestAlreadyApplied`).
+fn stream_ensured<T>(
+ stream_name: &str,
+ created: Result<T, BridgeError>,
+) -> Result<(), BridgeError> {
+ match created {
+ Ok(_created) => {
+ info!("created Iggy stream '{stream_name}'");
+ Ok(())
+ }
+ Err(BridgeError::Iggy(
+ IggyError::StreamNameAlreadyExists(_) | IggyError::RequestAlreadyApplied,
+ )) => {
+ debug!("Iggy stream '{stream_name}' already exists");
+ Ok(())
+ }
+ Err(err) => Err(err),
+ }
+}
+
/// `message_expiry` and whether it was set explicitly, from one `get_topics` row.
///
/// Enough for `DescribeConfigs`. The partition list on `TopicDetails` is not loaded. Only
@@ -706,3 +787,21 @@ async fn load_stream(
}
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn given_a_replayed_stream_create_that_already_landed_should_treat_the_stream_as_present() {
+ assert!(stream_ensured("kafka", Ok(())).is_ok());
+ let exists = IggyError::StreamNameAlreadyExists("kafka".to_string());
+ assert!(stream_ensured::<()>("kafka", Err(BridgeError::Iggy(exists))).is_ok());
+ let replayed = IggyError::RequestAlreadyApplied;
+ assert!(stream_ensured::<()>("kafka", Err(BridgeError::Iggy(replayed))).is_ok());
+ assert!(matches!(
+ stream_ensured::<()>("kafka", Err(BridgeError::Timeout)),
+ Err(BridgeError::Timeout)
+ ));
+ }
+}
diff --git a/gateways/kafka/src/bridge/mod.rs b/gateways/kafka/src/bridge/mod.rs
index 6f9e76ef6..2865d4e1b 100644
--- a/gateways/kafka/src/bridge/mod.rs
+++ b/gateways/kafka/src/bridge/mod.rs
@@ -29,7 +29,8 @@ pub mod topic_map;
pub use config::{DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig, IggyTls, ProduceSessionLimits};
pub use error::BridgeError;
pub use iggy_bridge::{
- IggyBridge, KafkaTopicMetadata, OFFSET_GROUP_PREFIX, OffsetCalls, ProduceTurn, Producer,
- StreamTopicCache, TopicCreationOutcome, TopicLoad, TopicTarget,
+ DEFAULT_TOPIC_PARTITIONS, IggyBridge, KafkaTopicMetadata, MAX_TOPIC_PARTITIONS,
+ OFFSET_GROUP_PREFIX, OffsetCalls, ProduceTurn, Producer, StreamTopicCache,
+ TopicCreationOutcome, TopicLoad, TopicTarget,
};
pub use topic_map::{TopicMapping, TopicOverride};
diff --git a/gateways/kafka/src/bridge/topic_map.rs b/gateways/kafka/src/bridge/topic_map.rs
index 66e2ed808..df294fe10 100644
--- a/gateways/kafka/src/bridge/topic_map.rs
+++ b/gateways/kafka/src/bridge/topic_map.rs
@@ -80,51 +80,64 @@ pub(crate) fn validate_identifier_name(field: &str, value: &str) -> Result<(), B
/// `field` is `"topic mapping key"` or `"kafka_topic"` in current callers, not a generic template -
/// kept as a parameter (like [`validate_identifier_name`]) so the message names the actual site.
pub(crate) fn validate_kafka_topic_name(field: &str, value: &str) -> Result<(), BridgeError> {
+ let Some(violation) = kafka_topic_name_violation(value) else {
+ return Ok(());
+ };
+ let reason = match violation {
+ KafkaTopicNameViolation::Empty => format!("{field} must not be empty"),
+ KafkaTopicNameViolation::Padded => {
+ format!("{field} must not have leading or trailing whitespace")
+ }
+ KafkaTopicNameViolation::TooLong => format!(
+ "{field} is {} bytes, over Kafka's {MAX_KAFKA_TOPIC_NAME_LEN}-byte topic name limit",
+ value.len()
+ ),
+ KafkaTopicNameViolation::DotName => format!("{field} cannot be \".\" or \"..\""),
+ KafkaTopicNameViolation::IllegalByte(illegal) => format!(
+ "{field} contains {:?}, outside Kafka's legal topic-name characters \
+ (letters, digits, '.', '_', '-')",
+ illegal as char
+ ),
+ };
+ Err(BridgeError::InvalidKafkaTopicName {
+ kafka_topic: value.to_string(),
+ reason,
+ })
+}
+
+/// Whether `value` passes [`validate_kafka_topic_name`], without building its error.
+pub(crate) fn is_valid_kafka_topic_name(value: &str) -> bool {
+ kafka_topic_name_violation(value).is_none()
+}
+
+/// The first of Kafka's topic-name rules `value` breaks.
+enum KafkaTopicNameViolation {
+ Empty,
+ Padded,
+ TooLong,
+ DotName,
+ IllegalByte(u8),
+}
+
+fn kafka_topic_name_violation(value: &str) -> Option<KafkaTopicNameViolation> {
if value.is_empty() {
- return Err(BridgeError::InvalidKafkaTopicName {
- kafka_topic: value.to_string(),
- reason: format!("{field} must not be empty"),
- });
+ return Some(KafkaTopicNameViolation::Empty);
}
if value.trim() != value {
- return Err(BridgeError::InvalidKafkaTopicName {
- kafka_topic: value.to_string(),
- reason: format!("{field} must not have leading or trailing whitespace"),
- });
+ return Some(KafkaTopicNameViolation::Padded);
}
if value.len() > MAX_KAFKA_TOPIC_NAME_LEN {
- return Err(BridgeError::InvalidKafkaTopicName {
- kafka_topic: value.to_string(),
- reason: format!(
- "{field} is {} bytes, over Kafka's {MAX_KAFKA_TOPIC_NAME_LEN}-byte topic name limit",
- value.len()
- ),
- });
+ return Some(KafkaTopicNameViolation::TooLong);
}
// Kafka's own `Topic.validate` special-cases these two names as invalid even though every
- // byte in them is otherwise legal (`.` is itself an allowed character) - not a traversal risk
- // here, since Iggy builds on-disk paths from numeric ids rather than the topic name, but the
- // doc above claims this function applies Kafka's own rules, and these are two of them.
+ // byte in them is otherwise legal (`.` is itself an allowed character).
if value == "." || value == ".." {
- return Err(BridgeError::InvalidKafkaTopicName {
- kafka_topic: value.to_string(),
- reason: format!("{field} cannot be \".\" or \"..\""),
- });
+ return Some(KafkaTopicNameViolation::DotName);
}
- if let Some(illegal) = value
+ value
.bytes()
.find(|b| !(b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'-')))
- {
- return Err(BridgeError::InvalidKafkaTopicName {
- kafka_topic: value.to_string(),
- reason: format!(
- "{field} contains {:?}, outside Kafka's legal topic-name characters \
- (letters, digits, '.', '_', '-')",
- illegal as char
- ),
- });
- }
- Ok(())
+ .map(KafkaTopicNameViolation::IllegalByte)
}
/// Explicit Kafka-topic → Iggy stream/topic override. Absent entries fall back to
@@ -271,6 +284,19 @@ impl TopicMapping {
)
}
+ /// Whether `kafka_topic` is no override's key yet resolves to the Iggy stream and topic an
+ /// override targets, so writing through it would land in that override's data.
+ #[must_use]
+ pub fn aliases_an_override(&self, kafka_topic: &str) -> bool {
+ if self.topics.contains_key(kafka_topic) {
+ return false;
+ }
+ let target = self.resolve(kafka_topic);
+ self.topics
+ .values()
+ .any(|over| (over.stream.as_str(), over.topic.as_str()) == target)
+ }
+
/// Parses a `TopicMapping` from a TOML document.
///
/// # Errors
@@ -569,4 +595,62 @@ mod tests {
fn validate_kafka_topic_name_accepts_every_legal_character_class() {
validate_kafka_topic_name("kafka_topic", "Order.Events_2024-v2").unwrap();
}
+
+ #[test]
+ fn is_valid_kafka_topic_name_agrees_with_validate_kafka_topic_name() {
+ let too_long = "a".repeat(MAX_KAFKA_TOPIC_NAME_LEN + 1);
+ for name in [
+ "orders",
+ "Order.Events_2024-v2",
+ "",
+ " orders",
+ ".",
+ "..",
+ "has a space",
+ "orders/2024",
+ too_long.as_str(),
+ ] {
+ assert_eq!(
+ is_valid_kafka_topic_name(name),
+ validate_kafka_topic_name("kafka_topic", name).is_ok(),
+ "{name:?}"
+ );
+ }
+ }
+
+ #[test]
+ fn given_an_override_key_or_an_unrelated_name_should_not_report_an_alias() {
+ let mut topics = HashMap::new();
+ topics.insert(
+ "orders".to_string(),
+ TopicOverride {
+ stream: "billing".to_string(),
+ topic: "orders_v2".to_string(),
+ },
+ );
+ let mapping = mapping_with("kafka", topics);
+ assert!(!mapping.aliases_an_override("orders"));
+ assert!(!mapping.aliases_an_override("orders_v2"));
+ assert!(!mapping.aliases_an_override("payments"));
+ }
+
+ /// `new` refuses this shape, so the mapping is built by hand: the predicate must not rely on
+ /// that check to keep a name from writing into an override's topic.
+ #[test]
+ fn given_a_name_resolving_onto_an_override_target_should_report_an_alias() {
+ let mut topics = HashMap::new();
+ topics.insert(
+ "orders".to_string(),
+ TopicOverride {
+ stream: "kafka".to_string(),
+ topic: "orders_v2".to_string(),
+ },
+ );
+ let mapping = TopicMapping {
+ default_stream: "kafka".to_string(),
+ topics,
+ };
+ assert!(mapping.aliases_an_override("orders_v2"));
+ assert!(!mapping.aliases_an_override("orders"));
+ }
}
diff --git a/gateways/kafka/src/main.rs b/gateways/kafka/src/main.rs
index 0d755d3c8..3a11b603a 100644
--- a/gateways/kafka/src/main.rs
+++ b/gateways/kafka/src/main.rs
@@ -25,11 +25,14 @@ use tokio::sync::{Semaphore, broadcast};
use tracing::{info, warn};
use iggy_gateway_kafka::auth::IggyAuthenticator;
-use iggy_gateway_kafka::bridge::{IggyBridge, IggyBridgeConfig, IggyTls};
+use iggy_gateway_kafka::bridge::{
+ DEFAULT_TOPIC_PARTITIONS, IggyBridge, IggyBridgeConfig, IggyTls, MAX_TOPIC_PARTITIONS,
+};
use iggy_gateway_kafka::env::parse_bool;
use iggy_gateway_kafka::listener::{
ListenerConfig, SecurityProtocol, apply_advertised_listeners, parse_listeners,
};
+use iggy_gateway_kafka::protocol::auto_topic_creation::AutoTopicCreationConfig;
use iggy_gateway_kafka::server::{bind_listeners, init_tracing};
use iggy_gateway_kafka::tls::{ClientAuth, TlsConfig};
use iggy_gateway_kafka::{GatewayConfig, KafkaGateway};
@@ -141,6 +144,8 @@ const KNOWN_KAFKA_ENV_VARS: &[&str] = &[
"IGGY_KAFKA_SASL_ENABLED",
"IGGY_KAFKA_PRE_AUTH_TIMEOUT_SECS",
"IGGY_KAFKA_MAX_CONCURRENT_AUTHENTICATIONS",
+ "IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED",
+ "IGGY_KAFKA_AUTO_CREATE_TOPICS_PARTITIONS",
];
/// The variables of the single-listener form, which `IGGY_KAFKA_LISTENERS` replaces as a whole.
@@ -312,6 +317,13 @@ fn load_config() -> Result<GatewayConfig, String> {
Iggy cluster needs its own value, or they hand out colliding producer ids"
);
}
+ config.auto_topic_creation = load_auto_topic_creation(bridge_enabled()?)?;
+ if config.auto_topic_creation.enabled {
+ info!(
+ partitions = config.auto_topic_creation.partitions,
+ "auto topic creation is on: a Metadata lookup of a missing topic creates it"
+ );
+ }
reject_unread_iggy_tls(
config.sasl_enabled(),
bridge_enabled()?,
@@ -530,6 +542,47 @@ where
Ok(value)
}
+/// `IGGY_KAFKA_AUTO_CREATE_TOPICS_*`. A partition count with creation off, or creation without a
+/// bridge to create through, is refused rather than silently ignored.
+fn load_auto_topic_creation(bridge_enabled: bool) -> Result<AutoTopicCreationConfig, String> {
+ let enabled = match env_var("IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED") {
+ Some(raw) => parse_bool("IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED", &raw)?,
+ None => false,
+ };
+ let partitions = env_var("IGGY_KAFKA_AUTO_CREATE_TOPICS_PARTITIONS");
+ if !enabled {
+ if partitions.is_some() {
+ return Err("IGGY_KAFKA_AUTO_CREATE_TOPICS_PARTITIONS is set but \
+ IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED is not true"
+ .to_string());
+ }
+ return Ok(AutoTopicCreationConfig::default());
+ }
+ if !bridge_enabled {
+ return Err(
+ "IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED needs IGGY_KAFKA_BRIDGE_ENABLED=true"
+ .to_string(),
+ );
+ }
+ let partitions = match partitions {
+ Some(raw) => {
+ let count: u32 = parse_positive("IGGY_KAFKA_AUTO_CREATE_TOPICS_PARTITIONS", &raw)?;
+ if count > MAX_TOPIC_PARTITIONS {
+ return Err(format!(
+ "IGGY_KAFKA_AUTO_CREATE_TOPICS_PARTITIONS {count} exceeds Iggy's per-topic \
+ maximum {MAX_TOPIC_PARTITIONS}"
+ ));
+ }
+ count
+ }
+ None => DEFAULT_TOPIC_PARTITIONS,
+ };
+ Ok(AutoTopicCreationConfig {
+ enabled,
+ partitions,
+ })
+}
+
/// Wait for Ctrl-C (SIGINT) or, on Unix, SIGTERM (`docker stop`).
async fn shutdown_signal() {
let ctrl_c = async {
@@ -694,6 +747,59 @@ mod tests {
}
}
+ #[test]
+ #[serial]
+ fn given_no_auto_create_vars_when_loading_config_should_leave_auto_creation_off() {
+ let config = load_with(&[]).expect("defaults load");
+ assert!(!config.auto_topic_creation.enabled);
+ }
+
+ #[test]
+ #[serial]
+ fn given_auto_creation_with_a_bridge_when_loading_config_should_read_the_partition_count() {
+ let config = load_with(&[
+ ("IGGY_KAFKA_BRIDGE_ENABLED", "true"),
+ ("IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED", "true"),
+ ("IGGY_KAFKA_AUTO_CREATE_TOPICS_PARTITIONS", "6"),
+ ])
+ .expect("a valid auto-creation config loads");
+ assert!(config.auto_topic_creation.enabled);
+ assert_eq!(config.auto_topic_creation.partitions, 6);
+ }
+
+ #[test]
+ #[serial]
+ fn given_an_invalid_auto_create_setting_when_loading_config_should_refuse_and_name_it() {
+ let cases: [(&[(&str, &str)], &str); 4] = [
+ (
+ &[("IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED", "true")],
+ "IGGY_KAFKA_BRIDGE_ENABLED",
+ ),
+ (
+ &[("IGGY_KAFKA_AUTO_CREATE_TOPICS_PARTITIONS", "3")],
+ "IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED",
+ ),
+ (
+ &[
+ ("IGGY_KAFKA_BRIDGE_ENABLED", "true"),
+ ("IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED", "true"),
+ ("IGGY_KAFKA_AUTO_CREATE_TOPICS_PARTITIONS", "1001"),
+ ],
+ "IGGY_KAFKA_AUTO_CREATE_TOPICS_PARTITIONS",
+ ),
+ (
+ &[("IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED", "yes")],
+ "IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED",
+ ),
+ ];
+ for (vars, named) in cases {
+ let error = load_with(vars)
+ .err()
+ .unwrap_or_else(|| panic!("{vars:?} must be refused"));
+ assert!(error.contains(named), "{vars:?}: {error}");
+ }
+ }
+
/// `#[serial]` for the same reason as the test above: it mutates process-wide env state.
#[test]
#[serial]
diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs
index 9a003d949..3aa432f2e 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -35,6 +35,7 @@ use crate::protocol::acl::{
self, AclBinding, AclFilter, PrincipalPermissions, encode_describe_acls_error_response,
encode_describe_acls_response,
};
+use crate::protocol::auto_topic_creation::{AutoTopicCreationConfig, AutoTopicCreator};
use crate::protocol::bounds_guard::{
validate_describe_acls_shape, validate_sasl_authenticate_shape, validate_sasl_handshake_shape,
};
@@ -338,6 +339,8 @@ pub struct GatewayState {
pub(crate) unproven_empty: Mutex<fetch::Spells>,
/// Topic probes that every Fetch shares.
pub(crate) probe_board: ProbeBoard,
+ /// Present only while `IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED` is on.
+ pub(crate) auto_topic_creation: Option<AutoTopicCreator>,
}
/// Large Produce requests that decode and send at once. More at once only contend for the
@@ -372,6 +375,38 @@ impl GatewayState {
loading: Mutex::default(),
unproven_empty: Mutex::default(),
probe_board: ProbeBoard::default(),
+ auto_topic_creation: None,
+ }
+ }
+
+ /// Turns on Kafka's `auto.create.topics.enable` behaviour for `Metadata` when `config` says
+ /// so.
+ ///
+ /// # Errors
+ ///
+ /// [`KafkaProtocolError::InvalidConfig`](crate::error::KafkaProtocolError::InvalidConfig)
+ /// when creation is on and `config.partitions` is 0 or over Iggy's per-topic cap.
+ pub fn with_auto_topic_creation(mut self, config: AutoTopicCreationConfig) -> Result<Self> {
+ self.auto_topic_creation = if config.enabled {
+ Some(AutoTopicCreator::new(config.partitions)?)
+ } else {
+ None
+ };
+ Ok(self)
+ }
+
+ /// Cancels the auto topic creations still running and waits up to `deadline` for them to
+ /// end. An embedder that builds its own `GatewayState` calls it before closing the bridge.
+ pub async fn shutdown_auto_topic_creation(&self, deadline: Duration) {
+ let Some(creator) = &self.auto_topic_creation else {
+ return;
+ };
+ if creator.shutdown(deadline).await
+ && let Some(bridge) = &self.bridge
+ {
+ // A cancelled create can keep its client busy, and the bridge's sign-out would queue
+ // behind it.
+ bridge.discard_auto_create_client();
}
}
diff --git a/gateways/kafka/src/protocol/auto_topic_creation.rs b/gateways/kafka/src/protocol/auto_topic_creation.rs
new file mode 100644
index 000000000..d742db87f
--- /dev/null
+++ b/gateways/kafka/src/protocol/auto_topic_creation.rs
@@ -0,0 +1,503 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka's `auto.create.topics.enable`: a `Metadata` lookup of a topic that does not exist starts
+//! creating it.
+//!
+//! As in Kafka (`DefaultAutoTopicCreationManager`), the lookup that starts a creation is answered
+//! `UNKNOWN_TOPIC_OR_PARTITION` at once, and the client finds the topic on a later refresh. A
+//! topic already being created is not started again.
+
+use std::collections::{HashMap, HashSet};
+use std::sync::{Arc, Mutex, PoisonError};
+use std::time::Duration;
+
+use iggy::prelude::IggyError;
+use tokio::time::Instant;
+use tokio_util::sync::CancellationToken;
+use tokio_util::task::TaskTracker;
+use tracing::{debug, info, warn};
+
+use crate::bridge::{
+ BridgeError, DEFAULT_TOPIC_PARTITIONS, IggyBridge, MAX_TOPIC_PARTITIONS, TopicCreationOutcome,
+};
+use crate::error::KafkaProtocolError;
+
+/// Creations one gateway runs at once.
+///
+/// They share one Iggy client, so a client naming thousands of missing topics in one `Metadata`
+/// must not queue them all. A name not started now is started by a later refresh.
+pub(crate) const MAX_IN_FLIGHT_CREATIONS: usize = 100;
+
+/// How long a topic whose creation failed or timed out is not tried again, unless
+/// [`REFUSED_CREATION_BACKOFF`] applies.
+///
+/// A producer refreshes `Metadata` every `retry.backoff.ms` (100 ms by default) while it waits, so
+/// without this a creation Iggy refuses, for example at `partitions_max`, would run ten times a
+/// second.
+pub(crate) const FAILED_CREATION_BACKOFF: Duration = Duration::from_secs(1);
+
+/// How long a topic whose creation Iggy refused for the bridge user's permissions is not tried
+/// again.
+///
+/// Iggy decides that refusal only when it applies the create, after journaling and replicating
+/// it, so each attempt costs a metadata write. Only an operator's grant changes the answer, so a
+/// minute between attempts delays little next to a write a second per name.
+const REFUSED_CREATION_BACKOFF: Duration = Duration::from_secs(60);
+
+/// How long shutdown waits for cancelled creations to end. A cancelled creation ends at its next
+/// poll, so this only covers a runtime too busy to poll it.
+pub const SHUTDOWN_WAIT: Duration = Duration::from_secs(1);
+
+/// How long the failures of one name log at debug after one logged at warn. A name Iggy refuses
+/// is retried every [`FAILED_CREATION_BACKOFF`] for as long as clients name it.
+const REPEAT_WARNING_INTERVAL: Duration = Duration::from_secs(60);
+
+/// Names [`FailureWarnings`] remembers. A failure past it warns without being remembered.
+const MAX_WARNED_NAMES: usize = 1024;
+
+/// `IGGY_KAFKA_AUTO_CREATE_TOPICS_ENABLED` (Kafka's `auto.create.topics.enable`) and
+/// `IGGY_KAFKA_AUTO_CREATE_TOPICS_PARTITIONS`.
+///
+/// Off by default, unlike Kafka: a gateway that started creating topics on its own after an
+/// upgrade would surprise operators who rely on `CreateTopics` being the only way in.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct AutoTopicCreationConfig {
+ pub enabled: bool,
+ /// From 1 to [`MAX_TOPIC_PARTITIONS`].
+ pub partitions: u32,
+}
+
+impl Default for AutoTopicCreationConfig {
+ fn default() -> Self {
+ Self {
+ enabled: false,
+ partitions: DEFAULT_TOPIC_PARTITIONS,
+ }
+ }
+}
+
+/// The topics Kafka creates for its own coordinators (`Topic.GROUP_METADATA_TOPIC_NAME`,
+/// `TRANSACTION_STATE_TOPIC_NAME`, `SHARE_GROUP_STATE_TOPIC_NAME`).
+const KAFKA_INTERNAL_TOPICS: [&str; 3] = [
+ "__consumer_offsets",
+ "__transaction_state",
+ "__share_group_state",
+];
+
+/// Whether `kafka_topic` is one of Kafka's internal topics. This gateway keeps none of them in
+/// Iggy, so none is auto-created. Other `__` names, such as Debezium's heartbeat topics, are
+/// ordinary topics.
+#[must_use]
+pub(crate) fn is_internal_topic_name(kafka_topic: &str) -> bool {
+ KAFKA_INTERNAL_TOPICS.contains(&kafka_topic)
+}
+
+/// Topics this gateway is creating on behalf of `Metadata` lookups.
+#[derive(Debug)]
+pub(crate) struct AutoTopicCreator {
+ partitions: u32,
+ in_flight: Arc<Mutex<HashSet<Arc<str>>>>,
+ warnings: Arc<FailureWarnings>,
+ tasks: TaskTracker,
+ cancel: CancellationToken,
+}
+
+impl AutoTopicCreator {
+ /// # Errors
+ ///
+ /// [`KafkaProtocolError::InvalidConfig`] unless `partitions` is from 1 to
+ /// [`MAX_TOPIC_PARTITIONS`].
+ pub(crate) fn new(partitions: u32) -> Result<Self, KafkaProtocolError> {
+ if partitions == 0 || partitions > MAX_TOPIC_PARTITIONS {
+ return Err(KafkaProtocolError::InvalidConfig(format!(
+ "auto topic creation partitions must be from 1 to {MAX_TOPIC_PARTITIONS}, got \
+ {partitions}"
+ )));
+ }
+ Ok(Self {
+ partitions,
+ in_flight: Arc::new(Mutex::new(HashSet::new())),
+ warnings: Arc::default(),
+ tasks: TaskTracker::new(),
+ cancel: CancellationToken::new(),
+ })
+ }
+
+ /// Starts creating `kafka_topic` in the background, unless it is already being created, it
+ /// failed less than [`FAILED_CREATION_BACKOFF`] (or [`REFUSED_CREATION_BACKOFF`]) ago,
+ /// [`MAX_IN_FLIGHT_CREATIONS`] are running, or shutdown began. Returns whether it started one.
+ /// The caller answers `UNKNOWN_TOPIC_OR_PARTITION` either way.
+ pub(crate) fn start(&self, bridge: &Arc<IggyBridge>, kafka_topic: &str) -> bool {
+ if self.cancel.is_cancelled() {
+ return false;
+ }
+ let Some(claim) = InFlight::claim(&self.in_flight, kafka_topic) else {
+ return false;
+ };
+ let bridge = Arc::clone(bridge);
+ let partitions = self.partitions;
+ let cancel = self.cancel.clone();
+ let warnings = Arc::clone(&self.warnings);
+ self.tasks.spawn(async move {
+ tokio::select! {
+ () = cancel.cancelled() => {}
+ () = create(&bridge, &claim.name, partitions, &warnings) => {}
+ }
+ });
+ true
+ }
+
+ /// Cancels the creations still running and waits up to `deadline` for them to end. Returns
+ /// whether any was running. A cancelled creation may still land in Iggy, and the next lookup
+ /// then finds the topic.
+ pub(crate) async fn shutdown(&self, deadline: Duration) -> bool {
+ let cancelled = !self.tasks.is_empty();
+ self.cancel.cancel();
+ self.tasks.close();
+ if tokio::time::timeout(deadline, self.tasks.wait())
+ .await
+ .is_err()
+ {
+ warn!(
+ ?deadline,
+ "auto topic creations did not end before shutdown went on"
+ );
+ }
+ cancelled
+ }
+
+ #[cfg(test)]
+ pub(crate) fn in_flight(&self) -> usize {
+ self.in_flight
+ .lock()
+ .unwrap_or_else(PoisonError::into_inner)
+ .len()
+ }
+}
+
+/// Creates `kafka_topic`, then, if that failed, holds it for as long as [`backoff_after`] says.
+async fn create(
+ bridge: &IggyBridge,
+ kafka_topic: &str,
+ partitions: u32,
+ warnings: &FailureWarnings,
+) {
+ let hold = match bridge
+ .auto_create_kafka_topic(kafka_topic, partitions)
+ .await
+ {
+ Ok(TopicCreationOutcome::Created) => {
+ warnings.forget(kafka_topic);
+ info!(
+ kafka_topic,
+ partitions, "auto-created a topic a Metadata request named"
+ );
+ None
+ }
+ Ok(TopicCreationOutcome::AlreadyExists)
+ | Err(BridgeError::Iggy(IggyError::RequestAlreadyApplied)) => {
+ warnings.forget(kafka_topic);
+ None
+ }
+ // The create may still land. Held like a failure only to space out retries of this name:
+ // a retry that races it is harmless, since Iggy refuses a name already taken.
+ Err(
+ error @ (BridgeError::Timeout | BridgeError::Iggy(IggyError::OperationUncertain(_))),
+ ) => {
+ debug!(kafka_topic, %error, "auto topic creation outcome unknown, it may still land");
+ Some(backoff_after(&error))
+ }
+ Err(error) if warnings.should_warn(kafka_topic, Instant::now()) => {
+ warn!(kafka_topic, %error, "auto topic creation failed");
+ Some(backoff_after(&error))
+ }
+ Err(error) => {
+ debug!(kafka_topic, %error, "auto topic creation failed again");
+ Some(backoff_after(&error))
+ }
+ };
+ if let Some(hold) = hold {
+ tokio::time::sleep(hold).await;
+ }
+}
+
+/// How long a name is held after its creation failed with `error`.
+const fn backoff_after(error: &BridgeError) -> Duration {
+ if matches!(error, BridgeError::Iggy(IggyError::Unauthorized)) {
+ REFUSED_CREATION_BACKOFF
+ } else {
+ FAILED_CREATION_BACKOFF
+ }
+}
+
+/// When each name whose creation failed last warned, so a name Iggy keeps refusing warns once per
+/// [`REPEAT_WARNING_INTERVAL`], not once per retry.
+#[derive(Debug, Default)]
+struct FailureWarnings(Mutex<HashMap<Arc<str>, Instant>>);
+
+impl FailureWarnings {
+ /// Whether a failure of `kafka_topic` at `now` logs at warn, and if so remembers that it did.
+ fn should_warn(&self, kafka_topic: &str, now: Instant) -> bool {
+ let recent = |warned_at: &Instant| now.duration_since(*warned_at) < REPEAT_WARNING_INTERVAL;
+ let mut warned = self.0.lock().unwrap_or_else(PoisonError::into_inner);
+ if warned.get(kafka_topic).is_some_and(recent) {
+ return false;
+ }
+ if warned.len() >= MAX_WARNED_NAMES {
+ warned.retain(|_, warned_at| recent(warned_at));
+ }
+ if warned.len() < MAX_WARNED_NAMES {
+ warned.insert(Arc::from(kafka_topic), now);
+ }
+ true
+ }
+
+ fn forget(&self, kafka_topic: &str) {
+ self.0
+ .lock()
+ .unwrap_or_else(PoisonError::into_inner)
+ .remove(kafka_topic);
+ }
+}
+
+/// One name in the in-flight set, released on drop so a panicking or cancelled creation frees
+/// its name and slot too.
+struct InFlight {
+ names: Arc<Mutex<HashSet<Arc<str>>>>,
+ name: Arc<str>,
+}
+
+impl InFlight {
+ fn claim(names: &Arc<Mutex<HashSet<Arc<str>>>>, kafka_topic: &str) -> Option<Self> {
+ let mut held = names.lock().unwrap_or_else(PoisonError::into_inner);
+ if held.len() >= MAX_IN_FLIGHT_CREATIONS || held.contains(kafka_topic) {
+ return None;
+ }
+ let name: Arc<str> = Arc::from(kafka_topic);
+ held.insert(Arc::clone(&name));
+ drop(held);
+ Some(Self {
+ names: Arc::clone(names),
+ name,
+ })
+ }
+}
+
+impl Drop for InFlight {
+ fn drop(&mut self) {
+ self.names
+ .lock()
+ .unwrap_or_else(PoisonError::into_inner)
+ .remove(&self.name);
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::group::{GroupCoordinator, GroupCoordinatorConfig};
+ use crate::protocol::api::GatewayState;
+
+ /// The offline bridge fails every call at once, so each creation lands in its failure backoff
+ /// and stays in flight for exactly that long.
+ #[tokio::test(start_paused = true)]
+ async fn given_a_topic_being_created_when_it_is_named_again_should_not_start_it_twice() {
+ let creator = AutoTopicCreator::new(1).expect("a valid partition count");
+ let bridge = Arc::new(IggyBridge::offline());
+
+ assert!(creator.start(&bridge, "orders"));
+ assert!(!creator.start(&bridge, "orders"));
+ tokio::task::yield_now().await;
+
+ assert_eq!(creator.in_flight(), 1);
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn given_the_in_flight_cap_when_more_topics_are_named_should_start_only_up_to_it() {
+ let creator = AutoTopicCreator::new(1).expect("a valid partition count");
+ let bridge = Arc::new(IggyBridge::offline());
+
+ for index in 0..MAX_IN_FLIGHT_CREATIONS {
+ assert!(creator.start(&bridge, &format!("topic-{index}")));
+ }
+
+ assert!(!creator.start(&bridge, "one-too-many"));
+
+ assert_eq!(creator.in_flight(), MAX_IN_FLIGHT_CREATIONS);
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn given_a_failed_creation_when_the_backoff_passes_should_allow_another_attempt() {
+ let creator = AutoTopicCreator::new(1).expect("a valid partition count");
+ let bridge = Arc::new(IggyBridge::offline());
+
+ assert!(creator.start(&bridge, "orders"));
+ tokio::task::yield_now().await;
+ tokio::time::sleep(FAILED_CREATION_BACKOFF / 2).await;
+ assert_eq!(creator.in_flight(), 1, "still backing off");
+
+ tokio::time::sleep(FAILED_CREATION_BACKOFF).await;
+ tokio::task::yield_now().await;
+ assert_eq!(creator.in_flight(), 0);
+ assert!(creator.start(&bridge, "orders"));
+ }
+
+ /// The creation is in its backoff when shutdown begins, so only cancelling it ends it sooner.
+ #[tokio::test(start_paused = true)]
+ async fn given_a_running_creation_when_shutting_down_should_cancel_it_and_wait_for_it() {
+ let creator = AutoTopicCreator::new(1).expect("a valid partition count");
+ let bridge = Arc::new(IggyBridge::offline());
+ assert!(creator.start(&bridge, "orders"));
+ tokio::task::yield_now().await;
+
+ let began = Instant::now();
+ creator.shutdown(FAILED_CREATION_BACKOFF * 10).await;
+
+ assert!(
+ began.elapsed() < FAILED_CREATION_BACKOFF,
+ "cancelled, not waited out"
+ );
+ assert_eq!(
+ creator.in_flight(),
+ 0,
+ "the task ended and released its name"
+ );
+ assert!(
+ !creator.start(&bridge, "orders"),
+ "nothing starts after shutdown"
+ );
+ }
+
+ #[test]
+ fn given_a_name_that_keeps_failing_should_warn_once_per_interval() {
+ let warnings = FailureWarnings::default();
+ let start = Instant::now();
+
+ assert!(warnings.should_warn("orders", start));
+ assert!(!warnings.should_warn("orders", start + FAILED_CREATION_BACKOFF));
+ assert!(
+ warnings.should_warn("payments", start + FAILED_CREATION_BACKOFF),
+ "each name warns on its own"
+ );
+ assert!(warnings.should_warn("orders", start + REPEAT_WARNING_INTERVAL));
+ }
+
+ #[test]
+ fn given_a_name_that_was_created_when_it_fails_again_should_warn() {
+ let warnings = FailureWarnings::default();
+ let start = Instant::now();
+ assert!(warnings.should_warn("orders", start));
+
+ warnings.forget("orders");
+
+ assert!(warnings.should_warn("orders", start + FAILED_CREATION_BACKOFF));
+ }
+
+ #[test]
+ fn given_every_remembered_name_when_one_more_fails_should_still_warn_and_stay_bounded() {
+ let warnings = FailureWarnings::default();
+ let start = Instant::now();
+ for index in 0..MAX_WARNED_NAMES {
+ assert!(warnings.should_warn(&format!("topic-{index}"), start));
+ }
+
+ assert!(warnings.should_warn("one-too-many", start));
+ assert!(
+ warnings.should_warn("one-too-many", start),
+ "not remembered while every remembered name is recent"
+ );
+ let later = start + REPEAT_WARNING_INTERVAL;
+ assert!(warnings.should_warn("one-too-many", later));
+ assert!(
+ !warnings.should_warn("one-too-many", later),
+ "the stale names were pruned to make room"
+ );
+ assert!(
+ warnings
+ .0
+ .lock()
+ .unwrap_or_else(PoisonError::into_inner)
+ .len()
+ <= MAX_WARNED_NAMES
+ );
+ }
+
+ fn gateway_with(bridge: &Arc<IggyBridge>) -> GatewayState {
+ GatewayState::new(
+ Some(Arc::clone(bridge)),
+ 1024,
+ 0,
+ GroupCoordinator::new(GroupCoordinatorConfig::default(), CancellationToken::new()),
+ )
+ .with_auto_topic_creation(AutoTopicCreationConfig {
+ enabled: true,
+ partitions: 1,
+ })
+ .expect("a valid partition count")
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn given_a_running_creation_when_the_gateway_shuts_it_down_should_discard_its_client() {
+ let bridge = Arc::new(IggyBridge::offline());
+ let state = gateway_with(&bridge);
+ let creator = state.auto_topic_creation.as_ref().expect("creation is on");
+ assert!(creator.start(&bridge, "orders"));
+ tokio::task::yield_now().await;
+
+ state.shutdown_auto_topic_creation(SHUTDOWN_WAIT).await;
+
+ assert!(!bridge.holds_auto_create_client().await);
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn given_no_running_creation_when_the_gateway_shuts_it_down_should_keep_the_client_to_sign_out()
+ {
+ let bridge = Arc::new(IggyBridge::offline());
+ let state = gateway_with(&bridge);
+
+ state.shutdown_auto_topic_creation(SHUTDOWN_WAIT).await;
+
+ assert!(bridge.holds_auto_create_client().await);
+ }
+
+ #[test]
+ fn given_a_partition_count_outside_iggys_range_should_refuse_to_build() {
+ assert!(AutoTopicCreator::new(0).is_err());
+ assert!(AutoTopicCreator::new(MAX_TOPIC_PARTITIONS + 1).is_err());
+ assert!(AutoTopicCreator::new(MAX_TOPIC_PARTITIONS).is_ok());
+ }
+
+ #[test]
+ fn given_a_name_kafka_reserves_should_report_it_as_internal() {
+ for name in [
+ "__consumer_offsets",
+ "__transaction_state",
+ "__share_group_state",
+ ] {
+ assert!(is_internal_topic_name(name), "{name}");
+ }
+ assert!(!is_internal_topic_name("_orders"));
+ assert!(!is_internal_topic_name("orders"));
+ }
+
+ #[test]
+ fn given_a_double_underscore_name_kafka_does_not_reserve_should_allow_creating_it() {
+ assert!(!is_internal_topic_name("__debezium-heartbeat.x"));
+ assert!(!is_internal_topic_name("__consumer_offsets_copy"));
+ }
+}
diff --git a/gateways/kafka/src/protocol/handlers/create_topics.rs b/gateways/kafka/src/protocol/handlers/create_topics.rs
index 83ce99395..e78f0e909 100644
--- a/gateways/kafka/src/protocol/handlers/create_topics.rs
+++ b/gateways/kafka/src/protocol/handlers/create_topics.rs
@@ -28,7 +28,9 @@ use kafka_protocol::protocol::StrBytes;
use tokio::time::Instant;
-use crate::bridge::{BridgeError, IggyBridge, TopicCreationOutcome};
+use crate::bridge::{
+ BridgeError, DEFAULT_TOPIC_PARTITIONS, IggyBridge, MAX_TOPIC_PARTITIONS, TopicCreationOutcome,
+};
use crate::error::Result;
use crate::protocol::api::{
API_KEY_CREATE_TOPICS, ApiVersionRange, ERROR_INVALID_CONFIG, ERROR_INVALID_PARTITIONS,
@@ -51,20 +53,6 @@ pub const RANGE: ApiVersionRange = ApiVersionRange {
max_version: 5,
};
-/// KIP-464 `num_partitions = -1` with no manual assignment: the count this bridge creates.
-///
-/// Matches real Kafka's own out-of-box `num.partitions=1` broker default. Not read from any
-/// bridge config - there is no such config surface today.
-const DEFAULT_PARTITION_COUNT: u32 = 1;
-
-/// Iggy's own server-side per-topic partition cap (`rewrite.rs`'s `IggyError::TooManyPartitions`).
-///
-/// Not exported by `iggy::prelude`, so mirrored here as a named constant rather than a bare
-/// literal repeated at every call site. Enforcing it locally means `validate_only` answers the
-/// same rejection the real path would eventually get from the bridge, instead of reporting
-/// `NONE` for a partition count the real path can never actually create.
-const MAX_PARTITIONS_COUNT: u32 = 1000;
-
pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
let Some(bridge) = &state.bridge else {
return handle_versioned_request(
@@ -408,7 +396,7 @@ fn validate_create_topic_shape(topic: &CreatableTopic) -> core::result::Result<u
{
return Err(ERROR_INVALID_REPLICA_ASSIGNMENT);
}
- return Ok(u32::try_from(topic.assignments.len()).unwrap_or(DEFAULT_PARTITION_COUNT));
+ return Ok(u32::try_from(topic.assignments.len()).unwrap_or(DEFAULT_TOPIC_PARTITIONS));
}
// Checked ahead of (and separately from) the partition-count check below: that check's own
@@ -418,7 +406,7 @@ fn validate_create_topic_shape(topic: &CreatableTopic) -> core::result::Result<u
// is the same code the real (non-`validate_only`) path eventually gets back from the bridge
// once a count this large actually reaches `create_kafka_topic`, so `validate_only` now
// answers what the real create would.
- if u32::try_from(topic.num_partitions).is_ok_and(|count| count > MAX_PARTITIONS_COUNT) {
+ if u32::try_from(topic.num_partitions).is_ok_and(|count| count > MAX_TOPIC_PARTITIONS) {
return Err(ERROR_INVALID_REQUEST);
}
@@ -437,9 +425,9 @@ fn validate_create_topic_shape(topic: &CreatableTopic) -> core::result::Result<u
}
let partition_count = if topic.num_partitions > 0 {
- u32::try_from(topic.num_partitions).unwrap_or(DEFAULT_PARTITION_COUNT)
+ u32::try_from(topic.num_partitions).unwrap_or(DEFAULT_TOPIC_PARTITIONS)
} else {
- DEFAULT_PARTITION_COUNT
+ DEFAULT_TOPIC_PARTITIONS
};
Ok(partition_count)
}
@@ -574,11 +562,11 @@ mod tests {
fn broker_default_sentinels_with_no_assignments_resolve_to_the_default_count() {
assert_eq!(
validate_create_topic_shape(&creatable_topic(-1, -1)),
- Ok(DEFAULT_PARTITION_COUNT)
+ Ok(DEFAULT_TOPIC_PARTITIONS)
);
assert_eq!(
validate_create_topic_shape(&creatable_topic(-1, 1)),
- Ok(DEFAULT_PARTITION_COUNT)
+ Ok(DEFAULT_TOPIC_PARTITIONS)
);
assert_eq!(validate_create_topic_shape(&creatable_topic(4, -1)), Ok(4));
}
@@ -660,10 +648,10 @@ mod tests {
#[test]
fn partition_count_at_the_cap_is_accepted() {
- let topic = creatable_topic(i32::try_from(MAX_PARTITIONS_COUNT).unwrap(), 1);
+ let topic = creatable_topic(i32::try_from(MAX_TOPIC_PARTITIONS).unwrap(), 1);
assert_eq!(
validate_create_topic_shape(&topic),
- Ok(MAX_PARTITIONS_COUNT)
+ Ok(MAX_TOPIC_PARTITIONS)
);
}
@@ -673,7 +661,7 @@ mod tests {
// condition. This is the same code the real (non-validate_only) path eventually gets
// back from the bridge once IggyError::TooManyPartitions reaches it
// (`bridge/error.rs::too_many_partitions_maps_to_invalid_request_not_invalid_partitions`).
- let topic = creatable_topic(i32::try_from(MAX_PARTITIONS_COUNT).unwrap() + 1, 1);
+ let topic = creatable_topic(i32::try_from(MAX_TOPIC_PARTITIONS).unwrap() + 1, 1);
assert_eq!(
validate_create_topic_shape(&topic),
Err(ERROR_INVALID_REQUEST)
diff --git a/gateways/kafka/src/protocol/handlers/metadata.rs b/gateways/kafka/src/protocol/handlers/metadata.rs
index 2c41fc29d..ab2ad9dba 100644
--- a/gateways/kafka/src/protocol/handlers/metadata.rs
+++ b/gateways/kafka/src/protocol/handlers/metadata.rs
@@ -17,6 +17,8 @@
//! Metadata (API key 3).
+use std::sync::Arc;
+
use bytes::Bytes;
use kafka_protocol::messages::metadata_response::{
MetadataResponseBroker, MetadataResponsePartition, MetadataResponseTopic,
@@ -24,13 +26,17 @@ use kafka_protocol::messages::metadata_response::{
use kafka_protocol::messages::{BrokerId, MetadataRequest, MetadataResponse, TopicName};
use kafka_protocol::protocol::{Encodable, StrBytes};
+use crate::auth::AuthenticatedPrincipal;
use crate::bridge::IggyBridge;
+use crate::bridge::topic_map::is_valid_kafka_topic_name;
use crate::error::{KafkaProtocolError, Result};
use crate::protocol::api::{
- API_KEY_METADATA, ApiVersionRange, BrokerAdvertise, ConnectionState, ERROR_NONE,
- ERROR_REQUEST_TIMED_OUT, ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome,
- REQUEST_DEADLINE, is_supported_version, supported_max_version,
+ API_KEY_METADATA, ApiVersionRange, BrokerAdvertise, ConnectionState,
+ ERROR_INVALID_TOPIC_EXCEPTION, ERROR_NONE, ERROR_REQUEST_TIMED_OUT,
+ ERROR_TOPIC_AUTHORIZATION_FAILED, ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState,
+ HandleOutcome, REQUEST_DEADLINE, is_supported_version, supported_max_version,
};
+use crate::protocol::auto_topic_creation::is_internal_topic_name;
use crate::protocol::bounds_guard::validate_metadata_shape;
use crate::protocol::dedup::dedup_first_seen;
use crate::protocol::handlers::{decode_guarded, encode_message, respond_or_close};
@@ -50,6 +56,7 @@ const BLOCKING_ENCODE_PARTITION_THRESHOLD: u32 = 20_000;
pub async fn handle(
state: &GatewayState,
connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
api_version: i16,
body: Bytes,
) -> HandleOutcome {
@@ -85,17 +92,18 @@ pub async fn handle(
};
};
- let requested = match decode_requested_topics(api_version, body, state.max_frame_size) {
- Ok(requested) => requested,
- Err(error) => {
- tracing::debug!(
- %error,
- api_version,
- "Failed to decode Metadata request; closing connection"
- );
- return HandleOutcome::Close;
- }
- };
+ let (requested, allow_auto_topic_creation) =
+ match decode_requested_topics(api_version, body, state.max_frame_size) {
+ Ok(requested) => requested,
+ Err(error) => {
+ tracing::debug!(
+ %error,
+ api_version,
+ "Failed to decode Metadata request; closing connection"
+ );
+ return HandleOutcome::Close;
+ }
+ };
let per_partition_bytes = template_partition_bytes(api_version);
@@ -128,7 +136,13 @@ pub async fn handle(
return HandleOutcome::Close;
}
},
- Some(names) => resolve_requested_named_topics(bridge, &names).await,
+ Some(names) => {
+ let mut results = resolve_requested_named_topics(bridge, &names).await;
+ if allow_auto_topic_creation {
+ auto_create_missing(state, bridge, principal, &mut results);
+ }
+ results
+ }
};
// `bounds_guard` cannot see this: it charges the projected response by *requested* element
@@ -207,8 +221,12 @@ async fn resolve_requested_named_topics(
name,
error_code: ERROR_NONE,
partitions_count: metadata.partitions_count,
+ missing: false,
+ },
+ None => TopicResult {
+ missing: true,
+ ..error_result(name, ERROR_UNKNOWN_TOPIC_OR_PARTITION)
},
- None => error_result(name, ERROR_UNKNOWN_TOPIC_OR_PARTITION),
})
.collect(),
Ok(Err(error)) => {
@@ -234,6 +252,54 @@ async fn resolve_requested_named_topics(
}
}
+/// Kafka's `auto.create.topics.enable` for the topics a named lookup found missing
+/// (`KafkaApis.getTopicMetadata`).
+///
+/// A principal that may not create topics gets `TOPIC_AUTHORIZATION_FAILED` for them, one whose
+/// permissions were never read gets `UNKNOWN_TOPIC_OR_PARTITION` and nothing is created, and an
+/// invalid name gets `INVALID_TOPIC_EXCEPTION`. A Kafka-internal name, or one that resolves onto
+/// an override's Iggy topic without being its key, is left `UNKNOWN_TOPIC_OR_PARTITION` and not
+/// created. Every other missing topic starts being
+/// created and is answered `UNKNOWN_TOPIC_OR_PARTITION`, so the client finds it on a later refresh.
+fn auto_create_missing(
+ state: &GatewayState,
+ bridge: &Arc<IggyBridge>,
+ principal: Option<&AuthenticatedPrincipal>,
+ results: &mut [TopicResult],
+) {
+ let Some(creator) = &state.auto_topic_creation else {
+ return;
+ };
+ if !results.iter().any(|result| result.missing) {
+ return;
+ }
+ let refusal = match principal {
+ None => None,
+ Some(principal) if !principal.permissions_known => {
+ tracing::warn!(
+ principal = %principal.username,
+ "Metadata asked to auto-create topics on a connection whose permissions were \
+ never read; denying"
+ );
+ // Not 29: Java consumers, Streams and producers treat it as fatal, and an unread set
+ // would fail every refresh for the connection's life while Produce and Fetch work.
+ Some(ERROR_UNKNOWN_TOPIC_OR_PARTITION)
+ }
+ Some(principal) if principal.permissions.manage_topics => None,
+ Some(_) => Some(ERROR_TOPIC_AUTHORIZATION_FAILED),
+ };
+ for result in results.iter_mut().filter(|result| result.missing) {
+ let name = result.name.as_str();
+ if let Some(error_code) = refusal {
+ result.error_code = error_code;
+ } else if !is_valid_kafka_topic_name(name) {
+ result.error_code = ERROR_INVALID_TOPIC_EXCEPTION;
+ } else if !is_internal_topic_name(name) && !bridge.aliases_an_override(name) {
+ let _started = creator.start(bridge, name);
+ }
+ }
+}
+
/// Fallback per-partition byte cost, used only if [`template_partition_bytes`]'s own
/// `compute_size` call somehow fails for the fixed, always-well-formed shape it builds. Chosen
/// well above every version's real cost (measured 26-34 bytes) so a fallback can only truncate
@@ -339,6 +405,8 @@ struct TopicResult {
name: StrBytes,
error_code: i16,
partitions_count: u32,
+ /// The lookup found no such topic, as opposed to failing.
+ missing: bool,
}
fn found_result(metadata: crate::bridge::KafkaTopicMetadata) -> TopicResult {
@@ -346,6 +414,7 @@ fn found_result(metadata: crate::bridge::KafkaTopicMetadata) -> TopicResult {
name: StrBytes::from_string(metadata.kafka_topic),
error_code: ERROR_NONE,
partitions_count: metadata.partitions_count,
+ missing: false,
}
}
@@ -354,6 +423,7 @@ const fn error_result(name: StrBytes, error_code: i16) -> TopicResult {
name,
error_code,
partitions_count: 0,
+ missing: false,
}
}
@@ -473,14 +543,17 @@ fn decode_topics(api_version: i16, body: Bytes, max_frame_size: usize) -> Result
/// (`MetadataRequest.isAllTopics()`) is `topics == null || (topics.isEmpty() && version == 0)` -
/// there is no v0 wire shape for "cluster info only, zero topics" (that distinct shape, KIP-4's
/// `describeCluster()`, starts at v1), so an empty array at v0 can only mean "all topics."
+/// Also returns the request's `allow_auto_topic_creation`, which below v4 is always true, as in
+/// Kafka.
fn decode_requested_topics(
api_version: i16,
body: Bytes,
max_frame_size: usize,
-) -> Result<Option<Vec<StrBytes>>> {
+) -> Result<(Option<Vec<StrBytes>>, bool)> {
let req = decode_guarded::<MetadataRequest>(api_version, body, |v, b| {
validate_metadata_shape(v, b, max_frame_size)
})?;
+ let allow_auto_topic_creation = req.allow_auto_topic_creation;
let requested = req
.topics
.map(|topics| {
@@ -495,15 +568,20 @@ fn decode_requested_topics(
.collect::<Result<Vec<_>>>()
})
.transpose()?;
- Ok(match requested {
+ let requested = match requested {
Some(topics) if topics.is_empty() && api_version == 0 => None,
other => other,
- })
+ };
+ Ok((requested, allow_auto_topic_creation))
}
#[cfg(test)]
mod tests {
+ use tokio_util::sync::CancellationToken;
+
use super::*;
+ use crate::group::{GroupCoordinator, GroupCoordinatorConfig};
+ use crate::protocol::auto_topic_creation::AutoTopicCreationConfig;
const TEST_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024;
@@ -544,7 +622,7 @@ mod tests {
// None ("all topics"), not Some(vec![]) ("these zero topics").
let body = Bytes::from_static(&[0xff, 0xff, 0xff, 0xff]); // -1
let requested = decode_requested_topics(0, body, TEST_MAX_FRAME_SIZE).unwrap();
- assert_eq!(requested, None);
+ assert_eq!(requested.0, None);
}
#[test]
@@ -553,7 +631,7 @@ mod tests {
// zero topics" (KIP-4's describeCluster() shape).
let body = Bytes::from_static(&[0x00, 0x00, 0x00, 0x00]); // 0 topics, not -1
let requested = decode_requested_topics(1, body, TEST_MAX_FRAME_SIZE).unwrap();
- assert_eq!(requested, Some(Vec::new()));
+ assert_eq!(requested.0, Some(Vec::new()));
}
#[test]
@@ -562,7 +640,7 @@ mod tests {
// client can express "cluster info only, zero topics" - that shape starts at v1.
let body = Bytes::from_static(&[0x00, 0x00, 0x00, 0x00]); // 0 topics, not -1
let requested = decode_requested_topics(0, body, TEST_MAX_FRAME_SIZE).unwrap();
- assert_eq!(requested, None);
+ assert_eq!(requested.0, None);
}
#[test]
@@ -572,7 +650,7 @@ mod tests {
0x00, 0x06, b'o', b'r', b'd', b'e', b'r', b's', // "orders"
]);
let requested = decode_requested_topics(0, body, TEST_MAX_FRAME_SIZE).unwrap();
- assert_eq!(requested, Some(vec![StrBytes::from_static_str("orders")]));
+ assert_eq!(requested.0, Some(vec![StrBytes::from_static_str("orders")]));
}
#[test]
@@ -585,11 +663,53 @@ mod tests {
assert!(!response_would_exceed_frame_size(1024, 1024));
}
+ /// Below v4 the request has no `allow_auto_topic_creation` field, and Kafka treats it as
+ /// allowed.
+ #[test]
+ fn given_a_v3_request_when_decoding_should_allow_auto_topic_creation() {
+ let body = Bytes::from_static(&[
+ 0x00, 0x00, 0x00, 0x01, // one topic
+ 0x00, 0x06, b'o', b'r', b'd', b'e', b'r', b's',
+ ]);
+ let (requested, allow) = decode_requested_topics(3, body, TEST_MAX_FRAME_SIZE).unwrap();
+ assert_eq!(requested, Some(vec![StrBytes::from_static_str("orders")]));
+ assert!(allow);
+ }
+
+ /// A lookup that failed tells nothing about whether the topic exists, so an Iggy outage must
+ /// not start creating every topic clients name while it lasts.
+ #[tokio::test]
+ async fn given_auto_creation_when_the_lookup_fails_should_not_start_creating() {
+ let bridge = Arc::new(IggyBridge::offline());
+ let state = GatewayState::new(
+ Some(Arc::clone(&bridge)),
+ TEST_MAX_FRAME_SIZE,
+ 0,
+ GroupCoordinator::new(GroupCoordinatorConfig::default(), CancellationToken::new()),
+ )
+ .with_auto_topic_creation(AutoTopicCreationConfig {
+ enabled: true,
+ partitions: 1,
+ })
+ .expect("a valid partition count");
+ let body = Bytes::from_static(&[
+ 0x00, 0x00, 0x00, 0x01, // one topic
+ 0x00, 0x06, b'o', b'r', b'd', b'e', b'r', b's',
+ ]);
+
+ let outcome = handle(&state, &ConnectionState::default(), None, 3, body).await;
+
+ let _ = outcome.expect_response("a failed lookup is still answered");
+ let creator = state.auto_topic_creation.as_ref().expect("creation is on");
+ assert_eq!(creator.in_flight(), 0);
+ }
+
fn topic_result(name: &'static str, error_code: i16, partitions_count: u32) -> TopicResult {
TopicResult {
name: StrBytes::from_static_str(name),
error_code,
partitions_count,
+ missing: false,
}
}
diff --git a/gateways/kafka/src/protocol/handlers/mod.rs b/gateways/kafka/src/protocol/handlers/mod.rs
index 3748b3a1a..2d06a26ae 100644
--- a/gateways/kafka/src/protocol/handlers/mod.rs
+++ b/gateways/kafka/src/protocol/handlers/mod.rs
@@ -114,7 +114,7 @@ pub async fn dispatch(
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, connection, api_version, body).await,
+ API_KEY_METADATA => metadata::handle(state, connection, principal, api_version, body).await,
API_KEY_OFFSET_COMMIT => offset_commit::handle(state, connection, api_version, body).await,
API_KEY_OFFSET_FETCH => offset_fetch::handle(state, connection, api_version, body).await,
API_KEY_API_VERSIONS => api_versions::handle(connection, api_version, body).await,
diff --git a/gateways/kafka/src/protocol/mod.rs b/gateways/kafka/src/protocol/mod.rs
index 1ade5911f..cc36a468f 100644
--- a/gateways/kafka/src/protocol/mod.rs
+++ b/gateways/kafka/src/protocol/mod.rs
@@ -17,6 +17,7 @@
pub mod acl;
pub mod api;
+pub mod auto_topic_creation;
pub mod bounds_guard;
pub mod dedup;
pub mod handlers;
diff --git a/gateways/kafka/src/server.rs b/gateways/kafka/src/server.rs
index ab3e71d52..b7b84d3b8 100644
--- a/gateways/kafka/src/server.rs
+++ b/gateways/kafka/src/server.rs
@@ -51,6 +51,7 @@ use crate::protocol::api::{
decode_sasl_mechanism, describe_acls_outcome, encode_error_for_key, handle_connection_request,
respond_describe_acls_error, sasl_authenticate_outcome, sasl_handshake_outcome,
};
+use crate::protocol::auto_topic_creation::{self, AutoTopicCreationConfig};
use crate::protocol::handlers::api_versions::Advertisement;
use crate::protocol::header::{request_header_version, response_header_version};
use crate::protocol::sasl::{
@@ -160,6 +161,9 @@ pub struct GatewayConfig {
///
/// Client certificate user lookups get the same number of slots again, separate from these.
pub max_concurrent_authentications: usize,
+ /// Whether a `Metadata` lookup of a missing topic creates it, and with how many partitions
+ /// (`IGGY_KAFKA_AUTO_CREATE_TOPICS_*`). Off by default.
+ pub auto_topic_creation: AutoTopicCreationConfig,
}
impl Default for GatewayConfig {
@@ -180,6 +184,7 @@ impl Default for GatewayConfig {
instance_id: 0,
pre_auth_timeout: Duration::from_secs(15),
max_concurrent_authentications: 4,
+ auto_topic_creation: AutoTopicCreationConfig::default(),
}
}
}
@@ -478,12 +483,15 @@ impl KafkaGateway {
// coordinator can take a child token: a parked JoinGroup or SyncGroup waiter would
// otherwise hold the drain open for a full rebalance timeout.
let cancel = CancellationToken::new();
- let state = Arc::new(GatewayState::new(
- self.bridge.clone(),
- self.config.max_frame_size,
- self.config.instance_id,
- GroupCoordinator::new(self.config.group.clone(), cancel.child_token()),
- ));
+ let state = Arc::new(
+ GatewayState::new(
+ self.bridge.clone(),
+ self.config.max_frame_size,
+ self.config.instance_id,
+ GroupCoordinator::new(self.config.group.clone(), cancel.child_token()),
+ )
+ .with_auto_topic_creation(self.config.auto_topic_creation)?,
+ );
let shared_auth = Arc::new(SharedAuth::new(
self.authenticator.clone(),
resolver,
@@ -536,6 +544,11 @@ impl KafkaGateway {
cancel.cancel();
accept_loops.shutdown().await;
drain(&tracker, &cancel, drain_timeout).await;
+ // After the drain, so no connection starts one after it, and before `run` returns, so
+ // none still runs when the bridge signs out.
+ state
+ .shutdown_auto_topic_creation(auto_topic_creation::SHUTDOWN_WAIT)
+ .await;
outcome
}
diff --git a/gateways/kafka/tests/bridge_iggy_integration_tests.rs b/gateways/kafka/tests/bridge_iggy_integration_tests.rs
index 0651d0261..6273c4f1e 100644
--- a/gateways/kafka/tests/bridge_iggy_integration_tests.rs
+++ b/gateways/kafka/tests/bridge_iggy_integration_tests.rs
@@ -92,15 +92,19 @@ async fn given_timed_out_writes_when_writing_again_should_resolve_all_client_lan
let bridge = IggyBridge::connect(config).await.expect("bridge");
let raw = raw_client(&server).await;
let stream = Identifier::named("kafka").expect("stream name");
- // The first stream write commits successfully; the second commits an already-exists rejection.
+ bridge
+ .ensure_stream_and_topic("rejected", 1)
+ .await
+ .expect("seed the topic the second write is refused for");
+ // The first topic write commits successfully; the second commits an already-exists rejection.
for (delayed, next) in [("delayed", "next"), ("rejected", "after_rejection")] {
replies.send(false).expect("hold replies");
let result = bridge.create_kafka_topic(delayed, 1).await;
assert!(matches!(result, Err(BridgeError::Timeout)), "{result:?}");
assert!(
- raw.get_stream(&stream)
+ raw.get_topic(&stream, &Identifier::named(delayed).expect("topic name"))
.await
- .expect("stream lookup")
+ .expect("topic lookup")
.is_some(),
"the canceled write committed before its reply was released"
);
@@ -139,7 +143,7 @@ async fn given_timed_out_writes_when_writing_again_should_resolve_all_client_lan
.expect("next topic")
.is_some()
);
- for unsent in [delayed, "blocked", "blocked_deadline"] {
+ for unsent in ["blocked", "blocked_deadline"] {
assert!(
bridge
.get_kafka_topic(unsent)
diff --git a/gateways/kafka/tests/common/iggy_server.rs b/gateways/kafka/tests/common/iggy_server.rs
index 092a17841..c44101669 100644
--- a/gateways/kafka/tests/common/iggy_server.rs
+++ b/gateways/kafka/tests/common/iggy_server.rs
@@ -262,7 +262,12 @@ impl TestServer {
/// its assertions, orphaning a process that still holds its `PortGuard` slot, which then
/// fails the next test that draws that slot to bind.
pub async fn spawn_with_password(data_dir: &std::path::Path, password: &str) -> Self {
- Self::spawn_configured(data_dir, password, None).await
+ Self::spawn_configured(data_dir, password, None, &[]).await
+ }
+
+ /// [`Self::spawn`] with server settings of its own, as `IGGY_*` variables.
+ pub async fn spawn_with_env(data_dir: &std::path::Path, env: &[(&str, &str)]) -> Self {
+ Self::spawn_configured(data_dir, "iggy", None, env).await
}
/// Spawns `iggy-server` whose TCP listener speaks only TLS, with the given PEM pair.
@@ -274,13 +279,14 @@ impl TestServer {
cert_file: &std::path::Path,
key_file: &std::path::Path,
) -> Self {
- Self::spawn_configured(data_dir, "iggy", Some((cert_file, key_file))).await
+ Self::spawn_configured(data_dir, "iggy", Some((cert_file, key_file)), &[]).await
}
async fn spawn_configured(
data_dir: &std::path::Path,
password: &str,
tls: Option<(&std::path::Path, &std::path::Path)>,
+ env: &[(&str, &str)],
) -> Self {
let port_guard = PortGuard::acquire();
let address = format!("127.0.0.1:{}", port_guard.port);
@@ -321,7 +327,8 @@ impl TestServer {
// `--with-default-root-credentials` is off by default (args.rs), and without these
// the server refuses to boot. The password is the one the bridge logs in with.
.env("IGGY_ROOT_USERNAME", "iggy")
- .env("IGGY_ROOT_PASSWORD", password);
+ .env("IGGY_ROOT_PASSWORD", password)
+ .envs(env.iter().copied());
if let Some((cert_file, key_file)) = tls {
// `self_signed` is off explicitly: left on, the server would still load an existing
// pair, but the test should fail loudly on a missing file rather than mint one.
diff --git a/gateways/kafka/tests/kafka_client_e2e_tests.rs b/gateways/kafka/tests/kafka_client_e2e_tests.rs
index 1b68b1a43..3f6007c00 100644
--- a/gateways/kafka/tests/kafka_client_e2e_tests.rs
+++ b/gateways/kafka/tests/kafka_client_e2e_tests.rs
@@ -57,6 +57,7 @@ use iggy_gateway_kafka::bridge::{
TopicMapping,
};
use iggy_gateway_kafka::listener::{ListenerConfig, SecurityProtocol};
+use iggy_gateway_kafka::protocol::auto_topic_creation::AutoTopicCreationConfig;
use iggy_gateway_kafka::server::{BoundListener, bind_listener};
use iggy_gateway_kafka::{GatewayConfig, KafkaGateway};
@@ -338,7 +339,17 @@ impl BridgedGateway {
/// As [`spawn_bridged_gateway`], creating each `(topic, partitions)` first.
async fn start_bridged_gateway(server: &TestServer, topics: &[(&str, u32)]) -> BridgedGateway {
- start_gateway(server, topics, false).await
+ start_bridged_gateway_with(server, topics, GatewayConfig::default()).await
+}
+
+/// [`start_bridged_gateway`] with `gateway_config` instead of the defaults. Its listeners are
+/// replaced by one plaintext listener on an ephemeral port.
+async fn start_bridged_gateway_with(
+ server: &TestServer,
+ topics: &[(&str, u32)],
+ gateway_config: GatewayConfig,
+) -> BridgedGateway {
+ start_gateway(server, topics, false, gateway_config).await
}
/// A bridge to `server` as the root user, with the default topic mapping.
@@ -358,8 +369,14 @@ async fn root_bridge(server: &TestServer) -> IggyBridge {
.expect("the bridge connects to a ready server")
}
-/// A bridged gateway, with SASL on when `sasl` is set, creating each `(topic, partitions)` first.
-async fn start_gateway(server: &TestServer, topics: &[(&str, u32)], sasl: bool) -> BridgedGateway {
+/// A bridged gateway on `gateway_config`, with SASL on when `sasl` is set, creating each
+/// `(topic, partitions)` first. Its listeners are replaced by one on an ephemeral port.
+async fn start_gateway(
+ server: &TestServer,
+ topics: &[(&str, u32)],
+ sasl: bool,
+ gateway_config: GatewayConfig,
+) -> BridgedGateway {
let bridge = root_bridge(server).await;
for &(topic, partitions) in topics {
bridge
@@ -375,8 +392,7 @@ async fn start_gateway(server: &TestServer, topics: &[(&str, u32)], sasl: bool)
SecurityProtocol::Plaintext
};
let listener = BoundListener::new(ListenerConfig::new(protocol, addr.to_string()), listener);
- let mut gateway =
- KafkaGateway::new(GatewayConfig::default()).with_bridge(Some(Arc::new(bridge)));
+ let mut gateway = KafkaGateway::new(gateway_config).with_bridge(Some(Arc::new(bridge)));
if sasl {
gateway =
gateway.with_authenticator(Arc::new(IggyAuthenticator::new(server.address.clone())));
@@ -703,7 +719,7 @@ async fn bridged_stack() -> Option<(TestServer, SocketAddr)> {
/// serves it.
async fn sasl_bridged_stack() -> Option<(TestServer, SocketAddr)> {
let server = ready_server()?;
- let gateway = start_gateway(&server, &[(TOPIC, 1)], true).await;
+ let gateway = start_gateway(&server, &[(TOPIC, 1)], true, GatewayConfig::default()).await;
// Held for the test's lifetime: dropping the sender shuts the gateway down mid-exchange.
std::mem::forget(gateway.shutdown);
Some((server, gateway.addr))
@@ -1321,6 +1337,73 @@ fn consume_as_group(gateway: SocketAddr, count: usize) -> String {
.expect_ran("java console consumer in a group")
}
+/// With auto topic creation on, a stock Java producer sends to a topic nobody created and succeeds.
+/// Its first `Metadata` is answered `UNKNOWN_TOPIC_OR_PARTITION` while the gateway creates the
+/// topic, and it retries until the topic appears.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn given_auto_topic_creation_when_the_java_producer_sends_to_a_new_topic_should_succeed() {
+ const NEW_TOPIC: &str = "auto-created";
+ let Some(server) = ready_server() else {
+ return;
+ };
+ let gateway = start_bridged_gateway_with(
+ &server,
+ &[],
+ GatewayConfig {
+ auto_topic_creation: AutoTopicCreationConfig {
+ enabled: true,
+ partitions: 1,
+ },
+ ..GatewayConfig::default()
+ },
+ )
+ .await;
+ let bootstrap = gateway.addr.to_string();
+
+ let produce = format!(
+ "printf 'k1:alpha\\n' | /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server \
+ {bootstrap} --topic {NEW_TOPIC} --property parse.key=true --property key.separator=: \
+ --producer-property max.block.ms=20000"
+ );
+ let produced =
+ run_client(KAFKA_IMAGE, &["sh", "-c", &produce], &[]).expect_ran("java console producer");
+ assert!(
+ !produced.contains("ERROR"),
+ "the producer must send to the new topic, got: {produced}"
+ );
+
+ let consumed = run_client(
+ KAFKA_IMAGE,
+ &[
+ "/opt/kafka/bin/kafka-console-consumer.sh",
+ "--bootstrap-server",
+ &bootstrap,
+ "--topic",
+ NEW_TOPIC,
+ "--partition",
+ "0",
+ "--offset",
+ "earliest",
+ "--max-messages",
+ "1",
+ "--timeout-ms",
+ "20000",
+ "--property",
+ "print.offset=true",
+ "--property",
+ "print.key=true",
+ ],
+ &[],
+ )
+ .expect_ran("java console consumer");
+ assert_eq!(
+ records_read(&consumed),
+ vec!["Offset:0\tk1\talpha"],
+ "the auto-created topic must hold the record, got: {consumed}"
+ );
+ gateway.stop().await;
+}
+
/// The records a console consumer printed, as `Offset:<offset>\t<key>\t<value>`.
fn records_read(output: &str) -> Vec<&str> {
output
diff --git a/gateways/kafka/tests/metadata_real_bridge_tests.rs b/gateways/kafka/tests/metadata_real_bridge_tests.rs
index 9a45bd8d4..df1b8131a 100644
--- a/gateways/kafka/tests/metadata_real_bridge_tests.rs
+++ b/gateways/kafka/tests/metadata_real_bridge_tests.rs
@@ -21,18 +21,29 @@
//! than through `kafka_protocol`'s own `Encodable`/`Decodable` (this crate builds
//! `broker`-feature-only: request `Decodable` and response `Encodable`, not the reverse).
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
+use std::path::Path;
use std::sync::Arc;
+use std::time::Duration;
use bytes::Bytes;
+use iggy::prelude::{
+ GlobalPermissions, Identifier, Permissions, StreamClient, StreamPermissions, UserClient,
+ UserStatus,
+};
+use secrecy::SecretString;
use serial_test::serial;
use tokio_util::sync::CancellationToken;
+use iggy_gateway_kafka::auth::AuthenticatedPrincipal;
use iggy_gateway_kafka::bridge::{IggyBridge, TopicMapping, TopicOverride};
use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig};
+use iggy_gateway_kafka::protocol::acl::PrincipalPermissions;
use iggy_gateway_kafka::protocol::api::{
- ConnectionState, ERROR_NONE, ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState,
+ ConnectionState, ERROR_INVALID_TOPIC_EXCEPTION, ERROR_NONE, ERROR_TOPIC_AUTHORIZATION_FAILED,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState,
};
+use iggy_gateway_kafka::protocol::auto_topic_creation::AutoTopicCreationConfig;
use iggy_gateway_kafka::protocol::handlers::metadata;
#[path = "common/codec.rs"]
@@ -41,7 +52,7 @@ mod codec;
mod iggy_server;
use codec::{Decoder, Encoder};
-use iggy_server::TestServer;
+use iggy_server::{TestServer, raw_client};
const REQUEST_VERSION: i16 = 9;
const TEST_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024;
@@ -49,6 +60,11 @@ const TEST_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024;
/// Builds a v9 flexible `Metadata` request. `topics: None` means "all topics" (the null-array
/// sentinel); `Some(names)` requests exactly those topics.
fn build_request(topics: Option<&[&str]>) -> Bytes {
+ build_request_allowing(topics, false)
+}
+
+/// [`build_request`] with `allow_auto_topic_creation` set as given.
+fn build_request_allowing(topics: Option<&[&str]>, allow_auto_topic_creation: bool) -> Bytes {
let mut enc = Encoder::with_capacity(128);
match topics {
None => enc.write_varint(0), // null compact array: all topics
@@ -60,7 +76,7 @@ fn build_request(topics: Option<&[&str]>) -> Bytes {
}
}
}
- enc.write_bool(false); // allow_auto_topic_creation
+ enc.write_bool(allow_auto_topic_creation);
enc.write_bool(false); // include_cluster_authorized_operations
enc.write_bool(false); // include_topic_authorized_operations
enc.write_empty_tagged_fields(); // top-level tagged fields
@@ -145,7 +161,14 @@ fn expected_partitions(count: u32) -> Vec<PartitionEntry> {
async fn send(state: &GatewayState, topics: Option<&[&str]>) -> Vec<TopicEntry> {
let body = build_request(topics);
- let outcome = metadata::handle(state, &ConnectionState::default(), REQUEST_VERSION, body).await;
+ let outcome = metadata::handle(
+ state,
+ &ConnectionState::default(),
+ None,
+ REQUEST_VERSION,
+ body,
+ )
+ .await;
let resp_body = outcome.expect_response("Metadata request always answers");
decode_topics(resp_body)
}
@@ -157,13 +180,16 @@ async fn connected_state(server: &TestServer) -> (GatewayState, IggyBridge) {
let seed = IggyBridge::connect(server.test_config())
.await
.expect("seed bridge should connect to a ready server");
- let state = GatewayState::new(
+ (gateway_on(bridge), seed)
+}
+
+fn gateway_on(bridge: IggyBridge) -> GatewayState {
+ GatewayState::new(
Some(Arc::new(bridge)),
TEST_MAX_FRAME_SIZE,
0,
GroupCoordinator::new(GroupCoordinatorConfig::default(), CancellationToken::new()),
- );
- (state, seed)
+ )
}
#[tokio::test]
@@ -323,6 +349,7 @@ async fn a_response_projected_over_max_frame_size_closes_instead_of_answering()
let outcome = metadata::handle(
&tiny_state,
&ConnectionState::default(),
+ None,
REQUEST_VERSION,
body,
)
@@ -424,3 +451,443 @@ async fn a_named_lookup_of_more_than_the_old_topic_cap_resolves_every_name() {
assert_eq!(*error_code, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
}
}
+
+const AUTO_CREATED_PARTITIONS: u32 = 3;
+
+/// How long a test waits for a background creation to land before giving up.
+const AUTO_CREATE_WAIT: Duration = Duration::from_secs(10);
+
+/// How long a test waits before concluding nothing was created. A creation the gateway did start
+/// lands in milliseconds against a local server.
+const NO_CREATE_WAIT: Duration = Duration::from_millis(500);
+
+fn auto_creating(state: GatewayState) -> GatewayState {
+ state
+ .with_auto_topic_creation(AutoTopicCreationConfig {
+ enabled: true,
+ partitions: AUTO_CREATED_PARTITIONS,
+ })
+ .expect("a valid partition count")
+}
+
+async fn send_as(
+ state: &GatewayState,
+ principal: Option<&AuthenticatedPrincipal>,
+ names: &[&str],
+ allow_auto_topic_creation: bool,
+) -> Vec<TopicEntry> {
+ let body = build_request_allowing(Some(names), allow_auto_topic_creation);
+ let outcome = metadata::handle(
+ state,
+ &ConnectionState::default(),
+ principal,
+ REQUEST_VERSION,
+ body,
+ )
+ .await;
+ decode_topics(outcome.expect_response("Metadata request always answers"))
+}
+
+async fn exists(seed: &IggyBridge, name: &str) -> bool {
+ seed.get_kafka_topic(name)
+ .await
+ .expect("topic lookup")
+ .is_some()
+}
+
+fn principal(manage_topics: bool) -> AuthenticatedPrincipal {
+ principal_reading(manage_topics, true)
+}
+
+fn principal_reading(manage_topics: bool, permissions_known: bool) -> AuthenticatedPrincipal {
+ AuthenticatedPrincipal {
+ username: "alice".to_string(),
+ permissions: PrincipalPermissions {
+ manage_topics,
+ ..PrincipalPermissions::default()
+ },
+ permissions_known,
+ }
+}
+
+/// Kafka's own sequence: the first lookup answers `UNKNOWN_TOPIC_OR_PARTITION` while the topic is
+/// being created, and a later refresh finds it with the configured partition count.
+#[tokio::test]
+#[serial]
+async fn given_auto_creation_when_a_lookup_allows_it_should_create_the_missing_topic() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (state, _seed) = connected_state(&server).await;
+ let state = auto_creating(state);
+
+ let first = send_as(&state, None, &["orders"], true).await;
+ assert_eq!(
+ first,
+ vec![(
+ Some("orders".to_string()),
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ Vec::new()
+ )]
+ );
+
+ let deadline = tokio::time::Instant::now() + AUTO_CREATE_WAIT;
+ loop {
+ let topics = send_as(&state, None, &["orders"], true).await;
+ if topics[0].1 == ERROR_NONE {
+ assert_eq!(topics[0].2, expected_partitions(AUTO_CREATED_PARTITIONS));
+ break;
+ }
+ assert!(
+ tokio::time::Instant::now() < deadline,
+ "the topic was not created in time"
+ );
+ tokio::time::sleep(Duration::from_millis(50)).await;
+ }
+}
+
+#[tokio::test]
+#[serial]
+async fn given_auto_creation_when_the_lookup_does_not_allow_it_should_not_create() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (state, seed) = connected_state(&server).await;
+ let state = auto_creating(state);
+
+ let topics = send_as(&state, None, &["orders"], false).await;
+
+ assert_eq!(topics[0].1, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ tokio::time::sleep(NO_CREATE_WAIT).await;
+ assert!(!exists(&seed, "orders").await);
+}
+
+#[tokio::test]
+#[serial]
+async fn given_auto_creation_off_when_a_lookup_allows_it_should_not_create() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (state, seed) = connected_state(&server).await;
+
+ let topics = send_as(&state, None, &["orders"], true).await;
+
+ assert_eq!(topics[0].1, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ tokio::time::sleep(NO_CREATE_WAIT).await;
+ assert!(!exists(&seed, "orders").await);
+}
+
+#[tokio::test]
+#[serial]
+async fn given_auto_creation_when_the_name_is_invalid_should_answer_invalid_topic() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (state, _seed) = connected_state(&server).await;
+ let state = auto_creating(state);
+
+ let topics = send_as(&state, None, &["has a space"], true).await;
+
+ assert_eq!(topics[0].1, ERROR_INVALID_TOPIC_EXCEPTION);
+}
+
+#[tokio::test]
+#[serial]
+async fn given_auto_creation_when_the_principal_lacks_manage_topics_should_refuse_and_not_create() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (state, seed) = connected_state(&server).await;
+ let state = auto_creating(state);
+
+ let topics = send_as(&state, Some(&principal(false)), &["orders"], true).await;
+
+ assert_eq!(topics[0].1, ERROR_TOPIC_AUTHORIZATION_FAILED);
+ tokio::time::sleep(NO_CREATE_WAIT).await;
+ assert!(!exists(&seed, "orders").await);
+}
+
+#[tokio::test]
+#[serial]
+async fn given_auto_creation_when_the_principal_holds_manage_topics_should_create() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (state, seed) = connected_state(&server).await;
+ let state = auto_creating(state);
+
+ let topics = send_as(&state, Some(&principal(true)), &["orders"], true).await;
+
+ assert_eq!(topics[0].1, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ let deadline = tokio::time::Instant::now() + AUTO_CREATE_WAIT;
+ while !exists(&seed, "orders").await {
+ assert!(
+ tokio::time::Instant::now() < deadline,
+ "the topic was not created in time"
+ );
+ tokio::time::sleep(Duration::from_millis(50)).await;
+ }
+}
+
+/// An existing topic in the same request is answered as usual, and only the missing one is
+/// created.
+#[tokio::test]
+#[serial]
+async fn given_auto_creation_when_one_of_two_topics_exists_should_answer_it_and_create_the_other() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (state, seed) = connected_state(&server).await;
+ let state = auto_creating(state);
+ seed.ensure_stream_and_topic("payments", 2)
+ .await
+ .expect("seed the topic");
+
+ let topics = send_as(&state, None, &["payments", "orders"], true).await;
+
+ assert_eq!(
+ topics[0],
+ (
+ Some("payments".to_string()),
+ ERROR_NONE,
+ expected_partitions(2)
+ )
+ );
+ assert_eq!(topics[1].1, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ let deadline = tokio::time::Instant::now() + AUTO_CREATE_WAIT;
+ while !exists(&seed, "orders").await {
+ assert!(
+ tokio::time::Instant::now() < deadline,
+ "the topic was not created in time"
+ );
+ tokio::time::sleep(Duration::from_millis(50)).await;
+ }
+}
+
+/// A permission set that was never read may not create, even when its fallback says
+/// `manage_topics`. Answered as unknown, not refused, so the client keeps refreshing.
+#[tokio::test]
+#[serial]
+async fn given_auto_creation_when_the_principals_permissions_were_never_read_should_answer_unknown_and_not_create()
+ {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (state, seed) = connected_state(&server).await;
+ let state = auto_creating(state);
+
+ let unread = principal_reading(true, false);
+ let topics = send_as(&state, Some(&unread), &["orders"], true).await;
+
+ assert_eq!(topics[0].1, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ tokio::time::sleep(NO_CREATE_WAIT).await;
+ assert!(!exists(&seed, "orders").await);
+}
+
+/// A connection that proved no identity on a gateway that names principals holds nothing.
+#[tokio::test]
+#[serial]
+async fn given_auto_creation_when_the_connection_is_anonymous_should_refuse() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (state, seed) = connected_state(&server).await;
+ let state = auto_creating(state);
+
+ let anonymous = AuthenticatedPrincipal::anonymous();
+ let topics = send_as(&state, Some(&anonymous), &["orders"], true).await;
+
+ assert_eq!(topics[0].1, ERROR_TOPIC_AUTHORIZATION_FAILED);
+ tokio::time::sleep(NO_CREATE_WAIT).await;
+ assert!(!exists(&seed, "orders").await);
+}
+
+#[tokio::test]
+#[serial]
+async fn given_auto_creation_when_the_name_is_kafka_internal_should_not_create() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (state, seed) = connected_state(&server).await;
+ let state = auto_creating(state);
+
+ let topics = send_as(&state, None, &["__consumer_offsets"], true).await;
+
+ assert_eq!(topics[0].1, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ tokio::time::sleep(NO_CREATE_WAIT).await;
+ assert!(!exists(&seed, "__consumer_offsets").await);
+}
+
+/// Only Kafka's own internal topics are reserved. Debezium names its heartbeat topics with `__`.
+#[tokio::test]
+#[serial]
+async fn given_auto_creation_when_a_double_underscore_name_is_not_kafka_internal_should_create() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (state, seed) = connected_state(&server).await;
+ let state = auto_creating(state);
+ let name = "__debezium-heartbeat.inventory";
+
+ let topics = send_as(&state, None, &[name], true).await;
+
+ assert_eq!(topics[0].1, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ let deadline = tokio::time::Instant::now() + AUTO_CREATE_WAIT;
+ while !exists(&seed, name).await {
+ assert!(
+ tokio::time::Instant::now() < deadline,
+ "the topic was not created in time"
+ );
+ tokio::time::sleep(Duration::from_millis(50)).await;
+ }
+}
+
+async fn wait_until_created(seed: &IggyBridge, name: &str) {
+ let deadline = tokio::time::Instant::now() + AUTO_CREATE_WAIT;
+ while !exists(seed, name).await {
+ assert!(
+ tokio::time::Instant::now() < deadline,
+ "the topic was not created in time"
+ );
+ tokio::time::sleep(Duration::from_millis(50)).await;
+ }
+}
+
+/// The default stream does not exist yet, so the creation makes it before the topic.
+#[tokio::test]
+#[serial]
+async fn given_auto_creation_when_the_mapped_stream_is_missing_should_create_the_stream_and_the_topic()
+ {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (state, seed) = connected_state(&server).await;
+ let state = auto_creating(state);
+ let root = raw_client(&server).await;
+ let stream = Identifier::named("kafka").expect("stream name");
+ assert!(
+ root.get_stream(&stream)
+ .await
+ .expect("stream lookup")
+ .is_none()
+ );
+
+ let topics = send_as(&state, None, &["orders"], true).await;
+
+ assert_eq!(topics[0].1, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ wait_until_created(&seed, "orders").await;
+}
+
+const LIMITED_USER: &str = "kafka-bridge";
+const LIMITED_PASSWORD: &str = "kafka-bridge-secret";
+
+/// Iggy lets a user create topics in a stream it holds `manage_topics` on, while creating a
+/// stream needs the global `manage_streams`. Such a bridge user auto-creates into a stream that
+/// already exists.
+#[tokio::test]
+#[serial]
+async fn given_a_bridge_user_with_manage_topics_on_the_stream_only_when_auto_creating_there_should_create()
+ {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let manage_topics = StreamPermissions {
+ manage_topics: true,
+ ..StreamPermissions::default()
+ };
+ let state = auto_creating(gateway_on(limited_bridge(&server, manage_topics).await));
+ let seed = IggyBridge::connect(server.test_config())
+ .await
+ .expect("seed bridge");
+
+ let topics = send_as(&state, None, &["orders"], true).await;
+
+ assert_eq!(topics[0].1, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ wait_until_created(&seed, "orders").await;
+}
+
+/// Creates the default stream, then a bridge user holding only `on_stream` on it, signed in.
+async fn limited_bridge(server: &TestServer, on_stream: StreamPermissions) -> IggyBridge {
+ let root = raw_client(server).await;
+ let stream = root.create_stream("kafka").await.expect("seed the stream");
+ let stream_id = usize::try_from(stream.id).expect("stream id fits usize");
+ let permissions = Permissions {
+ global: GlobalPermissions::default(),
+ streams: Some(BTreeMap::from([(stream_id, on_stream)])),
+ };
+ root.create_user(
+ LIMITED_USER,
+ LIMITED_PASSWORD,
+ UserStatus::Active,
+ Some(permissions),
+ )
+ .await
+ .expect("create the bridge user");
+ let mut config = server.test_config();
+ config.username = LIMITED_USER.to_string();
+ config.password = SecretString::from(LIMITED_PASSWORD);
+ IggyBridge::connect(config)
+ .await
+ .expect("the limited bridge user signs in")
+}
+
+/// Lets the first creation sign its client in, which Iggy journals, before the journal is
+/// measured.
+const FIRST_CREATION_WAIT: Duration = Duration::from_millis(500);
+
+/// Long enough for two retries after the first creation's hold.
+const RETRY_WINDOW: Duration = Duration::from_secs(3);
+
+fn journal_len(data_dir: &Path) -> u64 {
+ std::fs::metadata(data_dir.join("metadata").join("journal.wal"))
+ .expect("the metadata journal")
+ .len()
+}
+
+/// Starts creating `kafka_topic`, then names it every 100 ms for [`RETRY_WINDOW`], as a producer
+/// waiting for it does. Returns the journal's length before the first creation, and after it.
+async fn name_repeatedly(state: &GatewayState, data_dir: &Path, kafka_topic: &str) -> (u64, u64) {
+ let before = journal_len(data_dir);
+ let _ = send_as(state, None, &[kafka_topic], true).await;
+ tokio::time::sleep(FIRST_CREATION_WAIT).await;
+ let after_first = journal_len(data_dir);
+ let deadline = tokio::time::Instant::now() + RETRY_WINDOW;
+ while tokio::time::Instant::now() < deadline {
+ let topics = send_as(state, None, &[kafka_topic], true).await;
+ assert_eq!(topics[0].1, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ tokio::time::sleep(Duration::from_millis(100)).await;
+ }
+ (before, after_first)
+}
+
+/// At `metadata.partitions_max`, Iggy refuses a topic before journaling it. A retried creation
+/// into a stream that exists must not write anything else, such as a `create_stream` Iggy only
+/// refuses after journaling it.
+#[tokio::test]
+#[serial]
+async fn given_auto_creation_at_the_partition_cap_when_it_retries_should_journal_nothing() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server =
+ TestServer::spawn_with_env(data_dir.path(), &[("IGGY_METADATA_PARTITIONS_MAX", "1")]).await;
+ let (state, seed) = connected_state(&server).await;
+ let state = auto_creating(state);
+ seed.ensure_stream_and_topic("payments", 1)
+ .await
+ .expect("fill the partition cap");
+
+ let (before, after_first) = name_repeatedly(&state, data_dir.path(), "orders").await;
+
+ assert!(after_first > before, "the first creation signed in");
+ assert_eq!(journal_len(data_dir.path()), after_first);
+ assert!(!exists(&seed, "orders").await);
+}
+
+/// Iggy refuses a create the bridge user may not make only after journaling it, so a name it
+/// refused is not retried while clients keep naming it.
+#[tokio::test]
+#[serial]
+async fn given_a_bridge_user_iggy_refuses_when_the_topic_is_named_again_should_not_retry_soon() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let read_only = StreamPermissions {
+ read_stream: true,
+ ..StreamPermissions::default()
+ };
+ let state = auto_creating(gateway_on(limited_bridge(&server, read_only).await));
+ let seed = IggyBridge::connect(server.test_config())
+ .await
+ .expect("seed bridge");
+
+ let (before, after_first) = name_repeatedly(&state, data_dir.path(), "orders").await;
+
+ assert!(after_first > before, "the first creation reached Iggy");
+ assert_eq!(journal_len(data_dir.path()), after_first);
+ assert!(!exists(&seed, "orders").await);
+}