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/30 12:03:56 UTC
(kafka) branch trunk updated: MINOR: Fix FindCoordinator v4+ routing in KafkaProtocolFaultProxy (#23571)
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 26f92e80d84 MINOR: Fix FindCoordinator v4+ routing in KafkaProtocolFaultProxy (#23571)
26f92e80d84 is described below
commit 26f92e80d844e94896e275f38943a64848f6567a
Author: Bill Bejeck <bb...@apache.org>
AuthorDate: Wed Sep 30 08:03:49 2026 -0400
MINOR: Fix FindCoordinator v4+ routing in KafkaProtocolFaultProxy (#23571)
`applyRouting` set the top-level host/port on every FindCoordinator
response, but only v0-3 have those fields. On v4+ the response failed to
re-serialize and was forwarded unrewritten, so group coordinator
requests bypassed the proxy and its faults.
Now v0-3 responses get the top-level host/port rewritten and v4+
responses get the `coordinators` list rewritten. Added
`KafkaProtocolFaultProxyTest.shouldRewriteFindCoordinatorResponsesToPointAtTheProxy`
over all FindCoordinator versions; v4-6 fail without the fix.
Reviewers: Rahul <ra...@gmail.com>, Lucas Brutschy
<lb...@confluent.io>
---
checkstyle/import-control.xml | 1 +
.../faultproxy/KafkaProtocolFaultProxyTest.java | 137 +++++++++++++++++++++
.../test/faultproxy/KafkaProtocolFaultProxy.java | 14 ++-
3 files changed, 148 insertions(+), 4 deletions(-)
diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index c44d69bbf9e..512fd9de47f 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -457,6 +457,7 @@
<allow pkg="org.apache.kafka" />
<allow pkg="org.bouncycastle" />
<allow pkg="org.rocksdb" />
+ <allow class="org.apache.logging.log4j.Level" />
</subpackage>
<subpackage name="raft">
diff --git a/clients/src/test/java/org/apache/kafka/test/faultproxy/KafkaProtocolFaultProxyTest.java b/clients/src/test/java/org/apache/kafka/test/faultproxy/KafkaProtocolFaultProxyTest.java
index d57a1585e84..8a7355a41b2 100644
--- a/clients/src/test/java/org/apache/kafka/test/faultproxy/KafkaProtocolFaultProxyTest.java
+++ b/clients/src/test/java/org/apache/kafka/test/faultproxy/KafkaProtocolFaultProxyTest.java
@@ -16,7 +16,33 @@
*/
package org.apache.kafka.test.faultproxy;
+import org.apache.kafka.common.message.FindCoordinatorRequestData;
+import org.apache.kafka.common.message.FindCoordinatorResponseData;
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.FindCoordinatorResponse;
+import org.apache.kafka.common.requests.RequestHeader;
+import org.apache.kafka.common.requests.RequestUtils;
+import org.apache.kafka.common.utils.LogCaptureAppender;
+import org.apache.kafka.common.utils.annotation.ApiKeyVersionsSource;
+
+import org.apache.logging.log4j.Level;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -24,6 +50,10 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class KafkaProtocolFaultProxyTest {
+ private static final String UPSTREAM_HOST = "upstream-broker";
+ private static final int UPSTREAM_PORT = 19092;
+ private static final String GROUP_ID = "group";
+
@Test
public void shouldRejectMultiBrokerBootstrap() {
// The proxy rewrites all routing to itself and forwards to a single upstream broker, so a
@@ -43,4 +73,111 @@ public class KafkaProtocolFaultProxyTest {
assertEquals(1, bootstrap.split(",").length, bootstrap);
}
}
+
+ @ParameterizedTest
+ @ApiKeyVersionsSource(apiKey = ApiKeys.FIND_COORDINATOR)
+ @Timeout(30)
+ public void shouldRewriteFindCoordinatorResponsesToPointAtTheProxy(final short version) throws Exception {
+ // Unless the coordinator address is rewritten, clients talk to the coordinator directly and every
+ // group-coordinator request bypasses the proxy, so faults registered for those APIs never fire.
+ final FindCoordinatorResponseData upstreamResponse = new FindCoordinatorResponseData();
+ if (version < FindCoordinatorRequest.MIN_BATCHED_VERSION) {
+ upstreamResponse.setNodeId(0).setHost(UPSTREAM_HOST).setPort(UPSTREAM_PORT);
+ } else {
+ upstreamResponse.setCoordinators(List.of(new FindCoordinatorResponseData.Coordinator()
+ .setKey(GROUP_ID).setNodeId(0).setHost(UPSTREAM_HOST).setPort(UPSTREAM_PORT)));
+ }
+ try (ServerSocket upstream = new ServerSocket(0);
+ KafkaProtocolFaultProxy proxy = KafkaProtocolFaultProxy.inFrontOf("localhost:" + upstream.getLocalPort())) {
+ final FindCoordinatorResponseData response =
+ findCoordinatorThroughProxy(upstream, proxy, version, upstreamResponse);
+
+ if (version < FindCoordinatorRequest.MIN_BATCHED_VERSION) {
+ assertEquals(proxy.bootstrapServers(), response.host() + ":" + response.port());
+ } else {
+ assertEquals(1, response.coordinators().size());
+ final FindCoordinatorResponseData.Coordinator coordinator = response.coordinators().get(0);
+ assertEquals(proxy.bootstrapServers(), coordinator.host() + ":" + coordinator.port());
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ApiKeyVersionsSource(apiKey = ApiKeys.FIND_COORDINATOR, fromVersion = FindCoordinatorRequest.MIN_BATCHED_VERSION)
+ @Timeout(30)
+ public void shouldReEncodeFindCoordinatorResponsesWithNoCoordinators(final short version) throws Exception {
+ // An empty coordinators list is valid at v4+ and leaves nothing to rewrite, so the client receives the
+ // same bytes whether the proxy re-encodes the response or fails and forwards it verbatim; only the
+ // warning logged on that failure tells the two apart.
+ try (LogCaptureAppender appender = LogCaptureAppender.createAndRegister(KafkaProtocolFaultProxy.class);
+ ServerSocket upstream = new ServerSocket(0);
+ KafkaProtocolFaultProxy proxy = KafkaProtocolFaultProxy.inFrontOf("localhost:" + upstream.getLocalPort())) {
+ appender.setClassLogger(KafkaProtocolFaultProxy.class, Level.WARN);
+
+ final FindCoordinatorResponseData response =
+ findCoordinatorThroughProxy(upstream, proxy, version, new FindCoordinatorResponseData());
+
+ assertEquals(List.of(), response.coordinators());
+ assertEquals(List.of(), appender.getMessages(Level.WARN));
+ }
+ }
+
+ // Sends a FindCoordinator request through the proxy, has the upstream broker answer it with
+ // upstreamResponse, and returns the response as the client receives it.
+ private static FindCoordinatorResponseData findCoordinatorThroughProxy(final ServerSocket upstream,
+ final KafkaProtocolFaultProxy proxy,
+ final short version,
+ final FindCoordinatorResponseData upstreamResponse)
+ throws Exception {
+ final CompletableFuture<Void> upstreamReplied =
+ CompletableFuture.runAsync(() -> replyFromUpstream(upstream, upstreamResponse, version));
+
+ final RequestHeader header = new RequestHeader(ApiKeys.FIND_COORDINATOR, version, "fault-proxy-test", 1);
+ final ByteBuffer request = new FindCoordinatorRequest.Builder(new FindCoordinatorRequestData()
+ .setKeyType(FindCoordinatorRequest.CoordinatorType.GROUP.id())
+ .setCoordinatorKeys(List.of(GROUP_ID)))
+ .build(version)
+ .serializeWithHeader(header);
+
+ final FindCoordinatorResponseData response;
+ final String[] proxyAddress = proxy.bootstrapServers().split(":");
+ try (Socket client = new Socket(proxyAddress[0], Integer.parseInt(proxyAddress[1]))) {
+ client.setSoTimeout(10_000);
+ writeFrame(new DataOutputStream(client.getOutputStream()), request);
+ response = ((FindCoordinatorResponse) AbstractResponse.parseResponse(
+ ByteBuffer.wrap(readFrame(new DataInputStream(client.getInputStream()))), header)).data();
+ }
+ upstreamReplied.get(10, TimeUnit.SECONDS);
+ return response;
+ }
+
+ private static void replyFromUpstream(final ServerSocket upstream,
+ final FindCoordinatorResponseData response,
+ final short version) {
+ try (Socket broker = upstream.accept()) {
+ final RequestHeader header =
+ RequestHeader.parse(ByteBuffer.wrap(readFrame(new DataInputStream(broker.getInputStream()))));
+ writeFrame(new DataOutputStream(broker.getOutputStream()), RequestUtils.serialize(
+ new ResponseHeaderData().setCorrelationId(header.correlationId()),
+ ApiKeys.FIND_COORDINATOR.responseHeaderVersion(version),
+ response,
+ version));
+ } catch (final IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private static byte[] readFrame(final DataInputStream in) throws IOException {
+ final byte[] frame = new byte[in.readInt()];
+ in.readFully(frame);
+ return frame;
+ }
+
+ private static void writeFrame(final DataOutputStream out, final ByteBuffer frame) throws IOException {
+ final byte[] bytes = new byte[frame.remaining()];
+ frame.duplicate().get(bytes);
+ out.writeInt(bytes.length);
+ out.write(bytes);
+ out.flush();
+ }
}
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
index ff5e19a240d..17d2261dc14 100644
--- a/clients/src/testFixtures/java/org/apache/kafka/test/faultproxy/KafkaProtocolFaultProxy.java
+++ b/clients/src/testFixtures/java/org/apache/kafka/test/faultproxy/KafkaProtocolFaultProxy.java
@@ -26,6 +26,7 @@ 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.FindCoordinatorRequest;
import org.apache.kafka.common.requests.FindCoordinatorResponse;
import org.apache.kafka.common.requests.InitProducerIdResponse;
import org.apache.kafka.common.requests.MetadataResponse;
@@ -393,7 +394,7 @@ public final class KafkaProtocolFaultProxy implements AutoCloseable {
final AbstractResponse response = AbstractResponse.parseResponse(ByteBuffer.wrap(frame), reqHeader);
if (routing) {
- applyRouting(response);
+ applyRouting(response, version);
}
if (fired != null && fired.action() == FaultRule.Action.INJECT_ERROR) {
ERROR_SETTERS.get(apiKey).accept(response, fired.error());
@@ -412,13 +413,18 @@ public final class KafkaProtocolFaultProxy implements AutoCloseable {
}
}
- private void applyRouting(final AbstractResponse response) {
+ private void applyRouting(final AbstractResponse response, final short version) {
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));
+ // v0-3 carry a single top-level host/port and v4+ only the (possibly empty) coordinators list;
+ // setting the top-level fields at v4+ makes re-serialization fail and the frame go out unrewritten
+ if (version < FindCoordinatorRequest.MIN_BATCHED_VERSION) {
+ fc.data().setHost(proxyHost).setPort(proxyPort);
+ } else {
+ fc.data().coordinators().forEach(c -> c.setHost(proxyHost).setPort(proxyPort));
+ }
}
}