This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 2d7fddba5 feat(gateways): Kafka gateway Phase 1 quick-start and Docker
Compose E2E (#4372)
2d7fddba5 is described below
commit 2d7fddba5552640518aa7f1372f69af9c77e436a
Author: ryerraguntla <[email protected]>
AuthorDate: Sat Oct 3 01:45:07 2026 -0400
feat(gateways): Kafka gateway Phase 1 quick-start and Docker Compose E2E
(#4372)
---
.config/nextest.toml | 3 +
.pr4372-comments.json | 0
gateways/kafka/Dockerfile | 64 +++++
gateways/kafka/Dockerfile.dockerignore | 48 ++++
gateways/kafka/README.md | 98 +++++++
gateways/kafka/docker-compose.yml | 141 ++++++++++
gateways/kafka/docs/MANUAL_TESTING.md | 35 ++-
gateways/kafka/docs/TEST_SUITE.md | 17 +-
gateways/kafka/tests/common/gateway_process.rs | 102 ++++++++
gateways/kafka/tests/common/iggy_server.rs | 86 ++++---
gateways/kafka/tests/phase1_e2e_tests.rs | 341 +++++++++++++++++++++++++
licenserc.toml | 1 +
12 files changed, 898 insertions(+), 38 deletions(-)
diff --git a/.config/nextest.toml b/.config/nextest.toml
index d9d69fbca..1cedeed01 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -63,6 +63,9 @@ filter = '''
+ binary_id(iggy-gateway-kafka::fetch_real_bridge_tests)
+ binary_id(iggy-gateway-kafka::list_offsets_real_bridge_tests)
+ binary_id(iggy-gateway-kafka::produce_real_bridge_tests)
+ + binary_id(iggy-gateway-kafka::create_topics_real_bridge_tests)
+ + binary_id(iggy-gateway-kafka::metadata_real_bridge_tests)
+ + binary_id(iggy-gateway-kafka::phase1_e2e_tests)
'''
test-group = "kafka_bridge"
diff --git a/.pr4372-comments.json b/.pr4372-comments.json
new file mode 100644
index 000000000..e69de29bb
diff --git a/gateways/kafka/Dockerfile b/gateways/kafka/Dockerfile
new file mode 100644
index 000000000..d5d95eb68
--- /dev/null
+++ b/gateways/kafka/Dockerfile
@@ -0,0 +1,64 @@
+# 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.
+
+# Build context is the repository root (a Cargo workspace member cannot build
standalone) -
+# `docker-compose.yml` in this directory sets `context: ../..` accordingly.
+
+ARG RUST_VERSION=1.98
+FROM rust:${RUST_VERSION}-slim-trixie AS builder
+
+WORKDIR /build
+# build-essential: the zstd/lz4/snap codecs kafka-protocol pulls in compile a
small amount of
+# vendored C. No libssl-dev - core/sdk's TLS is rustls, not openssl.
+RUN apt-get update && apt-get install -y \
+ build-essential \
+ pkg-config \
+ && rm -rf /var/lib/apt/lists/*
+
+COPY rust-toolchain.toml rust-toolchain.toml
+# Resolves/installs the toolchain now, in a layer keyed only on
rust-toolchain.toml - matching
+# RUST_VERSION above means this is a no-op resolution, not a download, but
isolates the cost
+# from the full COPY . . below, which invalidates on any source change in the
workspace.
+# Same ordering core/server/Dockerfile uses for the same reason.
+RUN cargo --version
+
+COPY . .
+# Cache mounts persist across builds but not into the image layer, so the
binary is copied out
+# to a plain path before the mount unmounts. --locked matches CI (scripts
never resolve a
+# dependency version CI didn't already lock).
+RUN --mount=type=cache,target=/usr/local/cargo/registry,sharing=locked \
+ --mount=type=cache,target=/build/target,sharing=locked \
+ cargo build --release --locked --package iggy-gateway-kafka --bin
iggy-gateway-kafka && \
+ cp target/release/iggy-gateway-kafka /iggy-gateway-kafka
+
+FROM debian:trixie-slim
+# netcat-openbsd: nothing in the gateway's own dependency graph needs it -
it's here only so
+# docker-compose.yml's healthcheck has a TCP-probe primitive to call.
+RUN apt-get update && apt-get install -y \
+ ca-certificates \
+ netcat-openbsd \
+ && rm -rf /var/lib/apt/lists/*
+COPY --from=builder /iggy-gateway-kafka .
+
+ENV IGGY_KAFKA_BIND_ADDR=0.0.0.0:9093
+# No baked IGGY_KAFKA_ADVERTISED_HOST default: wildcard bind means the gateway
cannot derive
+# its own advertised host (server.rs's BrokerAdvertise::from_server_config
refuses to start
+# without one), and that refusal is deliberate - it forces whoever runs this
image to say what
+# host a client can actually reach it on, rather than silently advertising
"localhost" to a
+# client on a different machine. docker-compose.yml sets it explicitly for
this quick start.
+
+CMD ["/iggy-gateway-kafka"]
diff --git a/gateways/kafka/Dockerfile.dockerignore
b/gateways/kafka/Dockerfile.dockerignore
new file mode 100644
index 000000000..6386a3b5b
--- /dev/null
+++ b/gateways/kafka/Dockerfile.dockerignore
@@ -0,0 +1,48 @@
+# 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.
+
+# This image only. Root .dockerignore must keep .git so the server image
+# can stamp vergen git metadata. This file replaces the root ignore file
+# for gateways/kafka/Dockerfile.
+.git
+.config
+.dockerignore
+.github
+.gitignore
+.idea
+.vscode
+.zed
+/assets
+/core/certs
+/helm
+/licenses
+/local_data
+/performance_results
+# Selectively exclude scripts/ subdirs not needed in container builds.
+# scripts/ci/third-party-licenses.sh + render-node-licenses.mjs are
+# required by the convenience-binary Dockerfiles to render LICENSE-binary
+# at build time.
+/scripts/dashboard
+/scripts/performance
+/scripts/*.sh
+/scripts/ci/*
+!/scripts/ci/third-party-licenses.sh
+!/scripts/ci/render-node-licenses.mjs
+**/target
+/target
+/web/node_modules
+/web/dist
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index f6a74231e..e8841b8d6 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -45,6 +45,104 @@ Default bind: `127.0.0.1:9093`. Environment variables:
| `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 |
+## Quick start (Docker Compose)
+
+The fastest way to see a Kafka client talk to Iggy: two containers, no local
Rust toolchain to
+build them. (Reading the record back below uses `docker compose exec`, not a
local build either.)
+
+```text
+ Kafka wire (TCP 9093) Iggy wire (TCP 8090)
+ kcat / librdkafka ─────────────────────▶ gateway ─────────────────────▶
iggy-server
+ (your Kafka client) iggy-gateway-kafka
(real broker)
+```
+
+```bash
+cd gateways/kafka
+docker compose up --build
+```
+
+`iggy-server` pulls the prebuilt `apache/iggy:edge` image (refreshed on a
master push only when `server` or `iggy-cli` is affected, or `web` or
`core/server/Dockerfile` changes),
+so only `iggy-gateway-kafka` compiles from source. Its Dockerfile mounts a
cargo registry/target
+cache, so a first run costs one real compile and every run after reuses it,
even across a source
+change elsewhere in the workspace. Testing a local `iggy-server` change:
comment out `image:` and
+uncomment `build:` in `docker-compose.yml` first. That build mounts cargo
registry, git, and
+target caches, so the first run is still a full release build of several
minutes and later runs
+reuse those mounts.
+
+This starts `iggy-server` (the real Iggy broker) and `iggy-gateway-kafka`
(bridge enabled,
+pointed at that server) on a shared Docker network, with the gateway waiting
for the server's
+health check before it starts. From another terminal:
+
+```bash
+# Broker discovery works with just kcat - it has no create-topic mode of its
own, though.
+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
+automated test, so adjust if the library's API has moved since this was
written):
+
+```python
+from kafka.admin import KafkaAdminClient, NewTopic
+from kafka import KafkaProducer
+
+admin = KafkaAdminClient(bootstrap_servers="127.0.0.1:9093")
+admin.create_topics([NewTopic(name="orders", num_partitions=1,
replication_factor=1)])
+
+producer = KafkaProducer(bootstrap_servers="127.0.0.1:9093")
+producer.send("orders", b"hello from kafka-python").get(timeout=5)
+```
+
+Read the record back with a real Kafka consumer - Fetch is wired to Iggy with
the bridge on (see
+the stub warning above), so `kcat` reads it the same way it would against a
real broker:
+
+```bash
+kcat -b 127.0.0.1:9093 -C -t orders -o beginning -e
+```
+
+`-e` exits once `kcat` catches up rather than waiting for more. No consumer
group needed for a
+one-shot read; see [docs/CONSUMER_GROUPS.md](docs/CONSUMER_GROUPS.md) for
group-based consuming.
+Polling Iggy directly still works too, if you want to see the record from the
other side:
+
+```bash
+docker compose exec iggy-server /usr/local/bin/iggy --username iggy \
+ --password iggy-gateway-quickstart message poll kafka orders 0 --offset 0
--message-count 1
+```
+
+**Ports**: gateway `9093` (Kafka wire, the one a client dials), Iggy `8090`
(TCP, `iggy` CLI/SDK),
+`3000`/`8080`/`8092` (Iggy HTTP/QUIC/WebSocket, unused by the gateway itself
but published for
+direct Iggy inspection).
+
+**Config**: the compose file sets fixed dev-only root credentials
(`IGGY_ROOT_PASSWORD`) so the
+gateway's `IGGY_KAFKA_IGGY_PASSWORD` is known ahead of time - never reuse them
outside a local
+stack. See `gateways/kafka/docker-compose.yml` and the environment variable
tables below for
+every other knob.
+
+**Limitations** (see the stub warning above for the full list): OffsetCommit
and OffsetFetch are
+not implemented yet, so a consumer group can form and get assigned partitions
but can't commit or
+resume from a saved offset - use `assign()` with an explicit start offset, as
the Fetch example
+above does with `-o beginning`. Single gateway, single Iggy node, SASL off by
default. This is a
+development quick start, not a production deployment shape.
+
+This quick start's topics use Iggy's default `Durability::Replicated` - a
Produce ack is a
+quorum commit, not an fsync. A `docker kill` or host power loss can lose acked
records despite
+the named Docker volume; the volume protects a clean `docker compose
down`/`up`, not a crash.
+Neither this quick start nor `phase1_e2e_tests.rs` exercises crash durability
- the automated
+test and the manual read-back above both only prove a produced record is
readable from the
+still-live process that wrote it. `docs/MANUAL_TESTING.md`'s Category I covers
a *restart* (a
+graceful `docker compose restart`, which flushes on the way down) as a smoke
test for "does the
+volume actually persist across a container restart" - it is not a crash test
and does not
+contradict this limitation.
+
+**Success criteria** (tracked for
[#3539](https://github.com/apache/iggy/issues/3539)): a new
+contributor should reach a successful produce (and an Iggy-side read-back
confirming it landed)
+within 15 minutes of `git clone`, and every Phase 1 API key (CreateTopics,
Metadata, Produce,
+ListOffsets) has an automated test exercising the real compiled binaries end
to end -
+`tests/phase1_e2e_tests.rs` is that test, spawning real `iggy-server` and
`iggy-gateway-kafka`
+processes (not `KafkaGateway::run` in-process, as every other suite in this
crate does) and
+driving CreateTopics → Metadata → Produce → ListOffsets over real TCP, then
reading the produced
+records back through the Iggy SDK.
+
## Test
```bash
diff --git a/gateways/kafka/docker-compose.yml
b/gateways/kafka/docker-compose.yml
new file mode 100644
index 000000000..af1c11a03
--- /dev/null
+++ b/gateways/kafka/docker-compose.yml
@@ -0,0 +1,141 @@
+# 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.
+
+# Quick-start: a real iggy-server plus the Kafka gateway bridged to it, for a
Kafka client to
+# talk to end to end. `docker compose up` from this directory. See README.md's
"Quick start".
+
+services:
+ iggy-server:
+ # Prebuilt. :edge refreshes on a master push only when `server` or
`iggy-cli` is
+ # affected, or `web` or `core/server/Dockerfile` changes - avoids a full
from-scratch workspace
+ # release build on first `up --build`. Testing a local server change:
comment this out
+ # and uncomment `build:` below instead - `core/server/Dockerfile`'s
`runtime` target is
+ # the same Dockerfile and target `apache/iggy:edge` itself is published
from
+ # (.github/config/publish.yml), so the healthcheck/volume/listener-address
settings below
+ # stay correct either way.
+ image: apache/iggy:edge
+ # :edge is a rolling tag - without this, compose reuses whatever was
pulled last time it
+ # happened to not be cached locally, and a stale server can reject a newer
gateway's login
+ # once the wire protocol moves. Always check for a newer edge instead of
trusting the cache.
+ pull_policy: always
+ # build:
+ # context: ../..
+ # dockerfile: core/server/Dockerfile
+ # target: runtime
+ restart: unless-stopped
+ cap_add:
+ - SYS_NICE
+ security_opt:
+ - seccomp:unconfined
+ ulimits:
+ memlock:
+ soft: -1
+ hard: -1
+ environment:
+ - IGGY_NODE_ADVERTISED_ADDRESS=localhost
+ # Fixed, non-default root credentials: this quick-start's gateway needs
a password it
+ # knows ahead of time, and iggy-server otherwise generates a random one
per start.
+ # Dev-only - never reuse these outside a local compose stack.
+ - IGGY_ROOT_USERNAME=iggy
+ - IGGY_ROOT_PASSWORD=iggy-gateway-quickstart
+ # config.toml's own defaults are all 127.0.0.1 - fine for a single
binary on the host,
+ # not reachable from a sibling container. The gateway dials
iggy-server:8090 over this
+ # compose network, so TCP at minimum has to bind the wildcard address;
the other three
+ # are overridden too since this file also publishes them to the host
(see bdd's own
+ # docker-compose.server.yml for the same pattern against the same image).
+ - IGGY_TCP_ADDRESS=0.0.0.0:8090
+ - IGGY_HTTP_ADDRESS=0.0.0.0:3000
+ - IGGY_QUIC_ADDRESS=0.0.0.0:8080
+ - IGGY_WEBSOCKET_ADDRESS=0.0.0.0:8092
+ networks:
+ - iggy-kafka-gateway
+ ports:
+ - 127.0.0.1:3000:3000
+ - 127.0.0.1:8080:8080
+ - 127.0.0.1:8090:8090
+ - 127.0.0.1:8092:8092
+ volumes:
+ # apache/iggy:edge (core/server/Dockerfile's
`runtime`/`runtime-prebuilt` targets) sets
+ # WORKDIR /app; config.toml's own `path = "local_data"` is relative to
that, not to `/`.
+ - iggy-server:/app/local_data
+ # This gates the gateway's start on the server actually answering a
command, not just
+ # accepting a TCP connect (a bare connect succeeds before the
request-processing loop is
+ # wired up; `ping` is unauthenticated - see
core/cli/src/commands/binary_system/ping.rs's
+ # `login_required() -> false` - so this still doesn't prove root-user
login is ready, only
+ # that the server itself is). Root is seeded during metadata recovery,
before the TCP
+ # listeners bind (core/server/src/boot/mod.rs), so there is no "server up,
root not ready"
+ # window for the gateway to race against - RECONNECTION_RETRIES in
bridge/iggy_bridge/mod.rs
+ # (3 retries after the first attempt, 4 attempts total) exists for
transient connect
+ # failures, not for waiting out root provisioning.
+ healthcheck:
+ test: ["CMD", "/usr/local/bin/iggy", "ping"]
+ # start_interval covers startup (forks a process + fresh TCP connect
every 2s only until
+ # healthy); interval is the steady-state cadence for the rest of the
container's life, so
+ # it stays slow enough not to matter.
+ start_interval: 2s
+ interval: 30s
+ timeout: 3s
+ start_period: 10s
+ retries: 30
+
+ iggy-gateway-kafka:
+ build:
+ context: ../..
+ dockerfile: gateways/kafka/Dockerfile
+ container_name: iggy-gateway-kafka
+ restart: unless-stopped
+ depends_on:
+ iggy-server:
+ condition: service_healthy
+ environment:
+ - IGGY_KAFKA_BIND_ADDR=0.0.0.0:9093
+ # Binding the wildcard address above means the gateway cannot derive its
own advertised
+ # host (server.rs::BrokerAdvertise::from_server_config refuses to start
without this) -
+ # same requirement root docker-compose.yml's
IGGY_NODE_ADVERTISED_ADDRESS already carries
+ # for iggy-server, for the same reason: the value a Kafka client's
Metadata response
+ # points it back at has to be reachable from outside the container.
+ #
+ # Literal 127.0.0.1, not "localhost": the port below is published on
127.0.0.1 only, and
+ # a client resolving "localhost" to ::1 first (common default on many
systems) would get
+ # a Metadata response pointing it at an address nothing is actually
listening on.
+ - IGGY_KAFKA_ADVERTISED_HOST=127.0.0.1
+ - IGGY_KAFKA_BRIDGE_ENABLED=true
+ - IGGY_KAFKA_IGGY_ADDR=iggy-server:8090
+ - IGGY_KAFKA_IGGY_USERNAME=iggy
+ - IGGY_KAFKA_IGGY_PASSWORD=iggy-gateway-quickstart
+ networks:
+ - iggy-kafka-gateway
+ ports:
+ - 127.0.0.1:9093:9093
+ # `depends_on: service_healthy` only gates the initial `up`, never a later
restart - without
+ # its own healthcheck this container crash-loops invisibly (no state
`docker compose ps` can
+ # show) if restarted while iggy-server happens to be down. The listener
binds only after the
+ # bridge connects (src/main.rs), so "port answers" already implies "bridge
is up".
+ healthcheck:
+ test: ["CMD", "nc", "-z", "127.0.0.1", "9093"]
+ interval: 10s
+ timeout: 3s
+ start_period: 15s
+ retries: 5
+
+volumes:
+ iggy-server:
+ driver: local
+
+networks:
+ iggy-kafka-gateway:
+ name: iggy-kafka-gateway
diff --git a/gateways/kafka/docs/MANUAL_TESTING.md
b/gateways/kafka/docs/MANUAL_TESTING.md
index ea4c1f668..3df52b832 100644
--- a/gateways/kafka/docs/MANUAL_TESTING.md
+++ b/gateways/kafka/docs/MANUAL_TESTING.md
@@ -297,9 +297,11 @@ for CreateTopics, and so on).
| F5 | Oversized frame | Send 4-byte length > 8 MiB | Connection
rejected/closed, no OOM |
| F6 | Graceful shutdown | Ctrl+C on gateway | Log "shutdown requested",
in-flight requests drain |
-### Category G — Real Kafka client (kcat)
+### Category G — Real Kafka client (kcat, Java console consumer)
-Requires `kcat` installed. Gateway does **not** implement SASL or full broker
semantics — expect limited success.
+Requires `kcat` installed. SASL is off by default
(`IGGY_KAFKA_SASL_ENABLED=false` - see Category S
+for the SASL-on flow) and the gateway does not implement full broker semantics
— expect limited
+success. This category predates the bridge landing: G2/G3's "fails at
metadata" framing assumes an unbridged gateway, and no longer holds with
`IGGY_KAFKA_BRIDGE_ENABLED=true` against a topic that was actually created
first. See Category I for the bridged, docker-compose flow.
| ID | Test | Command | Expected (foundation) |
| ---- | ------ | --------- | --------------------- |
@@ -325,6 +327,32 @@ Record kcat version and exact error strings in your test
log. G1 passing is the
| H5 | Transactional Produce | Send key 0 v3 with a non-null
`transactional_id` and `acks=1` | `ec=35` per partition, **not** `ec=6`;
connection stays open. With `acks=0`: no response, and the gateway closes the
connection |
| H6 | Transaction API keys | `send --host 127.0.0.1:9093 --api-key 24` (also
25, 26, 28) | Connection closes, no response bytes - they are never advertised |
+### Category I — Docker Compose quick start
([#3539](https://github.com/apache/iggy/issues/3539))
+
+Exercises the real compiled binaries the way an operator actually runs them -
not `cargo run`
+against a locally-built binary, and not `KafkaGateway::run` in-process the way
every automated
+suite except `tests/phase1_e2e_tests.rs` does.
+
+| ID | Test | Steps | Expected |
+| ---- | ------ | ------- | ---------- |
+| I1 | Stack comes up | `cd gateways/kafka && docker compose up --build` |
Both containers start; gateway does not exit (it would if the bridge connect
failed before the server's health check passed) |
+| I2 | Broker discovery | `kcat -b 127.0.0.1:9093 -L` | Broker listed,
reachable |
+| I3 | Create + produce | Run the `kafka-python` snippet from README.md's
Quick start | `create_topics` and `send(...).get()` both return without raising
|
+| I4 | Read-back via Kafka Fetch | `kcat -b 127.0.0.1:9093 -C -t orders -o
beginning -e` | The produced record's payload is printed - validates the actual
Kafka Fetch path, not just Iggy's own side |
+| I5 | Volume persistence across a restart | `docker compose restart
iggy-server`, wait for its health check to go healthy again, re-run I4's poll
command | The same record is still readable after the container restarts
against the same named volume |
+| I6 | Teardown | `docker compose down -v` | Both containers stop; volume
removed |
+
+I5 is a graceful restart (`docker compose restart` sends SIGTERM, which
flushes on the way down) -
+it checks that the named volume actually persists data across a container
restart, not that a
+produced record survives an unclean crash. It has no automated equivalent yet.
+`tests/phase1_e2e_tests.rs` and every step above I5 only prove
read-your-own-write on a live
+process (see README.md's Limitations, which also covers why this quick start's
default topic
+durability is not crash-safe). Treat an I5 failure as a real regression, not a
flake.
+
+Record wall-clock time from `git clone` to I3 passing - README.md's Quick
start states a 15-minute
+target for a new contributor
([#3539](https://github.com/apache/iggy/issues/3539) acceptance
+criteria).
+
---
## 4. Validation reference
@@ -414,8 +442,9 @@ kcat version (if used): ___________
[ ] D1–D10 Flexible vs legacy encoding
[ ] E1–E4 Metadata stub semantics
[ ] F1–F6 TCP / connection behavior
-[ ] G1–G8 Real clients (record errors for G2/G3)
+[ ] G1–G8 kcat / Java client (record errors for G2/G3)
[ ] H1–H6 Adversarial input
+[ ] I1–I6 Docker Compose quick start (record wall-clock time to I3; I5 is
restart durability)
Automated regression:
[ ] cargo test -p iggy-gateway-kafka — all passed (see `TEST_SUITE.md` for why
this checklist
diff --git a/gateways/kafka/docs/TEST_SUITE.md
b/gateways/kafka/docs/TEST_SUITE.md
index 4153ec8c2..e2a006958 100644
--- a/gateways/kafka/docs/TEST_SUITE.md
+++ b/gateways/kafka/docs/TEST_SUITE.md
@@ -36,6 +36,18 @@ in the same target directory; neither harness invokes `cargo
build` itself). Ski
fails with a clear "binary not found" message naming the build command to run,
not a hang or a
silent skip.
+### `iggy-gateway-kafka` binary (required for `phase1_e2e_tests`)
+
+`phase1_e2e_tests` spawns the compiled gateway itself, bridged to a spawned
`iggy-server`, so
+`iggy-server` still needs the manual build step above. `iggy-gateway-kafka`
does not: it is this
+package's own `[[bin]]`, so `CARGO_BIN_EXE_iggy-gateway-kafka` resolves at
compile time and Cargo
+builds it automatically as part of `cargo test -p iggy-gateway-kafka`.
+
+```bash
+cargo build --package server --bin iggy-server
+cargo test -p iggy-gateway-kafka
+```
+
---
## Test files
@@ -71,9 +83,10 @@ file under `tests/` anymore.
|
[`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, codes 3 and 43 | 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) |
| [`produce_real_bridge_tests.rs`](../tests/produce_real_bridge_tests.rs) |
Produce (key 0) through the whole handler against a real, spawned `iggy-server`
— records go in as Kafka wire bytes and come back through the Iggy SDK, plus
one error code per partition | No (needs the `iggy-server` binary - see
Prerequisites) |
+| [`phase1_e2e_tests.rs`](../tests/phase1_e2e_tests.rs) |
[#3539](https://github.com/apache/iggy/issues/3539) acceptance: CreateTopics →
Metadata → Produce → ListOffsets over real TCP against the real compiled
`iggy-gateway-kafka` binary bridged to a real spawned `iggy-server` - the only
suite that spawns the gateway as a process rather than calling
`KafkaGateway::run` in-process | No (needs both binaries - see Prerequisites) |
-`tests/common/` holds shared helpers (`codec.rs`, `fixtures.rs`, `scope.rs`,
`server.rs`,
-`iggy_server.rs`, `tcp.rs`, `wire.rs`), compiled per test binary via
`#[path]`, not a test binary itself. `codec.rs`
+`tests/common/` holds shared helpers (`codec.rs`, `fixtures.rs`,
`gateway_process.rs`, `scope.rs`,
+`server.rs`, `iggy_server.rs`, `tcp.rs`, `wire.rs`), compiled per test binary
via `#[path]`, not a test binary itself. `codec.rs`
is test-only primitive encode/decode scaffolding for hand-building
legacy/adversarial wire shapes
`kafka_protocol`'s spec-correct encoder cannot produce - it is not the
gateway's production codec.
diff --git a/gateways/kafka/tests/common/gateway_process.rs
b/gateways/kafka/tests/common/gateway_process.rs
new file mode 100644
index 000000000..b7e3f70e6
--- /dev/null
+++ b/gateways/kafka/tests/common/gateway_process.rs
@@ -0,0 +1,102 @@
+// 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.
+
+//! Real `iggy-gateway-kafka` process harness, mirroring `iggy_server.rs`'s
`TestServer` for the
+//! gateway binary itself. Every existing "e2e" suite in this crate drives
`KafkaGateway::run`
+//! in-process (`common/server.rs`), never the compiled binary an operator
actually runs - this
+//! type closes that gap for #3539's docker-compose-shaped end-to-end test.
+
+#![allow(dead_code)]
+
+use std::path::PathBuf;
+use std::process::{Child, Command};
+
+use crate::iggy_server::{PortGuard, TestServer, graceful_kill,
wait_for_listener};
+use std::env;
+
+/// Locates the already-built `iggy-gateway-kafka` binary.
+///
+/// Unlike `iggy_server.rs`'s `iggy_server_binary()`, this does not need an
ancestor walk from
+/// `env::current_exe()`: that workaround exists because `iggy-server` lives
in a *different*
+/// package, where `CARGO_BIN_EXE_*` never resolves. `iggy-gateway-kafka` is
this package's own
+/// `[[bin]]` (`Cargo.toml`), so `CARGO_BIN_EXE_iggy-gateway-kafka` is set at
compile time and
+/// Cargo builds the binary itself before running this test - same pattern
already used in
+/// `gateways/kafka/tools/kafka-tool/tests/generate_cli_tests.rs`.
+fn iggy_gateway_binary() -> PathBuf {
+ PathBuf::from(env!("CARGO_BIN_EXE_iggy-gateway-kafka"))
+}
+
+pub struct TestGateway {
+ child: Child,
+ pub address: String,
+ _port_guard: PortGuard,
+}
+
+impl TestGateway {
+ /// Spawns `iggy-gateway-kafka` with the bridge enabled against `server`,
then blocks until
+ /// its listener is ready or the startup budget is exhausted.
+ pub async fn spawn(server: &TestServer) -> Self {
+ let port_guard = PortGuard::acquire();
+ let address = format!("127.0.0.1:{}", port_guard.port);
+ let config = server.test_config();
+
+ let mut command = Command::new(iggy_gateway_binary());
+ // env_clear(): without it the child inherits this test process's full
environment, and
+ // the gateway's own reject_unknown_kafka_env_vars (main.rs)
hard-fails startup on any
+ // unrecognized inherited IGGY_KAFKA_* var - or, worse, a *recognized*
one (say
+ // IGGY_KAFKA_ADVERTISED_HOST, which this crate's own
README/docker-compose.yml teach a
+ // contributor to export) silently changes the spawned gateway's
behavior out from under
+ // this harness's assertions.
+ command.env_clear();
+ if let Some(llvm_profile_file) = env::var_os("LLVM_PROFILE_FILE") {
+ command.env("LLVM_PROFILE_FILE", llvm_profile_file);
+ }
+ command
+ .env("IGGY_KAFKA_BIND_ADDR", &address)
+ .env("IGGY_KAFKA_BRIDGE_ENABLED", "true")
+ .env("IGGY_KAFKA_IGGY_ADDR", &config.address)
+ .env("IGGY_KAFKA_IGGY_USERNAME", &config.username)
+ .env(
+ "IGGY_KAFKA_IGGY_PASSWORD",
+ secrecy::ExposeSecret::expose_secret(&config.password),
+ );
+ let child = command.spawn().expect("spawn iggy-gateway-kafka");
+
+ let mut gateway = Self {
+ child,
+ address,
+ _port_guard: port_guard,
+ };
+ gateway.wait_ready().await;
+ gateway
+ }
+
+ /// Bare TCP connect poll, same rationale as `TestServer::wait_ready`: a
full Kafka client
+ /// handshake would pay retry overhead on every failed attempt. A connect
slightly ahead of
+ /// the gateway's own accept loop can't happen here - `main.rs` connects
the bridge before
+ /// binding the listener, so a successful connect already implies the
bridge is up too.
+ async fn wait_ready(&mut self) {
+ wait_for_listener(&mut self.child, &self.address,
"iggy-gateway-kafka").await;
+ }
+}
+
+impl Drop for TestGateway {
+ fn drop(&mut self) {
+ graceful_kill(&mut self.child);
+ let _ = self.child.wait();
+ }
+}
diff --git a/gateways/kafka/tests/common/iggy_server.rs
b/gateways/kafka/tests/common/iggy_server.rs
index b8bef7c73..51cc9a746 100644
--- a/gateways/kafka/tests/common/iggy_server.rs
+++ b/gateways/kafka/tests/common/iggy_server.rs
@@ -23,6 +23,7 @@
#![allow(dead_code)]
use std::collections::HashMap;
+use std::env;
use std::fs::{File, OpenOptions};
use std::path::PathBuf;
use std::process::{Child, Command};
@@ -146,9 +147,18 @@ fn
given_a_ceiling_at_the_u16_maximum_when_choosing_a_band_should_refuse() {
}
/// Exclusive claim on one port, released (and the port freed for reuse) when
dropped - including
-/// on an unclean process exit, since the OS drops the `flock` with the file
descriptor. Unlike
-/// bind-then-drop, this process itself never binds the port before
`iggy-server` does; see
-/// `port_band`'s doc comment for the limits of that guarantee against *other*
processes.
+/// on an unclean process exit, since the OS drops the `flock` with the file
descriptor.
+///
+/// `acquire` also bind-probes the port (bind, then drop the listener with no
accepted
+/// connections, so nothing lingers in `TIME_WAIT`) before handing it out. The
flock alone only
+/// excludes another `PortGuard`-based process; it does nothing against a
process from a *prior*
+/// test run that's still holding the port (killed hard enough to skip `Drop`,
so its own flock
+/// died with it but its listener didn't). A bind-probe catches that case at
acquisition time
+/// instead of leaving callers to detect a stale listener after the fact -
detecting it after
+/// the fact can't work on the first readiness-poll iteration, before the real
child has even
+/// attempted its own `bind()` yet, so a successful connect this early is
ambiguous between
+/// "ours, racing ahead of its own listen" and "a stale orphan's". See
`port_band`'s doc comment
+/// for the limits of this guarantee against a `bind(0)` from an unrelated
process.
pub struct PortGuard {
pub port: u16,
_lock: File,
@@ -171,7 +181,10 @@ impl PortGuard {
else {
continue;
};
- if file.try_lock().is_ok() {
+ // A stale listener (not another PortGuard user, the flock already
excludes those)
+ // can still hold this port - skip it on a failed bind, leave the
lock file for the
+ // next acquire to try again, keep scanning.
+ if file.try_lock().is_ok() &&
std::net::TcpListener::bind(("127.0.0.1", port)).is_ok() {
return Self { port, _lock: file };
}
}
@@ -250,6 +263,13 @@ impl TestServer {
let address = format!("127.0.0.1:{}", port_guard.port);
let mut command = Command::new(iggy_server_binary());
+ // env_clear(): without it the child inherits this test process's full
environment, and
+ // a stray recognized IGGY_* var could silently change the spawned
server's behavior out
+ // from under this harness's assertions (same reasoning as
gateway_process.rs's spawn).
+ command.env_clear();
+ if let Some(llvm_profile_file) = env::var_os("LLVM_PROFILE_FILE") {
+ command.env("LLVM_PROFILE_FILE", llvm_profile_file);
+ }
command
.env("IGGY_PATH", data_dir.display().to_string())
.env("IGGY_TCP_ADDRESS", &address)
@@ -293,40 +313,15 @@ impl TestServer {
}
/// Polls with a bare TCP connect, not a full `IggyBridge::connect`: the
latter carries
- /// `RECONNECTION_RETRIES` (3 dials at ~1s apart) on every failed attempt,
so a poll loop built
+ /// `RECONNECTION_RETRIES` (3 retries after the first attempt, 4 dials at
~1s apart total) on
+ /// every failed attempt, so a poll loop built
/// on it pays several real seconds per iteration instead of running at
its own 100ms cadence,
/// and a *successful* poll iteration would authenticate a client and then
drop it without
/// `close()`, leaking a session server-side. A TCP accept slightly ahead
of the app being
/// ready to authenticate is fine - every caller's own subsequent
`IggyBridge::connect` already
/// retries a few times, which covers the last few hundred ms of that gap.
- ///
- /// Checks `try_wait()` every iteration: a server that fails at boot (bad
config, a port
- /// stolen between `PortGuard::acquire` and its own bind) exits almost
immediately, and without
- /// this check that reads as "still starting" for the full 30s budget -
the eventual failure
- /// names the wrong cause ("did not become ready") instead of the real one
(exited early, with
- /// whatever it printed before dying).
async fn wait_ready(&mut self) {
- let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
- loop {
- if let Some(status) = self.child.try_wait().expect("poll child
status") {
- panic!(
- "iggy-server at {} exited during startup with {status}",
- self.address
- );
- }
- if tokio::net::TcpStream::connect(self.address.as_str())
- .await
- .is_ok()
- {
- return;
- }
- assert!(
- tokio::time::Instant::now() < deadline,
- "iggy-server at {} did not become ready within the startup
budget",
- self.address
- );
- tokio::time::sleep(Duration::from_millis(100)).await;
- }
+ wait_for_listener(&mut self.child, &self.address, "iggy-server").await;
}
pub fn test_config(&self) -> IggyBridgeConfig {
@@ -352,7 +347,32 @@ impl TestServer {
const SIGTERM_TIMEOUT: Duration = Duration::from_secs(5);
const SIGKILL_POLL_INTERVAL: Duration = Duration::from_millis(50);
-fn graceful_kill(child: &mut Child) {
+/// Shared by `TestServer::wait_ready` and `TestGateway::wait_ready`
(`gateway_process.rs`) -
+/// identical poll loop, only the process label in the panic/assert messages
differs.
+///
+/// Checks `try_wait()` every iteration: a process that fails at boot (bad
config, a port stolen
+/// between `PortGuard::acquire` and its own bind) exits almost immediately,
and without this
+/// check that reads as "still starting" for the full 30s budget - the
eventual failure names the
+/// wrong cause ("did not become ready") instead of the real one (exited
early, with whatever it
+/// printed before dying).
+pub async fn wait_for_listener(child: &mut Child, address: &str, label: &str) {
+ let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
+ loop {
+ if let Some(status) = child.try_wait().expect("poll child status") {
+ panic!("{label} at {address} exited during startup with {status}");
+ }
+ if tokio::net::TcpStream::connect(address).await.is_ok() {
+ return;
+ }
+ assert!(
+ tokio::time::Instant::now() < deadline,
+ "{label} at {address} did not become ready within the startup
budget"
+ );
+ tokio::time::sleep(Duration::from_millis(100)).await;
+ }
+}
+
+pub fn graceful_kill(child: &mut Child) {
// `wait_ready` calls `try_wait()` too (and panics on an early exit,
unwinding into this via
// `Drop`) - if it already reaped the child, `child.id()` is a PID the OS
is free to hand to an
// unrelated process by the time we get here, and a raw `libc::kill`
(unlike `std::Child::kill`,
diff --git a/gateways/kafka/tests/phase1_e2e_tests.rs
b/gateways/kafka/tests/phase1_e2e_tests.rs
new file mode 100644
index 000000000..a03aa5f64
--- /dev/null
+++ b/gateways/kafka/tests/phase1_e2e_tests.rs
@@ -0,0 +1,341 @@
+// 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.
+
+//! #3539 acceptance test: the whole Phase 1 flow (`CreateTopics` ->
`Metadata` -> `Produce` ->
+//! `ListOffsets`) driven over real TCP against the real compiled
`iggy-gateway-kafka` binary,
+//! itself bridged to a real compiled `iggy-server` binary - the same two
processes
+//! `docker-compose.yml` wires together, spawned directly as host binaries
rather than through
+//! Docker Compose itself. Not `KafkaGateway::run` spawned in-process the way
every other "e2e"
+//! suite in this crate does (`common/server.rs`). A regression only the
actual binaries'
+//! startup, env parsing, or process wiring can produce (as opposed to a
handler-level bug the
+//! in-process suites already cover) fails here and nowhere else.
+//!
+//! This does **not** exercise `docker-compose.yml`, the `Dockerfile`s, image
builds, container
+//! networking, or the health-check gate - those are covered by
`docs/MANUAL_TESTING.md`'s
+//! Category I (manual) only, not by any automated suite. A broken compose
file, a missing
+//! runtime library in the image, or a wrong port/network mapping would not be
caught here.
+//!
+//! The final read-back step goes through the Iggy SDK directly rather than a
real Kafka Fetch,
+//! the same substitute `produce_real_bridge_tests.rs` uses - Fetch is
implemented
+//! (`src/protocol/handlers/fetch.rs`, exercised by
`fetch_real_bridge_tests.rs`), but wiring a
+//! hand-built Fetch request/response codec into this specific acceptance test
is left as a
+//! follow-up rather than bundled into this PR's scope.
+
+use std::net::SocketAddr;
+
+use bytes::{Bytes, BytesMut};
+use iggy::prelude::{Consumer, Identifier, MessageClient, PollingStrategy};
+use kafka_protocol::indexmap::IndexMap;
+use kafka_protocol::records::{
+ Compression, NO_PARTITION_LEADER_EPOCH, NO_PRODUCER_EPOCH, NO_PRODUCER_ID,
NO_SEQUENCE, Record,
+ RecordBatchEncoder, RecordEncodeOptions, TimestampType,
+};
+
+use iggy_gateway_kafka::protocol::api::{
+ API_KEY_CREATE_TOPICS, API_KEY_LIST_OFFSETS, API_KEY_METADATA,
API_KEY_PRODUCE, ERROR_NONE,
+};
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/gateway_process.rs"]
+mod gateway_process;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+#[path = "common/tcp.rs"]
+mod tcp;
+#[path = "common/wire.rs"]
+mod wire;
+
+use codec::{Decoder, Encoder};
+use gateway_process::TestGateway;
+use iggy_server::{TestServer, raw_client};
+use tcp::round_trip;
+
+const TOPIC: &str = "phase1-e2e-topic";
+const STREAM: &str = "kafka";
+const CREATE_TOPICS_VERSION: i16 = 5;
+const METADATA_VERSION: i16 = 9;
+const PRODUCE_VERSION: i16 = 7;
+const LIST_OFFSETS_VERSION: i16 = 6;
+const CREATE_TIME: i64 = 1_700_000_000_123;
+
+fn build_create_topics_request(topic: &str, num_partitions: i32) -> Bytes {
+ let mut enc = Encoder::with_capacity(128);
+ enc.write_varint(2); // one topic (compact array N+1)
+ enc.write_compact_nullable_string(Some(topic));
+ enc.write_i32(num_partitions);
+ enc.write_i16(1); // replication_factor
+ enc.write_varint(1); // no explicit assignments
+ enc.write_varint(1); // no configs
+ enc.write_empty_tagged_fields(); // topic tagged fields
+ enc.write_i32(5_000); // timeout_ms
+ enc.write_bool(false); // validate_only
+ enc.write_empty_tagged_fields(); // request tagged fields
+ enc.freeze()
+}
+
+/// `(error_code, error_message, num_partitions)` from the first (only) topic
result.
+fn decode_create_topics_response(body: Bytes) -> (i16, Option<String>, i32) {
+ let mut d = Decoder::new(body);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ let _topics_plus_one = d.read_varint().expect("topics array count");
+ let _name = d.read_compact_nullable_string().expect("topic name");
+ let error_code = d.read_i16().expect("error_code");
+ let error_message =
d.read_compact_nullable_string().expect("error_message");
+ let num_partitions = d.read_i32().expect("num_partitions");
+ (error_code, error_message, num_partitions)
+}
+
+/// `(topic error_code, partition error_code, leader_id of partition 0,
node_ids in the brokers
+/// array, controller_id, first broker's host, first broker's port)` for the
single requested
+/// topic.
+#[allow(clippy::type_complexity)]
+fn decode_metadata_response(body: Bytes) -> (i16, i16, i32, Vec<i32>, i32,
Option<String>, i32) {
+ let mut d = Decoder::new(body);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+
+ let brokers_plus_one = d.read_varint().expect("brokers array count");
+ assert!(brokers_plus_one >= 2, "at least one broker in the list");
+ let mut broker_host = None;
+ let mut broker_port = 0;
+ let mut node_ids = Vec::new();
+ for i in 1..brokers_plus_one {
+ let node_id = d.read_i32().expect("node_id");
+ let host = d.read_compact_nullable_string().expect("host");
+ let port = d.read_i32().expect("port");
+ d.read_compact_nullable_string().expect("rack");
+ d.read_tagged_fields().expect("broker tagged fields");
+ node_ids.push(node_id);
+ if i == 1 {
+ broker_host = host;
+ broker_port = port;
+ }
+ }
+ d.read_compact_nullable_string().expect("cluster_id");
+ let controller_id = d.read_i32().expect("controller_id");
+
+ let topics_plus_one = d.read_varint().expect("topics array count");
+ assert_eq!(topics_plus_one, 2, "exactly one topic was requested");
+ let error_code = d.read_i16().expect("topic error_code");
+ d.read_compact_nullable_string().expect("topic name");
+ d.read_bool().expect("is_internal");
+
+ let partitions_plus_one = d.read_varint().expect("partitions array count");
+ assert_eq!(partitions_plus_one, 2, "one partition was created");
+ let partition_error_code = d.read_i16().expect("partition error_code");
+ d.read_i32().expect("partition_index");
+ let leader_id = d.read_i32().expect("leader_id");
+ (
+ error_code,
+ partition_error_code,
+ leader_id,
+ node_ids,
+ controller_id,
+ broker_host,
+ broker_port,
+ )
+}
+
+fn assert_metadata_response(body: Bytes, gateway_addr: SocketAddr) {
+ let (
+ error_code,
+ partition_error_code,
+ leader_id,
+ node_ids,
+ controller_id,
+ broker_host,
+ broker_port,
+ ) = decode_metadata_response(body);
+ assert_eq!(error_code, ERROR_NONE, "Metadata must report the new topic");
+ assert_eq!(
+ partition_error_code, ERROR_NONE,
+ "the single partition must report no error"
+ );
+ assert_eq!(leader_id, 1, "this gateway is the sole broker");
+ assert!(
+ node_ids.contains(&leader_id),
+ "leader_id {leader_id} must name a broker actually present in the
brokers array \
+ {node_ids:?} - a real client cannot route to a leader it was never
told about"
+ );
+ assert_eq!(
+ controller_id, 1,
+ "single-broker cluster: the controller is the same broker"
+ );
+ assert_eq!(
+ broker_host.as_deref(),
+ Some(gateway_addr.ip().to_string().as_str()),
+ "Metadata must advertise the address a client can actually reach this
gateway on"
+ );
+ assert_eq!(
+ broker_port,
+ i32::from(gateway_addr.port()),
+ "Metadata must advertise the port this gateway is actually listening
on"
+ );
+}
+
+fn record(offset: i64, value: &[u8]) -> Record {
+ Record {
+ transactional: false,
+ control: false,
+ delete_horizon: false,
+ partition_leader_epoch: NO_PARTITION_LEADER_EPOCH,
+ producer_id: NO_PRODUCER_ID,
+ producer_epoch: NO_PRODUCER_EPOCH,
+ timestamp_type: TimestampType::Creation,
+ offset,
+ sequence: NO_SEQUENCE + i32::try_from(offset).expect("small offset"),
+ timestamp: CREATE_TIME,
+ key: None,
+ value: Some(Bytes::copy_from_slice(value)),
+ headers: IndexMap::new(),
+ }
+}
+
+fn build_produce_request(topic: &str, records: &[Record]) -> Bytes {
+ let mut batch_buf = BytesMut::new();
+ let options = RecordEncodeOptions {
+ version: 2,
+ compression: Compression::None,
+ };
+ RecordBatchEncoder::encode(&mut batch_buf, records.iter(), &options)
+ .expect("encode record batch");
+
+ let mut enc = Encoder::with_capacity(1024);
+ enc.write_nullable_string(None)
+ .expect("null transactional_id");
+ enc.write_i16(1); // acks: leader
+ enc.write_i32(5_000); // timeout_ms
+ enc.write_i32(1); // one topic
+ enc.write_nullable_string(Some(topic))
+ .expect("topic name fits");
+ enc.write_i32(1); // one partition
+ enc.write_i32(0); // partition_index
+ enc.write_nullable_bytes(Some(&batch_buf))
+ .expect("record batch fits");
+ enc.freeze()
+}
+
+/// `(error_code, base_offset)` from the first (only) partition result.
+fn decode_produce_response(body: Bytes) -> (i16, i64) {
+ let mut d = Decoder::new(body);
+ let _topics = d.read_i32().expect("topics array count");
+ let _name = d.read_nullable_string().expect("topic name");
+ let _partitions = d.read_i32().expect("partitions array count");
+ let _index = d.read_i32().expect("partition_index");
+ let error_code = d.read_i16().expect("error_code");
+ let base_offset = d.read_i64().expect("base_offset");
+ let _log_append_time = d.read_i64().expect("log_append_time");
+ let _log_start_offset = d.read_i64().expect("log_start_offset");
+ (error_code, base_offset)
+}
+
+/// `(error_code, offset)` for the requested partition's latest-offset result.
+fn decode_list_offsets_response(body: Bytes) -> (i16, i64) {
+ let mut d = Decoder::new(body);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ let _topics_plus_one = d.read_varint().expect("topics array count");
+ let _name = d.read_compact_nullable_string().expect("topic name");
+ let _partitions_plus_one = d.read_varint().expect("partitions array
count");
+ let _partition_index = d.read_i32().expect("partition_index");
+ let error_code = d.read_i16().expect("error_code");
+ let _timestamp = d.read_i64().expect("timestamp");
+ let offset = d.read_i64().expect("offset");
+ (error_code, offset)
+}
+
+#[tokio::test]
+async fn
phase1_produce_flow_through_real_gateway_process_and_real_iggy_server() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let gateway = TestGateway::spawn(&server).await;
+ let addr: SocketAddr = gateway.address.parse().expect("gateway address");
+
+ let (corr, body) = round_trip(
+ addr,
+ API_KEY_CREATE_TOPICS,
+ CREATE_TOPICS_VERSION,
+ 1,
+ &build_create_topics_request(TOPIC, 1),
+ )
+ .await;
+ assert_eq!(corr, 1, "correlation id must echo the request");
+ let (error_code, error_message, num_partitions) =
decode_create_topics_response(body);
+ assert_eq!(error_code, ERROR_NONE, "CreateTopics must succeed");
+ assert_eq!(
+ error_message, None,
+ "no error_message on a successful result"
+ );
+ assert_eq!(num_partitions, 1);
+
+ let (corr, body) = round_trip(
+ addr,
+ API_KEY_METADATA,
+ METADATA_VERSION,
+ 2,
+ &wire::build_metadata_flexible_request(&[TOPIC]),
+ )
+ .await;
+ assert_eq!(corr, 2, "correlation id must echo the request");
+ assert_metadata_response(body, addr);
+
+ let records = [record(0, b"phase1-hello"), record(1, b"phase1-world")];
+ let (corr, body) = round_trip(
+ addr,
+ API_KEY_PRODUCE,
+ PRODUCE_VERSION,
+ 3,
+ &build_produce_request(TOPIC, &records),
+ )
+ .await;
+ assert_eq!(corr, 3, "correlation id must echo the request");
+ let (error_code, base_offset) = decode_produce_response(body);
+ assert_eq!(error_code, ERROR_NONE, "Produce must succeed");
+ assert_eq!(base_offset, 0, "first batch into an empty partition");
+
+ let (corr, body) = round_trip(
+ addr,
+ API_KEY_LIST_OFFSETS,
+ LIST_OFFSETS_VERSION,
+ 4,
+ &wire::build_list_offsets_request(LIST_OFFSETS_VERSION, TOPIC, 0),
+ )
+ .await;
+ assert_eq!(corr, 4, "correlation id must echo the request");
+ let (error_code, offset) = decode_list_offsets_response(body);
+ assert_eq!(error_code, ERROR_NONE, "ListOffsets must succeed");
+ assert_eq!(offset, 2, "high watermark after two produced records");
+
+ // Read the produced records back through the Iggy SDK directly rather
than a real Kafka
+ // Fetch - see this file's module doc for why (Fetch itself is
implemented).
+ let raw = raw_client(&server).await;
+ let polled = raw
+ .poll_messages(
+ &Identifier::named(STREAM).expect("stream name"),
+ &Identifier::named(TOPIC).expect("topic name"),
+ Some(0),
+
&Consumer::new(Identifier::named("phase1-e2e-reader").expect("consumer name")),
+ &PollingStrategy::offset(0),
+ 2,
+ false,
+ )
+ .await
+ .expect("poll produced messages")
+ .messages;
+ assert_eq!(polled.len(), 2, "both produced records must be readable");
+ assert_eq!(polled[0].payload.as_ref(), b"phase1-hello");
+ assert_eq!(polled[1].payload.as_ref(), b"phase1-world");
+}
diff --git a/licenserc.toml b/licenserc.toml
index 4957b49f8..0025dec84 100644
--- a/licenserc.toml
+++ b/licenserc.toml
@@ -47,6 +47,7 @@ excludes = [
"**/.gitignore",
"**/.gitattributes",
"**/.dockerignore",
+ "**/Dockerfile.dockerignore",
"**/.helmignore",
"**/.prettierignore",
"**/.editorconfig",
