feat(gateways): Kafka Gateway add DescribeConfigs and AlterConfigs APIs

gateways/kafka/README.md

@@ -1,18 +1,18 @@
 # Kafka gateway (`iggy-gateway-kafka`)
 
-Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener on the Kafka wire port that decodes requests and validates scoped API keys and versions. With a bridge, Produce, Fetch, ListOffsets, Metadata, CreateTopics, OffsetCommit and OffsetFetch use Iggy. InitProducerId and consumer group coordination work with or without one.
+Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener on the Kafka wire port that decodes requests and validates scoped API keys and versions. With a bridge, Produce, Fetch, ListOffsets, Metadata, CreateTopics, DescribeConfigs and AlterConfigs use Iggy. InitProducerId and consumer group coordination work with or without one.
 
 > **Stub warning:** with `IGGY_KAFKA_BRIDGE_ENABLED=true`, Produce, Fetch, ListOffsets, Metadata,
-> CreateTopics, OffsetCommit and OffsetFetch use Iggy. With the bridge off (the default), those
+> CreateTopics, DescribeConfigs and AlterConfigs use Iggy. With the bridge off (the default), those
 > seven are stubs: Produce, Fetch and ListOffsets answer retriable `NOT_LEADER_OR_FOLLOWER` (6),
-> CreateTopics answers `NOT_CONTROLLER` (41), OffsetCommit and OffsetFetch answer retriable
-> `COORDINATOR_LOAD_IN_PROGRESS` (14), and Metadata reports every requested topic unknown.
-> **CreateTopics runs as the bridge's own
+> CreateTopics, DescribeConfigs and AlterConfigs answer `NOT_CONTROLLER` (41), and Metadata reports
+> every requested topic unknown. **CreateTopics 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). OffsetCommit creates Iggy
-> resources too: a consumer group `kafka.cg.<group>` per group id and topic, which nothing deletes
-> ([cleanup](docs/OFFSET_STORAGE.md#limits)). Every client reads and writes as the bridge's Iggy
-> user. See [docs/SCOPE.md](docs/SCOPE.md).
+> can reach this port can create topics (up to 1000 partitions each) or shorten a topic's
+> `retention.ms` low enough that the cleaner drops sealed segments, deleting data. With SASL on,
+> `AlterConfigs` additionally requires the authenticated principal to hold `manage_topics`; without
+> it every resource in the request fails `TOPIC_AUTHORIZATION_FAILED` (29) and nothing is written.
+> 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.
 >
@@ -193,6 +193,7 @@ See [docs/SCOPE.md](docs/SCOPE.md) for [#3421](https://github.com/apache/iggy/is
 - [docs/CONSUMER_GROUPS.md](docs/CONSUMER_GROUPS.md) — group membership, rebalances, and why one gateway per bootstrap endpoint
 - [docs/AUTHENTICATION.md](docs/AUTHENTICATION.md) — how a Kafka client authenticates, and why PLAIN only
 - [docs/ACL_MAPPING.md](docs/ACL_MAPPING.md) — how Iggy permissions are described as Kafka ACLs
+- [docs/CONFIGS.md](docs/CONFIGS.md) — DescribeConfigs and AlterConfigs for topic `retention.ms`
 
 ### Delivery guarantees
 
@@ -307,7 +308,8 @@ a topic per call) for `ListOffsets`, probes and polls partitions for Fetch, and
 group offsets for OffsetCommit and OffsetFetch.
 Produce ([#3535](https://github.com/apache/iggy/issues/3535)), Fetch ([#3536](https://github.com/apache/iggy/issues/3536)), ListOffsets
 ([#3537](https://github.com/apache/iggy/issues/3537)), Metadata ([#3534](https://github.com/apache/iggy/issues/3534)), CreateTopics
-([#3538](https://github.com/apache/iggy/issues/3538)), OffsetCommit and OffsetFetch
+([#3538](https://github.com/apache/iggy/issues/3538)), DescribeConfigs and AlterConfigs , OffsetCommit and OffsetFetchcall it.
+Topic configuration behavior is in [docs/CONFIGS.md](docs/CONFIGS.md).
 ([#3542](https://github.com/apache/iggy/issues/3542)) call it.
 Tested by `bridge`'s own unit tests, `tests/bridge_iggy_integration_tests.rs` and the
 `tests/*_real_bridge_tests.rs` suites. All but the unit tests start a real `iggy-server`.
@@ -487,9 +489,9 @@ call created it." A *different* `partition_count` against an already-existing to
 two concurrent callers requesting different counts for the same topic must not both see success.
 
 Topics created this way have **no message expiry** - Iggy's own server default, not Kafka's 7-day
-default. Nothing is bounding retention until it's configured explicitly (Iggy's own topic options,
-outside this bridge today); repointing a Kafka app that assumes bounded retention onto this bridge
-will accumulate data indefinitely unless you set that up yourself.
+default. `AlterConfigs` can set that expiry afterwards through `retention.ms`
+([docs/CONFIGS.md](docs/CONFIGS.md)). Until a client does, messages never expire, so a Kafka app
+that assumes the 7-day default will accumulate data on this bridge.
 
 They also use Iggy's default **durability**, `Durability::Replicated` - quorum commit without an
 additional stable-storage barrier, with the disk write itself threshold-gated (flushed at 1024
@@ -499,9 +501,9 @@ cut before that threshold is reached - worth knowing rather than discovering lat
 
 ### Concurrency ceiling
 
-- One `IggyClient` serves Produce, Metadata and CreateTopics for every Kafka connection, one Iggy
-  request at a time. Fetch polls use 4 more, one per read slot. Topic probes use 1 more. Offset
-  calls use 8 more: a commit client and a read client per offset slot.
+- One `IggyClient` serves Produce, Metadata, CreateTopics, DescribeConfigs and AlterConfigs for
+  every Kafka connection, one Iggy request at a time. Fetch polls use 4 more, one per read slot.
+  Topic probes use 1 more. Offset  calls use 8 more: a commit client and a read client per offset slot.
 - OffsetCommit and OffsetFetch cost one Iggy call per partition, spread over the offset slots. In a
   cluster, an offset commit client can also open one connection per node for stores.
 - `IGGY_KAFKA_MAX_CONNECTIONS` does not change that. A Produce client pool is a TODO in

gateways/kafka/docs/CONFIGS.md

@@ -0,0 +1,109 @@
+# Topic configuration
+
+`DescribeConfigs` (32) and `AlterConfigs` (33) apply to Kafka topic resources only
+(resource type 2). Only the deprecated `Admin.alterConfigs` (API 33) writes
+retention. `IncrementalAlterConfigs` (44) is not in this change and is not
+advertised in `ApiVersions`. A client picks its admin API from that response, so
+a Java or librdkafka client that would otherwise send 44 raises an
+`UnsupportedVersionException` (or the librdkafka equivalent) locally and never
+sends the request - the gateway never sees it, and the connection is not
+affected. Only `AlterConfigs` (33) writes retention.
+
+With the bridge off, both APIs answer `NOT_CONTROLLER` (41) for each resource and
+do not read or write Iggy.
+
+## Authorization
+
+`AlterConfigs` requires the `manage_topics` (or `manage_streams`) Iggy permission
+on the authenticated principal when SASL is enabled. A principal without it gets
+`TOPIC_AUTHORIZATION_FAILED` (29) for every resource in the request, and
+`update_topic` is never called. With SASL disabled there is no principal to
+check, and the request is not gated here. `DescribeConfigs` performs no
+authorization check of its own.
+
+## What a describe returns
+
+A null or empty `configuration_keys` list returns both keys below. A non-empty
+list returns only the requested names this gateway knows, in request order,
+with a name repeated in the list collapsed to one entry. A name it does not
+know, including `retention.minutes` and `retention.hours`, is omitted. The
+resource's error code stays `NONE` (0), and the known keys in that request are
+still returned.
+
+| Key | Value | Read only | Source |
+| --- | ----- | --------- | ------ |
+| `retention.ms` | Sealed-segment lifetime from Iggy `message_expiry`, or `-1` when those segments never expire | no | `DEFAULT_CONFIG` (5) for the unset never-expire default; `DYNAMIC_TOPIC_CONFIG` (1) when a client set it or a stored duration cannot be shown as milliseconds |
+| `cleanup.policy` | `delete` | yes | `DEFAULT_CONFIG` (5) |
+
+`is_sensitive` is false for both. From v3, `include_documentation` is honored:
+true attaches one sentence per known key, false sends no documentation.
+`include_synonyms` has no effect: this gateway models no config-inheritance
+chain beyond `retention.ms` itself, so the synonym list on every entry is
+always empty.
+
+Iggy stores `message_expiry` as `IggyExpiry`. A duration is `IggyDuration`, and
+that type counts microseconds. Retention applies to sealed segments only. Iggy
+does not time-roll, so the active segment does not expire. Describe divides by
+1000 only when the stored count is a positive whole number of milliseconds.
+Anything else, including 0 microseconds, is `INVALID_CONFIG` (40) with a null
+value and source `DYNAMIC_TOPIC_CONFIG` (1). That expiry was stored; it is not
+the never-expire default. The gateway does not round.
+
+## What an alter stores
+
+`retention.ms` is the only writable duration. Iggy stores milliseconds, through
+`update_topic`'s `message_expiry`. `retention.minutes` and `retention.hours` are
+not real Kafka topic-level configs - only the broker-level `log.retention.minutes`
+and `log.retention.hours` exist in real Kafka - so this gateway rejects both with
+`INVALID_CONFIG` (40), the same as any other unknown key.
+
+- `retention.ms` is the source of truth, and the only retention key this
+  gateway writes.
+- `-1` stores `IggyExpiry::NeverExpire`. The server treats `0`
+  (`IggyExpiry::ServerDefault`) as "leave the current expiry", so `-1` is
+  not sent as 0. A following describe reports `-1` with source
+  `DYNAMIC_TOPIC_CONFIG`.
+- A positive count must be the canonical decimal of that integer (`1`, not
+  `0001` or `+1`). The resulting duration must be at most `u32::MAX` seconds,
+  the same limit `IggyExpiry::from_str` enforces. A longer value is
+  `INVALID_CONFIG` (40) and is not stored. A following describe of
+  `retention.ms` reports that millisecond count with source
+  `DYNAMIC_TOPIC_CONFIG`.
+- `0` and every other value are `INVALID_CONFIG` (40).
+
+Kafka's `AlterConfigs` replaces a topic's full configuration. An empty `configs`
+list on that API is a reset to defaults. This gateway patches the named keys
+instead of replacing the set, so an empty list is `INVALID_CONFIG` (40) and
+writes nothing. A previously set expiry stays in place.
+
+`cleanup.policy`, `retention.minutes`, `retention.hours`, and any other key are
+`INVALID_CONFIG` (40). They are not stored. Kafka has no `NOT_CONFIGURABLE`
+code. One bad key fails that resource and writes nothing for it. Other
+resources in the same request are still processed.
+
+`validate_only` runs the same checks, including the topic lookup, and does not
+call `update_topic`.
+
+## Limitations
+
+- `IncrementalAlterConfigs` (API 44) is not implemented here. It is planned
+  after this work merges. It is not advertised, so a client falls back to
+  `AlterConfigs` (33); see "Authorization" above for what an unadvertised key
+  does to the client.
+- `retention.ms` remains the only value Iggy stores.
+
+## Other resource errors
+
+| Condition | Code | Notes |
+| --------- | ---- | ----- |
+| Caller lacks `manage_topics` (SASL on, `AlterConfigs` only) | `TOPIC_AUTHORIZATION_FAILED` (29) | Checked before the topic lookup. Every resource in the request |
+| Resource type other than topic | `INVALID_REQUEST` (42) | Message is `only topic resources are supported`. Other resources in the batch still run |
+| Topic name fails the same rules as CreateTopics | `INVALID_TOPIC_EXCEPTION` (17) | The message is the validation reason and does not repeat the topic name |
+| Topic does not exist | `UNKNOWN_TOPIC_OR_PARTITION` (3) | Checked before config keys, so a missing topic is not `INVALID_CONFIG` |
+| Empty `configs` list on alter | `INVALID_CONFIG` (40) | Kafka would replace the set, which is a reset. This gateway patches named keys, so an empty list is rejected and writes nothing |
+| More than 100 distinct topic names | `POLICY_VIOLATION` (44) | Every resource in the request, checked before the bridge-availability check |
+| A topic name repeated across resources | `INVALID_REQUEST` (42) | Every occurrence of the duplicate, not just the second one. Same choice CreateTopics makes for a repeated topic name |
+
+Unknown tagged fields at the request, resource, or config-entry level are
+ignored, not rejected: Kafka's flexible versions define them as a forward-compatible
+extension point a server that does not recognize them must skip, not refuse.

gateways/kafka/docs/MANUAL_TESTING.md

@@ -224,13 +224,15 @@ For each API key, test **min−1**, **min**, **max**, **max+1** using `kafka-mes
 | 13 | LeaveGroup | 0 | 5 | −1, 0, 5, 6 |
 | 14 | SyncGroup | 0 | 5 | −1, 0, 5, 6 |
 | 22 | InitProducerId | 0 | 5 | −1, 0, 5, 6 |
+| 32 | DescribeConfigs | 1 | 4 | 0, 1, 4, 5 |
+| 33 | AlterConfigs | 0 | 2 | −1, 0, 2, 3 |
 
 | ID | Test | Expected for in-range | Expected for out-of-range |
 | ---- | ------ | ---------------------- | --------------------------- |
 | B1 | ApiVersions negotiation | `error_code=0`; body lists 14 API keys with correct min/max | KIP-511 exception: still answers, `error_code=35` (UNSUPPORTED_VERSION), v0 response header regardless of the request's own encoding |
 | B2 | Metadata out-of-range | N/A | **Connection closes**, no response sent - Metadata has no top-level error field to carry a version-correct error in |
-| B3 | Produce/Fetch/ListOffsets/CreateTopics/InitProducerId/OffsetCommit/OffsetFetch out-of-range | N/A | **Connection closes** for both above-max and below-min - `kafka_protocol`'s schema floor for each of these seven messages equals `SUPPORTED_RANGES`' own min, so there is no encodable error response below min either (see `SCOPE.md`'s Governance model) |
-| B4 | ApiVersions lists only scoped keys | Decode response | Contains keys 0,1,2,3,8,9,10,11,12,13,14,18,19,22 only, and no transaction keys (24, 25, 26, 28) |
+| B3 | Produce/Fetch/ListOffsets/CreateTopics/InitProducerId/OffsetCommit/OffsetFetch  out-of-range | N/A | **Connection closes** for both above-max and below-min - `kafka_protocol`'s schema floor for each of these five messages equals `SUPPORTED_RANGES`' own min, so there is no encodable error response below min either (see `SCOPE.md`'s Governance model) |
+| B4 | ApiVersions lists only scoped keys | Decode response | Contains keys 0,1,2,3,8,9,10,11,12,13,14,18,19,22,32,33 only, and no transaction keys (24, 25, 26, 28) |
 
 An out-of-range version only ever produces `error_code=35` on ApiVersions (B1); every other API
 key's out-of-range case closes the connection - see B2/B3. InitProducerId and Produce also send