You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by bb...@apache.org on 2026/09/17 21:18:38 UTC
(kafka) branch trunk updated: KAFKA-21074: Kafka protocol fault proxy clients fixtures (#23438)
This is an automated email from the ASF dual-hosted git repository.
bbejeck pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 7cd729ff935 KAFKA-21074: Kafka protocol fault proxy clients fixtures (#23438)
7cd729ff935 is described below
commit 7cd729ff935ef57e3db075fcd92f768f5c2ef4a6
Author: Bill Bejeck <bb...@apache.org>
AuthorDate: Thu Sep 17 17:18:30 2026 -0400
KAFKA-21074: Kafka protocol fault proxy clients fixtures (#23438)
A lightweight Kafka wire-protocol fault-injection proxy for integration
tests. Sit it in front of an EmbeddedKafkaCluster, point any client’s
bootstrap.servers at it, and it can inject error codes, drop
connections, or blackhole a client — decoding/encoding with Kafka’s own
protocol classes so it is correct across every wire version (including
flexible/tagged-field ones), no hand-rolled byte offsets.
Placed in `clients/src/testFixtures` so it is reusable from any module.
This is internal test infrastructure only so it does not require a KIP.
Fault primitives, each armed with a fluent, deterministic trigger:
`injectError(apiKey, Errors)` — stamp an error code onto the matching
response, re-serialized at the correct wire version. Supported APIs:
`END_TXN`, `INIT_PRODUCER_ID`, `ADD_OFFSETS_TO_TXN`,
`TXN_OFFSET_COMMIT`, `PRODUCE`, `FETCH` (extensible by registering a
per-API setter). `disconnectOn(apiKey)` — drop the connection when the
matching response would return (models the EOS "commit gap"). Works
on any API. `delayOn(apiKey, Duration)` — hold back the matching
response to model a slow broker, isolated to that one connection.
`blackholeClient(clientIdSubstring)` — drop all requests from a targeted
client before the broker sees them, so the broker evicts it by
session timeout (a reversible, ungraceful one-node partition). Shared
trigger DSL (`Occurrence`): `once()`, `onCall`, `times`, `everyTime()`,
`withProbability(p)`. The deterministic triggers are safe for
assertions; probability is chaos-mode only. The proxy never closes
sockets unless a `disconnectOn(...)` rule fires, so it is not itself a
source of flakiness. Rules can be armed/disarmed live from the test
thread and each returns a handle exposing match/fire counts.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>, Lucas Brutschy
<lb...@confluent.io>
---
.../apache/kafka/test/faultproxy/FaultRule.java | 201 +++++++++
.../test/faultproxy/KafkaProtocolFaultProxy.java | 450 +++++++++++++++++++++
.../apache/kafka/test/faultproxy/Occurrence.java | 61 +++
.../FaultProxyDelayIntegrationTest.java | 89 ++++
.../FaultProxyExampleIntegrationTest.java | 136 +++++++
...ultProxyPlainClientsExampleIntegrationTest.java | 102 +++++
6 files changed, 1039 insertions(+)
diff --git a/clients/src/testFixtures/java/org/apache/kafka/test/faultproxy/FaultRule.java b/clients/src/testFixtures/java/org/apache/kafka/test/faultproxy/FaultRule.java
new file mode 100644
index 00000000000..9d77de4002e
--- /dev/null
+++ b/clients/src/testFixtures/java/org/apache/kafka/test/faultproxy/FaultRule.java
@@ -0,0 +1,201 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.test.faultproxy;
+
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.IntPredicate;
+
+/**
+ * A single fault to apply to responses of one {@link ApiKeys}. Built through the fluent DSL on
+ * {@link KafkaProtocolFaultProxy} ({@code injectError(...)} / {@code disconnectOn(...)} then a trigger),
+ * and returned to the caller as a handle so a test can inspect {@link #timesTriggered()} or
+ * {@link #remove()} it.
+ *
+ * <p>A rule counts each response offered to it ("matches"); the {@link IntPredicate} trigger decides, from
+ * the 1-based match count, whether the fault fires on that match — e.g. {@code n -> n == 1} is {@code once()},
+ * {@code n -> n <= 3} is {@code times(3)}. Responses are offered in registration order and consumed by the
+ * first rule that fires, so a later rule is not offered a response an earlier rule already acted on.
+ */
+public final class FaultRule {
+
+ enum Action { INJECT_ERROR, DISCONNECT, DELAY }
+
+ private final KafkaProtocolFaultProxy owner;
+ private final ApiKeys apiKey;
+ private final Action action;
+ private final Errors error; // only for INJECT_ERROR
+ private final long delayMillis; // only for DELAY
+ private final IntPredicate trigger;
+ private final String clientIdFilter; // null = any client; otherwise the request clientId must contain it
+ private final String description;
+
+ private final AtomicInteger matches = new AtomicInteger(0);
+ private final AtomicInteger triggered = new AtomicInteger(0);
+
+ FaultRule(final KafkaProtocolFaultProxy owner,
+ final ApiKeys apiKey,
+ final Action action,
+ final Errors error,
+ final long delayMillis,
+ final IntPredicate trigger,
+ final String clientIdFilter,
+ final String description) {
+ this.owner = owner;
+ this.apiKey = apiKey;
+ this.action = action;
+ this.error = error;
+ this.delayMillis = delayMillis;
+ this.trigger = trigger;
+ this.clientIdFilter = clientIdFilter;
+ this.description = description;
+ }
+
+ ApiKeys apiKey() {
+ return apiKey;
+ }
+
+ /**
+ * Whether this rule applies to a request from {@code clientId}. Lets a rule be scoped to a specific
+ * client — e.g. {@code forClient("restore")} to fault only the restore consumer's fetches, leaving the
+ * main consumer untouched. A null filter matches any client.
+ */
+ boolean matchesClient(final String clientId) {
+ return clientIdFilter == null || (clientId != null && clientId.contains(clientIdFilter));
+ }
+
+ Action action() {
+ return action;
+ }
+
+ Errors error() {
+ return error;
+ }
+
+ long delayMillis() {
+ return delayMillis;
+ }
+
+ /** Called by the proxy for each response of this rule's API; returns true if the fault should fire. */
+ boolean shouldFire() {
+ final int n = matches.incrementAndGet();
+ if (trigger.test(n)) {
+ triggered.incrementAndGet();
+ return true;
+ }
+ return false;
+ }
+
+ /** How many times this fault actually fired — for test assertions. */
+ public int timesTriggered() {
+ return triggered.get();
+ }
+
+ /** How many responses of this API the rule has observed (whether or not it fired). */
+ public int timesMatched() {
+ return matches.get();
+ }
+
+ /** Deregister this rule from the proxy. */
+ public void remove() {
+ owner.removeFault(this);
+ }
+
+ @Override
+ public String toString() {
+ return "FaultRule(" + description + ", triggered=" + triggered.get() + "/" + matches.get() + ")";
+ }
+
+ // ------------------------------------------------------------------
+ // Fluent trigger step: returned by injectError(...)/disconnectOn(...).
+ // Each terminal registers the built rule with the proxy and returns the handle.
+ // ------------------------------------------------------------------
+ public static final class Builder {
+ private final KafkaProtocolFaultProxy owner;
+ private final ApiKeys apiKey;
+ private final Action action;
+ private final Errors error;
+ private final long delayMillis;
+ private String clientIdFilter; // null = any client
+
+ Builder(final KafkaProtocolFaultProxy owner, final ApiKeys apiKey, final Action action,
+ final Errors error, final long delayMillis) {
+ this.owner = owner;
+ this.apiKey = apiKey;
+ this.action = action;
+ this.error = error;
+ this.delayMillis = delayMillis;
+ }
+
+ /**
+ * Scope this fault to requests whose clientId contains {@code substring} — e.g.
+ * {@code forClient("restore")} to fault only the restore consumer (leaving the main consumer and
+ * producer untouched). Without this, the rule matches any client.
+ */
+ public Builder forClient(final String substring) {
+ this.clientIdFilter = substring;
+ return this;
+ }
+
+ private FaultRule register(final IntPredicate trigger, final String triggerDesc) {
+ final String verb;
+ switch (action) {
+ case DISCONNECT:
+ verb = "disconnect";
+ break;
+ case DELAY:
+ verb = "delay " + delayMillis + "ms";
+ break;
+ default:
+ verb = "inject " + error;
+ break;
+ }
+ final String scope = clientIdFilter == null ? "" : " client~" + clientIdFilter;
+ final FaultRule rule = new FaultRule(owner, apiKey, action, error, delayMillis, trigger, clientIdFilter,
+ verb + " on " + apiKey + scope + " [" + triggerDesc + "]");
+ owner.addFault(rule);
+ return rule;
+ }
+
+ /** Fire on the first matching response only. */
+ public FaultRule once() {
+ return register(Occurrence.once(), "once");
+ }
+
+ /** Fire on exactly the {@code n}-th matching response (1-based). */
+ public FaultRule onCall(final int n) {
+ return register(Occurrence.call(n), "call #" + n);
+ }
+
+ /** Fire on the first {@code n} matching responses. */
+ public FaultRule times(final int n) {
+ return register(Occurrence.first(n), "first " + n);
+ }
+
+ /** Fire on every matching response until removed. */
+ public FaultRule everyTime() {
+ return register(Occurrence.always(), "every time");
+ }
+
+ /** Fire on each matching response with the given probability (chaos mode; non-deterministic). */
+ public FaultRule withProbability(final double p) {
+ return register(Occurrence.withProbability(p), "p=" + p);
+ }
+ }
+}
diff --git a/clients/src/testFixtures/java/org/apache/kafka/test/faultproxy/KafkaProtocolFaultProxy.java b/clients/src/testFixtures/java/org/apache/kafka/test/faultproxy/KafkaProtocolFaultProxy.java
new file mode 100644
index 00000000000..ff98312f272
--- /dev/null
+++ b/clients/src/testFixtures/java/org/apache/kafka/test/faultproxy/KafkaProtocolFaultProxy.java
@@ -0,0 +1,450 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.test.faultproxy;
+
+import org.apache.kafka.common.message.FetchResponseData;
+import org.apache.kafka.common.message.ProduceResponseData;
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.apache.kafka.common.message.TxnOffsetCommitResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.AddOffsetsToTxnResponse;
+import org.apache.kafka.common.requests.EndTxnResponse;
+import org.apache.kafka.common.requests.FetchResponse;
+import org.apache.kafka.common.requests.FindCoordinatorResponse;
+import org.apache.kafka.common.requests.InitProducerIdResponse;
+import org.apache.kafka.common.requests.MetadataResponse;
+import org.apache.kafka.common.requests.RequestHeader;
+import org.apache.kafka.common.requests.RequestUtils;
+import org.apache.kafka.common.requests.TxnOffsetCommitResponse;
+import org.apache.kafka.common.utils.Utils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BiConsumer;
+
+/**
+ * A lightweight, client-agnostic Kafka wire-protocol fault-injection proxy for fast integration tests.
+ *
+ * <p>Sit it in front of a real (embedded) broker and point any client's {@code bootstrap.servers} at it;
+ * it decodes requests/responses with Kafka's own protocol classes ({@link RequestHeader},
+ * {@link AbstractResponse}, {@link RequestUtils#serialize}) — so it is correct across every wire version,
+ * including flexible/tagged-field ones, with no hand-rolled byte offsets.
+ *
+ * <h2>Usage</h2>
+ * <pre>{@code
+ * try (var broker = new EmbeddedKafkaCluster(1)) {
+ * broker.start();
+ * try (var proxy = KafkaProtocolFaultProxy.inFrontOf(broker.bootstrapServers())) {
+ * // point clients here:
+ * props.put(BOOTSTRAP_SERVERS_CONFIG, proxy.bootstrapServers());
+ *
+ * proxy.injectError(ApiKeys.END_TXN, Errors.CONCURRENT_TRANSACTIONS).once();
+ * proxy.injectError(ApiKeys.PRODUCE, Errors.NOT_ENOUGH_REPLICAS).onCall(2);
+ * proxy.disconnectOn(ApiKeys.END_TXN).once(); // the EOS "commit gap"
+ * proxy.delayOn(ApiKeys.FETCH, Duration.ofSeconds(2)).everyTime(); // slow broker
+ * }
+ * }
+ * }</pre>
+ *
+ * <p>Routing is transparent: the proxy rewrites {@code Metadata}/{@code FindCoordinator} responses so every
+ * advertised address points back at itself, so a single-broker embedded cluster needs no special config
+ * (its own ephemeral port is discovered from {@code bootstrapServers()}).
+ *
+ * <p>Determinism: {@code once()}/{@code onCall(n)}/{@code times(n)} are deterministic and safe for
+ * assertions; {@code withProbability(p)} is chaos-mode only. The proxy never closes sockets unless a
+ * {@code disconnectOn(...)} rule fires, so it is not itself a source of flakiness.
+ */
+public final class KafkaProtocolFaultProxy implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaProtocolFaultProxy.class);
+
+ /**
+ * Per-API setter that stamps an injected {@link Errors} onto a decoded response. Only APIs registered
+ * here support {@code injectError(...)}; anything else fails fast at rule registration.
+ */
+ private static final Map<ApiKeys, BiConsumer<AbstractResponse, Errors>> ERROR_SETTERS = new EnumMap<>(ApiKeys.class);
+ static {
+ ERROR_SETTERS.put(ApiKeys.END_TXN, (r, e) -> ((EndTxnResponse) r).data().setErrorCode(e.code()));
+ ERROR_SETTERS.put(ApiKeys.INIT_PRODUCER_ID, (r, e) -> ((InitProducerIdResponse) r).data().setErrorCode(e.code()));
+ ERROR_SETTERS.put(ApiKeys.ADD_OFFSETS_TO_TXN, (r, e) -> ((AddOffsetsToTxnResponse) r).data().setErrorCode(e.code()));
+ // TxnOffsetCommit carries the consumed offsets into the transaction. Under EOS-v2 / transactions V2
+ // (KIP-890) the client sends this directly (AddOffsetsToTxn is skipped -- see TransactionManager
+ // #sendOffsetsToTransaction), so this is THE offset-commit-into-txn RPC to fault for KIP-1035. The
+ // response carries per-partition error codes, so stamp every partition of every topic.
+ ERROR_SETTERS.put(ApiKeys.TXN_OFFSET_COMMIT, (r, e) -> {
+ final TxnOffsetCommitResponseData data = ((TxnOffsetCommitResponse) r).data();
+ data.topics().forEach(topic ->
+ topic.partitions().forEach(p -> p.setErrorCode(e.code())));
+ });
+ ERROR_SETTERS.put(ApiKeys.PRODUCE, (r, e) -> {
+ final ProduceResponseData data = ((org.apache.kafka.common.requests.ProduceResponse) r).data();
+ data.responses().forEach(topic ->
+ topic.partitionResponses().forEach(p -> p.setErrorCode(e.code())));
+ });
+ // FETCH stamps the error on every partition of the response. Because a fetch fault is almost always
+ // scoped with forClient("restore") (or another clientId), this hits only the targeted consumer's
+ // fetches — e.g. inject OFFSET_OUT_OF_RANGE on the restore consumer to exercise the restore path.
+ ERROR_SETTERS.put(ApiKeys.FETCH, (r, e) -> {
+ final FetchResponseData data = ((FetchResponse) r).data();
+ data.responses().forEach(topic ->
+ topic.partitions().forEach(p -> p.setErrorCode(e.code())));
+ });
+ }
+
+ private final String targetHost;
+ private final int targetPort;
+ private final ExecutorService threadPool = Executors.newCachedThreadPool(r -> {
+ final Thread t = new Thread(r, "kafka-fault-proxy");
+ t.setDaemon(true);
+ return t;
+ });
+ private final AtomicBoolean running = new AtomicBoolean(false);
+ private final CopyOnWriteArrayList<FaultRule> rules = new CopyOnWriteArrayList<>();
+ private final Set<String> blackholedClients = ConcurrentHashMap.newKeySet();
+ // Live connections, so close() can force sockets shut and unblock the pump reads that own them.
+ private final Set<Connection> connections = ConcurrentHashMap.newKeySet();
+ private ServerSocket serverSocket;
+ private volatile String proxyHost;
+ private volatile int proxyPort;
+
+ private KafkaProtocolFaultProxy(final String targetBootstrap) {
+ final String hostPort = targetBootstrap.split(",")[0].trim();
+ final int idx = hostPort.lastIndexOf(':');
+ this.targetHost = hostPort.substring(0, idx);
+ this.targetPort = Integer.parseInt(hostPort.substring(idx + 1));
+ }
+
+ /** Create and start a proxy in front of the given broker bootstrap address. */
+ public static KafkaProtocolFaultProxy inFrontOf(final String targetBootstrap) {
+ final KafkaProtocolFaultProxy proxy = new KafkaProtocolFaultProxy(targetBootstrap);
+ try {
+ proxy.start();
+ } catch (final Exception e) {
+ throw new RuntimeException("Failed to start fault proxy", e);
+ }
+ return proxy;
+ }
+
+ private void start() throws Exception {
+ serverSocket = new ServerSocket(0);
+ proxyPort = serverSocket.getLocalPort();
+ proxyHost = "localhost";
+ running.set(true);
+ threadPool.submit(this::acceptLoop);
+ LOG.info("Fault proxy listening on {}:{} -> broker {}:{}", proxyHost, proxyPort, targetHost, targetPort);
+ }
+
+ /** The address to hand to a client's {@code bootstrap.servers}. */
+ public String bootstrapServers() {
+ return proxyHost + ":" + proxyPort;
+ }
+
+ // ------------------------------------------------------------------
+ // DSL
+ // ------------------------------------------------------------------
+
+ /** Rewrite responses of {@code apiKey} to carry {@code error}. Follow with a trigger, e.g. {@code .once()}. */
+ public FaultRule.Builder injectError(final ApiKeys apiKey, final Errors error) {
+ if (!ERROR_SETTERS.containsKey(apiKey)) {
+ throw new IllegalArgumentException("Error injection is not supported for " + apiKey
+ + " yet. Supported: " + ERROR_SETTERS.keySet()
+ + " (add a setter to ERROR_SETTERS to extend).");
+ }
+ if (apiKey == ApiKeys.METADATA || apiKey == ApiKeys.FIND_COORDINATOR) {
+ throw new IllegalArgumentException(apiKey + " is reserved for routing and cannot carry injected errors.");
+ }
+ return new FaultRule.Builder(this, apiKey, FaultRule.Action.INJECT_ERROR, error, 0L);
+ }
+
+ /** Drop the connection when a response of {@code apiKey} would be returned (models the EOS commit gap). */
+ public FaultRule.Builder disconnectOn(final ApiKeys apiKey) {
+ return new FaultRule.Builder(this, apiKey, FaultRule.Action.DISCONNECT, null, 0L);
+ }
+
+ /**
+ * Hold back a response of {@code apiKey} by {@code delay} before forwarding it to the client, modelling a
+ * slow broker. Only the matching response on its own connection is delayed (each connection/direction runs
+ * on its own thread), so other clients and the request path are unaffected. Follow with a trigger, e.g.
+ * {@code .everyTime()}. A delay longer than the client's {@code request.timeout.ms} will be seen by the
+ * client as a timeout (and handled like a disconnect), so size the delay against the timeout under test.
+ */
+ public FaultRule.Builder delayOn(final ApiKeys apiKey, final Duration delay) {
+ final long millis = delay.toMillis();
+ if (millis < 0) {
+ throw new IllegalArgumentException("delay must not be negative: " + delay);
+ }
+ return new FaultRule.Builder(this, apiKey, FaultRule.Action.DELAY, null, millis);
+ }
+
+ /**
+ * Blackhole every request from clients whose clientId contains {@code clientIdSubstring}: drop the request
+ * before it reaches the broker and close the connection. The broker stops hearing that instance's
+ * heartbeats and evicts it by session timeout — an ungraceful, reversible one-node network partition.
+ * Set a distinct {@code client.id} per Streams instance to target one instance. Undo via {@link #clearFaults()}.
+ */
+ public void blackholeClient(final String clientIdSubstring) {
+ blackholedClients.add(clientIdSubstring);
+ }
+
+ /** Remove all registered faults and client blackholes (routing rewrites are always on and unaffected). */
+ public void clearFaults() {
+ rules.clear();
+ blackholedClients.clear();
+ }
+
+ void addFault(final FaultRule rule) {
+ rules.add(rule);
+ }
+
+ void removeFault(final FaultRule rule) {
+ rules.remove(rule);
+ }
+
+ // ------------------------------------------------------------------
+ // Proxy internals
+ // ------------------------------------------------------------------
+
+ private void acceptLoop() {
+ while (running.get()) {
+ try {
+ final Socket client = serverSocket.accept();
+ final Socket broker = new Socket(targetHost, targetPort);
+ final Connection conn = new Connection(client, broker);
+ connections.add(conn);
+ // Guard the accept/close race: if close() already drained the registry, tear this one down.
+ if (!running.get()) {
+ connections.remove(conn);
+ conn.closeQuietly();
+ break;
+ }
+ threadPool.submit(() -> pumpRequests(conn));
+ threadPool.submit(() -> pumpResponses(conn));
+ } catch (final Exception e) {
+ if (running.get()) {
+ LOG.warn("accept loop error", e);
+ }
+ }
+ }
+ }
+
+ /** Per-connection state: the socket pair and correlationId -> request header for decoding responses. */
+ private static final class Connection {
+ private final Socket client;
+ private final Socket broker;
+ private final Map<Integer, RequestHeader> inflight = new ConcurrentHashMap<>();
+
+ Connection(final Socket client, final Socket broker) {
+ this.client = client;
+ this.broker = broker;
+ }
+
+ void closeQuietly() {
+ Utils.closeQuietly(client, "fault-proxy client socket");
+ Utils.closeQuietly(broker, "fault-proxy broker socket");
+ }
+ }
+
+ // client -> broker: forward verbatim, recording each request header for response decoding. If the
+ // connection's clientId is blackholed, drop the request (do NOT forward) and close the connection — this
+ // simulates a one-node network partition on the REQUEST path, so the broker stops hearing that instance's
+ // heartbeats and evicts it by session timeout (the ungraceful-crash path). Reversible via clearFaults().
+ private void pumpRequests(final Connection conn) {
+ try {
+ final DataInputStream in = new DataInputStream(conn.client.getInputStream());
+ final DataOutputStream out = new DataOutputStream(conn.broker.getOutputStream());
+ byte[] frame;
+ while (running.get() && (frame = readFrame(in)) != null) {
+ try {
+ final RequestHeader header = RequestHeader.parse(ByteBuffer.wrap(frame));
+ conn.inflight.put(header.correlationId(), header);
+ if (isBlackholed(header.clientId())) {
+ LOG.info("Fault: blackholing request {} from client {} (dropping, not forwarding)",
+ header.apiKey(), header.clientId());
+ break; // stop pumping; the broker never sees this request
+ }
+ } catch (final Exception parseErr) {
+ LOG.debug("could not parse request header (forwarding anyway)", parseErr);
+ }
+ writeFrame(out, frame);
+ }
+ } catch (final Exception e) {
+ LOG.debug("request pump closed", e);
+ } finally {
+ connections.remove(conn);
+ conn.closeQuietly();
+ }
+ }
+
+ private boolean isBlackholed(final String clientId) {
+ if (clientId == null || blackholedClients.isEmpty()) {
+ return false;
+ }
+ for (final String substring : blackholedClients) {
+ if (clientId.contains(substring)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ // broker -> client: rewrite for routing and/or apply a matching fault rule; otherwise forward verbatim.
+ private void pumpResponses(final Connection conn) {
+ try {
+ final DataInputStream in = new DataInputStream(conn.broker.getInputStream());
+ final DataOutputStream out = new DataOutputStream(conn.client.getOutputStream());
+ byte[] frame;
+ while (running.get() && (frame = readFrame(in)) != null) {
+ final int correlationId = ByteBuffer.wrap(frame).getInt(0);
+ final RequestHeader reqHeader = conn.inflight.remove(correlationId);
+
+ if (reqHeader == null) {
+ writeFrame(out, frame);
+ continue;
+ }
+
+ final ApiKeys apiKey = reqHeader.apiKey();
+ final boolean routing = apiKey == ApiKeys.METADATA || apiKey == ApiKeys.FIND_COORDINATOR;
+ final FaultRule fired = firstFiringRule(apiKey, reqHeader.clientId());
+
+ if (fired != null && fired.action() == FaultRule.Action.DISCONNECT) {
+ LOG.info("Fault: dropping connection on {} response ({})", apiKey, fired);
+ break; // stop pumping; the finally closes the connection
+ }
+
+ if (fired != null && fired.action() == FaultRule.Action.DELAY) {
+ LOG.info("Fault: delaying {} response by {}ms ({})", apiKey, fired.delayMillis(), fired);
+ Thread.sleep(fired.delayMillis()); // blocks only this connection's response thread
+ }
+
+ if (!routing && (fired == null || fired.action() == FaultRule.Action.DELAY)) {
+ writeFrame(out, frame); // DELAY does not alter the bytes, only their timing
+ continue;
+ }
+
+ writeFrame(out, transform(reqHeader, frame, routing, fired));
+ }
+ } catch (final Exception e) {
+ LOG.debug("response pump closed", e);
+ } finally {
+ connections.remove(conn);
+ conn.closeQuietly();
+ }
+ }
+
+ // Decode -> (routing rewrite and/or error inject) -> re-encode. Falls back to the original bytes on error.
+ private byte[] transform(final RequestHeader reqHeader, final byte[] frame,
+ final boolean routing, final FaultRule fired) {
+ final ApiKeys apiKey = reqHeader.apiKey();
+ final short version = reqHeader.apiVersion();
+ try {
+ final AbstractResponse response = AbstractResponse.parseResponse(ByteBuffer.wrap(frame), reqHeader);
+
+ if (routing) {
+ applyRouting(response);
+ }
+ if (fired != null && fired.action() == FaultRule.Action.INJECT_ERROR) {
+ ERROR_SETTERS.get(apiKey).accept(response, fired.error());
+ LOG.info("Fault: injected {} into {} response ({})", fired.error(), apiKey, fired);
+ }
+
+ final ResponseHeaderData headerData = new ResponseHeaderData().setCorrelationId(reqHeader.correlationId());
+ final ByteBuffer bb = RequestUtils.serialize(
+ headerData, apiKey.responseHeaderVersion(version), response.data(), version);
+ final byte[] out = new byte[bb.remaining()];
+ bb.get(out);
+ return out;
+ } catch (final Exception e) {
+ LOG.warn("failed to transform {} v{} response; forwarding verbatim", apiKey, version, e);
+ return frame;
+ }
+ }
+
+ private void applyRouting(final AbstractResponse response) {
+ if (response instanceof MetadataResponse) {
+ ((MetadataResponse) response).data().brokers().forEach(b -> b.setHost(proxyHost).setPort(proxyPort));
+ } else if (response instanceof FindCoordinatorResponse) {
+ final FindCoordinatorResponse fc = (FindCoordinatorResponse) response;
+ fc.data().setHost(proxyHost).setPort(proxyPort);
+ fc.data().coordinators().forEach(c -> c.setHost(proxyHost).setPort(proxyPort));
+ }
+ }
+
+ private FaultRule firstFiringRule(final ApiKeys apiKey, final String clientId) {
+ for (final FaultRule rule : rules) {
+ // Gate on apiKey and clientId before shouldFire() so a client-scoped rule advances its counters
+ // only on matching requests. The first rule to fire consumes the response; later rules are not
+ // offered it, so each rule's counters reflect only responses it actually saw.
+ if (rule.apiKey() == apiKey && rule.matchesClient(clientId) && rule.shouldFire()) {
+ return rule;
+ }
+ }
+ return null;
+ }
+
+ /** Reads one length-prefixed Kafka frame (without the 4-byte length). Returns null on clean EOF/close. */
+ private static byte[] readFrame(final DataInputStream in) {
+ try {
+ final int size = in.readInt();
+ final byte[] frame = new byte[size];
+ in.readFully(frame);
+ return frame;
+ } catch (final Exception e) {
+ return null; // clean EOF or torn frame; the connection is done
+ }
+ }
+
+ private static void writeFrame(final DataOutputStream out, final byte[] frame) throws Exception {
+ out.writeInt(frame.length);
+ out.write(frame);
+ out.flush();
+ }
+
+ @Override
+ public void close() {
+ running.set(false);
+ Utils.closeQuietly(serverSocket, "fault-proxy server socket"); // unblocks accept()
+ connections.forEach(Connection::closeQuietly); // unblocks the blocking pump reads
+ connections.clear();
+ threadPool.shutdown();
+ try {
+ if (!threadPool.awaitTermination(10, TimeUnit.SECONDS)) {
+ threadPool.shutdownNow();
+ }
+ } catch (final InterruptedException e) {
+ threadPool.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
+ }
+}
diff --git a/clients/src/testFixtures/java/org/apache/kafka/test/faultproxy/Occurrence.java b/clients/src/testFixtures/java/org/apache/kafka/test/faultproxy/Occurrence.java
new file mode 100644
index 00000000000..63d2c336b82
--- /dev/null
+++ b/clients/src/testFixtures/java/org/apache/kafka/test/faultproxy/Occurrence.java
@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.test.faultproxy;
+
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.IntPredicate;
+
+/**
+ * Shared trigger predicates for the fault-injection DSLs. A trigger decides, from the 1-based count of
+ * matching events seen so far, whether the fault should fire on this one. Used by both
+ * {@link FaultRule.Builder} (wire-protocol faults) and {@link ClientFault.Builder} (client-exception faults)
+ * so the two DSLs read identically ({@code once()}, {@code onCall(n)}, {@code times(n)}, {@code everyTime()},
+ * {@code withProbability(p)}).
+ */
+final class Occurrence {
+
+ private Occurrence() {
+ }
+
+ /** Fire on the first matching event only. */
+ static IntPredicate once() {
+ return n -> n == 1;
+ }
+
+ /** Fire on exactly the {@code n}-th matching event (1-based). */
+ static IntPredicate call(final int n) {
+ return matchNo -> matchNo == n;
+ }
+
+ /** Fire on the first {@code n} matching events. */
+ static IntPredicate first(final int n) {
+ return matchNo -> matchNo <= n;
+ }
+
+ /** Fire on every matching event. */
+ static IntPredicate always() {
+ return n -> true;
+ }
+
+ /** Fire on each matching event with probability {@code p} (chaos mode; non-deterministic). */
+ static IntPredicate withProbability(final double p) {
+ if (p < 0.0 || p > 1.0) {
+ throw new IllegalArgumentException("probability must be in [0.0, 1.0], got " + p);
+ }
+ return n -> ThreadLocalRandom.current().nextDouble() < p;
+ }
+}
diff --git a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/FaultProxyDelayIntegrationTest.java b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/FaultProxyDelayIntegrationTest.java
new file mode 100644
index 00000000000..2c412e2911f
--- /dev/null
+++ b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/FaultProxyDelayIntegrationTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.streams.integration;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.test.faultproxy.FaultRule;
+import org.apache.kafka.test.faultproxy.KafkaProtocolFaultProxy;
+
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.time.Duration;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@Tag("integration")
+@Timeout(120)
+public class FaultProxyDelayIntegrationTest {
+
+ @Test
+ public void shouldDelayASingleProduceResponse() throws Exception {
+ final long delayMs = 2000L;
+ final EmbeddedKafkaCluster cluster = new EmbeddedKafkaCluster(1);
+ cluster.start();
+ try {
+ final String topic = "delay-example";
+ cluster.createTopic(topic, 1, 1);
+
+ try (final KafkaProtocolFaultProxy proxy =
+ KafkaProtocolFaultProxy.inFrontOf(cluster.bootstrapServers())) {
+
+ // Deterministically delay only the FIRST produce response.
+ final FaultRule delay =
+ proxy.delayOn(ApiKeys.PRODUCE, Duration.ofMillis(delayMs)).once();
+
+ final Properties producerProps = new Properties();
+ producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, proxy.bootstrapServers());
+ producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
+
+ try (final KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps)) {
+ // The first send blocks on the (delayed) produce response; time it. Metadata/connection
+ // setup only adds time, so the elapsed lower bound is safe against flakiness.
+ final long start = System.nanoTime();
+ producer.send(new ProducerRecord<>(topic, "k", "v0")).get();
+ final long firstMs = Duration.ofNanos(System.nanoTime() - start).toMillis();
+
+ // A comfortable lower bound below the injected delay (Thread.sleep guarantees >= delayMs).
+ assertTrue(firstMs >= delayMs - 250,
+ "first produce should have been delayed ~" + delayMs + "ms but took " + firstMs + "ms");
+
+ // A subsequent send is not delayed (once()): it should be far faster than the delay.
+ final long start2 = System.nanoTime();
+ producer.send(new ProducerRecord<>(topic, "k", "v1")).get();
+ final long secondMs = Duration.ofNanos(System.nanoTime() - start2).toMillis();
+ assertTrue(secondMs < delayMs,
+ "second produce should not be delayed but took " + secondMs + "ms");
+ }
+
+ assertEquals(1, delay.timesTriggered(), "delay should have fired exactly once");
+ }
+ } finally {
+ cluster.stop();
+ }
+ }
+}
diff --git a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/FaultProxyExampleIntegrationTest.java b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/FaultProxyExampleIntegrationTest.java
new file mode 100644
index 00000000000..9ec884a0935
--- /dev/null
+++ b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/FaultProxyExampleIntegrationTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.streams.integration;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.streams.kstream.Materialized;
+import org.apache.kafka.streams.kstream.Produced;
+import org.apache.kafka.streams.state.Stores;
+import org.apache.kafka.test.TestUtils;
+import org.apache.kafka.test.faultproxy.FaultRule;
+import org.apache.kafka.test.faultproxy.KafkaProtocolFaultProxy;
+
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import static java.util.Collections.singletonList;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@Tag("integration")
+@Timeout(120)
+public class FaultProxyExampleIntegrationTest {
+
+ @Test
+ public void shouldStayExactlyOnceWhenCommitIsFencedOnce() throws Exception {
+ final int numRecords = 300;
+ final EmbeddedKafkaCluster cluster = new EmbeddedKafkaCluster(1);
+ cluster.start();
+ try {
+ final String input = "in";
+ final String output = "out";
+ cluster.createTopic(input, 2, 1);
+ cluster.createTopic(output, 2, 1);
+
+ try (final KafkaProtocolFaultProxy proxy =
+ KafkaProtocolFaultProxy.inFrontOf(cluster.bootstrapServers())) {
+
+ // Deterministic fault: fence the FIRST transactional commit. Streams treats PRODUCER_FENCED
+ // as a TaskMigrated (recoverable) -> the task is re-initialized and processing continues.
+ final FaultRule fence =
+ proxy.injectError(ApiKeys.END_TXN, Errors.PRODUCER_FENCED).once();
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.stream(input, Consumed.with(Serdes.String(), Serdes.String()))
+ .groupByKey()
+ .count(Materialized.<String, Long>as(Stores.persistentKeyValueStore("counts"))
+ .withKeySerde(Serdes.String()).withValueSerde(Serdes.Long()))
+ .toStream()
+ .to(output, Produced.with(Serdes.String(), Serdes.Long()));
+
+ final Properties props = new Properties();
+ props.put(StreamsConfig.APPLICATION_ID_CONFIG, "fault-proxy-example");
+ // NB: point Streams at the PROXY, not the broker.
+ props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, proxy.bootstrapServers());
+ props.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
+ props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
+ props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
+ props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
+ props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
+ props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+ try (final KafkaStreams streams = new KafkaStreams(builder.build(), props)) {
+ streams.cleanUp();
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(
+ singletonList(streams), Duration.ofSeconds(60));
+
+ // Produce directly to the broker (bypass the proxy for input load).
+ final Properties producerConfig = new Properties();
+ producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());
+ producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ final List<KeyValue<String, String>> records = new ArrayList<>();
+ for (int i = 0; i < numRecords; i++) {
+ records.add(KeyValue.pair("k" + (i % 3), "v" + i));
+ }
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ input, records, producerConfig, cluster.time);
+
+ final Properties consumerConfig = new Properties();
+ consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());
+ consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "fault-proxy-example-verifier");
+ consumerConfig.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
+ consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class);
+
+ IntegrationTestUtils.waitUntilFinalKeyValueRecordsReceived(
+ consumerConfig, output,
+ List.of(KeyValue.pair("k0", 100L), KeyValue.pair("k1", 100L), KeyValue.pair("k2", 100L)),
+ 60_000L);
+
+ // Committed output at the target counts implies the fence fired and was recovered from.
+ assertTrue(fence.timesTriggered() >= 1, "END_TXN fence never fired");
+
+ // The app settles back to RUNNING after recovering from the fence.
+ TestUtils.waitForCondition(() -> streams.state() == KafkaStreams.State.RUNNING,
+ 60_000L, "Streams did not return to RUNNING after recovering from the fence");
+ }
+ }
+ } finally {
+ cluster.stop();
+ }
+ }
+}
\ No newline at end of file
diff --git a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/FaultProxyPlainClientsExampleIntegrationTest.java b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/FaultProxyPlainClientsExampleIntegrationTest.java
new file mode 100644
index 00000000000..62b735407dc
--- /dev/null
+++ b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/FaultProxyPlainClientsExampleIntegrationTest.java
@@ -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.
+ */
+package org.apache.kafka.streams.integration;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.test.faultproxy.FaultRule;
+import org.apache.kafka.test.faultproxy.KafkaProtocolFaultProxy;
+
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.time.Duration;
+import java.util.Properties;
+
+import static java.util.Collections.singletonList;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@Tag("integration")
+@Timeout(120)
+public class FaultProxyPlainClientsExampleIntegrationTest {
+
+ @Test
+ public void producerRetriesAroundInjectedProduceErrorAndConsumerReadsAll() throws Exception {
+ final int numRecords = 50;
+ final EmbeddedKafkaCluster cluster = new EmbeddedKafkaCluster(1);
+ cluster.start();
+ try {
+ final String topic = "plain-example";
+ cluster.createTopic(topic, 1, 1);
+
+ try (final KafkaProtocolFaultProxy proxy =
+ KafkaProtocolFaultProxy.inFrontOf(cluster.bootstrapServers())) {
+
+ // Fault the FIRST produce response with a retriable error; the producer recovers on retry.
+ final FaultRule produceError =
+ proxy.injectError(ApiKeys.PRODUCE, Errors.NOT_ENOUGH_REPLICAS).once();
+
+ // --- Producer: point bootstrap.servers at the PROXY ---
+ final Properties producerProps = new Properties();
+ producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, proxy.bootstrapServers());
+ producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
+ // retries + delivery.timeout.ms default high enough to absorb the one injected error.
+ try (final KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps)) {
+ for (int i = 0; i < numRecords; i++) {
+ // .get() surfaces the send result; the retriable error is retried internally,
+ // so this still returns success once the (single) injected fault has passed.
+ producer.send(new ProducerRecord<>(topic, "k" + (i % 3), "v" + i)).get();
+ }
+ }
+ assertTrue(produceError.timesTriggered() >= 1, "PRODUCE error never fired");
+
+ // --- Consumer: also point bootstrap.servers at the PROXY ---
+ final Properties consumerProps = new Properties();
+ consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, proxy.bootstrapServers());
+ consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "plain-example-group");
+ consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps)) {
+ consumer.subscribe(singletonList(topic));
+ int consumed = 0;
+ final long deadline = System.currentTimeMillis() + 60_000L;
+ while (consumed < numRecords && System.currentTimeMillis() < deadline) {
+ final ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
+ consumed += records.count();
+ }
+ assertEquals(numRecords, consumed, "consumer did not read every record back");
+ }
+ }
+ } finally {
+ cluster.stop();
+ }
+ }
+}
\ No newline at end of file