ryerraguntla commented on code in PR #4245:
URL: https://github.com/apache/iggy/pull/4245#discussion_r4088969551
##########
gateways/kafka/src/auth.rs:
##########
@@ -241,12 +297,35 @@ impl SaslAuthenticator for IggyAuthenticator {
let client = builder.build().map_err(|error| classify(&error))?;
let connected = tokio::time::timeout(VERIFY_TIMEOUT,
client.connect()).await;
+ // The password verify the slot bounds is done. Holding it across the
permission read and
+ // teardown kept a slot busy for up to two more seconds per login, and
after a timed-out
+ // read for as long as the SDK's detached request still ran.
+ drop(slot);
let outcome = match connected {
Err(_elapsed) => Err(AuthError::Unavailable),
Ok(Err(error)) => Err(classify(&error)),
- Ok(Ok(())) => Ok(()),
+ // The login already proved the credentials. Reading the
principal's own record on the
+ // same session is the one extra round trip that lets
`DescribeAcls` answer later
+ // without a second login or a stored password. A user may always
read itself, with no
+ // permission required (`dispatch/authz.rs` exempts a
self-targeted read), so this
+ // cannot fail for want of a grant.
+ Ok(Ok(())) => fetch_permissions(&client,
&credentials.username).await,
};
+ // A timed-out permission read leaves the SDK's detached request task
holding the stream
+ // lock until its own response deadline, and `shutdown` needs that
lock first, so waiting
+ // on it could only burn `TEARDOWN_TIMEOUT` of the caller's
pre-authentication budget.
+ // Dropping the client instead still aborts the heartbeat, which is
the part that would
+ // reconnect. The socket itself lives on in that task until its
deadline, which nothing on
+ // this side of the SDK can shorten.
+ if matches!(outcome, Ok((_, PermissionRead::TimedOut))) {
Review Comment:
TimedOut get_user skips shutdown. Heartbeat ping() first (client.rs:971)
locksteps same stream. 1s cancel leaves SDK spawn (tcp_client.rs:1512) on
stream.lock() until RESPONSE_READ_TIMEOUT 30s. drop(slot) already :303. Iggy
TCP + client-table zombie, unbounded vs 16 auth slots. **Fix**: fetch before
heartbeat; do not timeout-cancel spawned lockstep read.
##########
gateways/kafka/tests/kafka_client_e2e_tests.rs:
##########
@@ -0,0 +1,702 @@
+// 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.
+
+//! End-to-end tests driving **real Kafka clients** against the gateway.
+//!
+//! Every other suite in this crate hand-builds wire frames, which cannot
catch a client
+//! compatibility problem by construction. One already slipped through that
way: advertising
+//! `SaslHandshake` from v1 rather than v0 passed every hand-built test and
made librdkafka report
+//! "SASL Handshake not supported by broker" before it sent anything. These
tests exist so that
+//! class of bug fails in CI instead of during manual testing.
+//!
+//! The stack is real on all three sides: a spawned `iggy-server` process, the
gateway in-process
+//! with a real `IggyAuthenticator`, and a client from a container. They
automate categories S and T
+//! of `docs/MANUAL_TESTING.md`.
+//!
+//! Prerequisites are Docker and an already-built `iggy-server`. Missing
either skips, the way the
+//! wire-fixture suites do, unless `KAFKA_E2E_REQUIRED=1` is set, which turns
a skip into a failure
+//! so a broken CI step cannot leave these silently green.
+
+use std::net::SocketAddr;
+use std::path::PathBuf;
+use std::process::{Child, Command, Stdio};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use iggy_gateway_kafka::GatewayConfig;
+use iggy_gateway_kafka::auth::IggyAuthenticator;
+
+#[path = "common/server.rs"]
+mod server;
+
+use server::spawn_test_server_with_authenticator;
+
+const KCAT_IMAGE: &str = "edenhill/kcat:1.7.1";
+const KAFKA_IMAGE: &str = "apache/kafka:3.9.0";
+const ROOT_USER: &str = "iggy";
+const ROOT_PASSWORD: &str = "iggy";
+/// Password for every non-root principal these tests create.
+const USER_PASSWORD: &str = "s3cretpass";
+
+/// Budget for `iggy-server` to start listening. Generous: it is a cold
process start, and a debug
+/// build on a loaded machine is not quick.
+const SERVER_READY_TIMEOUT: Duration = Duration::from_secs(45);
+
+/// Wall-clock cap on one client container. The ACL test runs five in a row
inside nextest's 300s
+/// kill budget, so a single wedged client must fail on its own terms, well
before that budget
+/// kills the test and hides which step hung.
+const CLIENT_RUN_TIMEOUT: &str = "45s";
+
+/// Per-request and per-call budget for the Java admin tools. Their defaults
(30s and 60s) let one
+/// unanswered request eat most of `CLIENT_RUN_TIMEOUT` retrying.
+const JAVA_CLIENT_TIMEOUT_MS: u32 = 10_000;
+
+/// Reports why the suite cannot run, and whether that is fatal.
+///
+/// Returns `true` when the caller should skip. `KAFKA_E2E_REQUIRED=1` makes
it panic instead, so a
+/// CI job that means to run these fails loudly rather than reporting a pass
over zero assertions.
+fn skip(reason: &str) -> bool {
+ assert!(
+ std::env::var("KAFKA_E2E_REQUIRED").as_deref() != Ok("1"),
+ "KAFKA_E2E_REQUIRED=1 but the suite cannot run: {reason}"
+ );
+ eprintln!("skipping real-client end-to-end test: {reason}");
+ true
+}
+
+fn docker_missing() -> bool {
+ let available = Command::new("docker")
+ .arg("info")
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .status()
+ .is_ok_and(|status| status.success());
+ if available {
+ false
+ } else {
+ skip("docker is unavailable")
+ }
+}
+
+/// Whether `image` is already in the local Docker store.
+///
+/// Containers run with `--pull never`, so a missing image would otherwise
surface as a client that
+/// failed to start, reported against whichever feature that test happened to
cover.
+fn image_present(image: &str) -> bool {
+ Command::new("docker")
+ .args(["image", "inspect", image])
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .status()
+ .is_ok_and(|status| status.success())
+}
+
+/// Locates the already-built `iggy-server` alongside this test binary. Does
not build it.
+fn iggy_server_binary() -> Option<PathBuf> {
+ let mut dir = std::env::current_exe().ok()?;
+ // .../target/<profile>/deps/<test binary> -> .../target/<profile>
+ dir.pop();
+ dir.pop();
+ let candidate = dir.join(format!("iggy-server{}",
std::env::consts::EXE_SUFFIX));
+ candidate.is_file().then_some(candidate)
+}
+
+/// Outcome of waiting for the spawned server to come up.
+enum Ready {
+ Listening,
+ /// The process exited before it bound, carrying its status.
+ Exited(String),
+ TimedOut,
+}
+
+/// A spawned `iggy-server`, killed on drop.
+///
+/// The environment recipe mirrors `bridge_iggy_integration_tests.rs`, which
explains each setting:
+/// the other listeners are off so concurrently spawned servers do not fight
over their fixed
+/// ports, and the shard pool is capped so a spawned server does not size
itself to the whole
+/// machine against unrelated packages' tests in the same run.
+struct TestServer {
+ child: Child,
+ address: String,
+ http_address: String,
+ _data_dir: tempfile::TempDir,
+}
+
+impl TestServer {
+ fn spawn() -> Result<Self, String> {
+ let Some(binary) = iggy_server_binary() else {
+ return Err(
+ "iggy-server is not built; run `cargo build --bin iggy-server`
first".to_string(),
+ );
+ };
+ let data_dir = tempfile::tempdir().expect("create server data dir");
+ let port = free_port();
+ let address = format!("127.0.0.1:{port}");
+ let http_address = format!("127.0.0.1:{}", free_port());
+
+ let child = Command::new(binary)
+ .arg("--fresh")
+ .env("IGGY_PATH", data_dir.path().display().to_string())
+ .env("IGGY_TCP_ADDRESS", &address)
+ // Left on, unlike the sibling suite: the three-principal
procedure creates its
+ // non-root users over the HTTP API, which is the only
administrative surface reachable
+ // from a test without pulling in the SDK.
+ .env("IGGY_HTTP_ENABLED", "true")
+ .env("IGGY_HTTP_ADDRESS", &http_address)
+ .env("IGGY_QUIC_ENABLED", "false")
+ .env("IGGY_WEBSOCKET_ENABLED", "false")
+ .env("IGGY_SHARDING_PIN_CORES", "false")
+ .env("IGGY_SHARDING_CPU_ALLOCATION", "0..4")
+ .env("IGGY_ROOT_USERNAME", ROOT_USER)
+ .env("IGGY_ROOT_PASSWORD", ROOT_PASSWORD)
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .spawn()
+ .expect("spawn iggy-server");
+
+ let mut server = Self {
+ child,
+ address,
+ http_address,
+ _data_dir: data_dir,
+ };
+ match server.wait_ready() {
+ Ready::Listening => Ok(server),
+ Ready::Exited(status) => Err(format!(
+ "iggy-server exited during startup with {status}; its output
is suppressed, so \
+ rerun it by hand with the same environment to see why"
+ )),
+ Ready::TimedOut => Err(format!(
+ "iggy-server never bound {} within {SERVER_READY_TIMEOUT:?}",
+ server.address
+ )),
+ }
+ }
+
+ fn wait_ready(&mut self) -> Ready {
+ let deadline = Instant::now() + SERVER_READY_TIMEOUT;
+ while Instant::now() < deadline {
+ // Both listeners, not just the data one: the server binds HTTP
after TCP, so probing
+ // TCP alone declares readiness while the provisioning calls that
follow would still
+ // be refused.
+ if std::net::TcpStream::connect(&self.address).is_ok()
+ && std::net::TcpStream::connect(&self.http_address).is_ok()
+ {
+ return Ready::Listening;
+ }
+ // Without this a server that dies at boot burns the whole budget
and is then reported
+ // as a timeout, which sends the reader looking at the wrong thing
entirely.
+ if let Ok(Some(status)) = self.child.try_wait() {
+ return Ready::Exited(status.to_string());
+ }
+ std::thread::sleep(Duration::from_millis(200));
+ }
+ Ready::TimedOut
+ }
+}
+
+impl Drop for TestServer {
+ fn drop(&mut self) {
+ let _ = self.child.kill();
+ let _ = self.child.wait();
+ }
+}
+
+/// Binds an ephemeral port and releases it, so the server can take it.
+///
+/// A window exists between release and rebind. Acceptable here because this
suite is serialized
+/// into its own nextest group, so nothing else in it is drawing ports
concurrently.
+fn free_port() -> u16 {
+ let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind
ephemeral port");
+ listener.local_addr().expect("local addr").port()
+}
+
+/// Starts the gateway with SASL on, verifying credentials against
`iggy_address`.
+async fn spawn_gateway(iggy_address: &str) -> SocketAddr {
+ let config = GatewayConfig {
+ sasl_enabled: true,
+ ..GatewayConfig::default()
+ };
+ let authenticator =
Arc::new(IggyAuthenticator::new(iggy_address.to_string()));
+ let (addr, shutdown) = spawn_test_server_with_authenticator(config,
authenticator).await;
+ // Held for the test's lifetime: dropping the sender shuts the gateway
down mid-exchange.
+ std::mem::forget(shutdown);
+ addr
+}
+
+/// Runs a container against the host network and returns its combined output.
+///
+/// `--network host` is what lets a containerised client reach a gateway bound
to the host's
+/// loopback. It is Linux-specific, which matches where CI runs.
+///
+/// This blocks the calling thread for the life of the container, which is why
every test here uses
+/// a multi-threaded runtime: the gateway runs as a spawned task, and on the
single-threaded runtime
+/// `#[tokio::test]` gives by default, this call would starve it and nothing
would ever listen.
+///
+/// `--pull never` keeps a registry download out of the test's time budget:
images are pulled up
+/// front, and `stack` skips when they are absent. `timeout` bounds the run
itself, and a run it
+/// cuts short exits non-zero, so `expect_ran` reports it rather than
asserting over partial output.
+fn run_client(image: &str, args: &[&str], mounts: &[(&str, &str)]) ->
ClientRun {
Review Comment:
GNU timeout --kill-after. macOS none. :269 expect panics after stack().
--network host Linux. **Fix**: skip unless GNU timeout / Linux host-net, or
wrap docker run.
##########
gateways/kafka/docs/TEST_SUITE.md:
##########
@@ -73,6 +74,34 @@ is test-only primitive encode/decode scaffolding for
hand-building legacy/advers
---
+## Real-client end-to-end suite
+
+`kafka_client_e2e_tests.rs` needs two things the rest of the suite does not:
Docker, and an
+already-built `iggy-server` in the same target directory. Missing either makes
it skip with a
Review Comment:
“20+ … ACLs, etc. | Later issues” while Phase 3+ [x] DescribeAcls (:153).
Table omits 17/29/36. Prior REQUEST CHANGES leftover. **Fix**: drop ACL from
later row; list SASL-gated keys.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
