This is an automated email from the ASF dual-hosted git repository.
AndrewJSchofield 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 7be741d08b3 KAFKA-20246: Add clusterId and nodeId to
ApiVersionsRequest (3/N) (#22512)
7be741d08b3 is described below
commit 7be741d08b3b06f6414ac868e57bf9b958f53a72
Author: Andrew Schofield <[email protected]>
AuthorDate: Mon Jul 6 10:20:39 2026 +0100
KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) (#22512)
Part of the implementation of
https://cwiki.apache.org/confluence/display/KAFKA/KIP-1242%3A+Detection+and+handling+of+misrouted+connections.
Support coordinator connections and mark ApiVersions v5 API as stable.
Reviewers: TaiJuWu <[email protected]>, David Jacot
<[email protected]>, Rajini Sivaram <[email protected]>
---
.../org/apache/kafka/clients/NetworkClient.java | 5 +-
.../consumer/internals/AbstractCoordinator.java | 8 +---
.../internals/CoordinatorRequestManager.java | 7 +--
.../consumer/internals/GroupCoordinatorNode.java | 42 +++++++++++++++++
.../main/java/org/apache/kafka/common/Node.java | 17 ++++---
.../common/message/ApiVersionsRequest.json | 1 -
.../kafka/clients/admin/KafkaAdminClientTest.java | 19 ++++----
.../kafka/clients/consumer/KafkaConsumerTest.java | 35 +++++++-------
.../clients/consumer/KafkaShareConsumerTest.java | 3 +-
.../internals/AbstractCoordinatorTest.java | 2 +-
.../consumer/internals/AsyncKafkaConsumerTest.java | 2 +-
.../internals/ConsumerCoordinatorTest.java | 2 +-
.../internals/CoordinatorRequestManagerTest.java | 4 +-
.../internals/GroupCoordinatorNodeTest.java | 54 ++++++++++++++++++++++
.../java/org/apache/kafka/clients/MockClient.java | 1 -
docs/getting-started/upgrade.md | 1 +
.../kafka/server/DefaultApiVersionManagerTest.java | 10 +---
17 files changed, 148 insertions(+), 65 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
index 3b826fe9ace..fd45ff14a91 100644
--- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -1138,10 +1138,7 @@ public class NetworkClient implements KafkaClient {
if (metadataRecoveryStrategy != MetadataRecoveryStrategy.NONE
&& metadataClusterCheckEnable) {
String clusterId = this.metadataUpdater.clusterId();
int nodeId = Integer.parseInt(node);
- // In order to allow separate connections to coordinators,
the client uses large positive node ID values
- // (Integer.MAX_VALUE - nodeId) for these connections
which do not match the target broker's actual node ID.
- // To avoid those, only check if the node ID is less than
half of Integer.MAX_VALUE.
- if (clusterId != null && nodeId >= 0 && nodeId <
Integer.MAX_VALUE / 2) {
+ if (clusterId != null && nodeId >= 0) {
apiVersionRequestBuilder.setClusterId(clusterId);
apiVersionRequestBuilder.setNodeId(nodeId);
}
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
index bfe848b4b56..6b104d37c22 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
@@ -934,12 +934,8 @@ public abstract class AbstractCoordinator implements
Closeable {
Errors error = Errors.forCode(coordinatorData.errorCode());
if (error == Errors.NONE) {
synchronized (AbstractCoordinator.this) {
- // use MAX_VALUE - node.id as the coordinator id to allow
separate connections
- // for the coordinator in the underlying network client
layer
- int coordinatorConnectionId = Integer.MAX_VALUE -
coordinatorData.nodeId();
-
- AbstractCoordinator.this.coordinator = new Node(
- coordinatorConnectionId,
+ AbstractCoordinator.this.coordinator = new
GroupCoordinatorNode(
+ coordinatorData.nodeId(),
coordinatorData.host(),
coordinatorData.port());
log.info("Discovered group coordinator {}", coordinator);
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java
index 872b213ad6a..2056ad66c23 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java
@@ -185,11 +185,8 @@ public class CoordinatorRequestManager implements
RequestManager {
final long currentTimeMs,
final FindCoordinatorResponseData.Coordinator coordinator
) {
- // use MAX_VALUE - node.id as the coordinator id to allow separate
connections
- // for the coordinator in the underlying network client layer
- int coordinatorConnectionId = Integer.MAX_VALUE - coordinator.nodeId();
- this.coordinator = new Node(
- coordinatorConnectionId,
+ this.coordinator = new GroupCoordinatorNode(
+ coordinator.nodeId(),
coordinator.host(),
coordinator.port());
log.info("Discovered group coordinator {}", coordinator);
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupCoordinatorNode.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupCoordinatorNode.java
new file mode 100644
index 00000000000..e6d98b61250
--- /dev/null
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupCoordinatorNode.java
@@ -0,0 +1,42 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.Node;
+
+/**
+ * This subclass of {@link Node} is used by the consumer for information about
+ * a Kafka node which is a group coordinator. It ensures that the idString
differs from
+ * a regular node with the same node ID so that the network code can maintain
separate
+ * network connections to the same node as a regular broker and as a group
coordinator.
+ * It achieves this by ensuring that the node ID is non-negative (which it
must be because
+ * negative node IDs are used for bootstrapping) and by prepending a '+' on
the node ID to
+ * create the idString. This maintains the requirement that the idString can
be parsed as
+ * an integer to obtain the actual node ID.
+ */
+public class GroupCoordinatorNode extends Node {
+ public GroupCoordinatorNode(int id, String host, int port) {
+ super(GroupCoordinatorNode.validateId(id), host, port, null, false,
"+" + id);
+ }
+
+ private static int validateId(int id) {
+ if (id < 0) {
+ throw new IllegalArgumentException("Node id for group coordinator
node cannot be negative");
+ }
+ return id;
+ }
+}
\ No newline at end of file
diff --git a/clients/src/main/java/org/apache/kafka/common/Node.java
b/clients/src/main/java/org/apache/kafka/common/Node.java
index 5ff1234e26a..013230ae01a 100644
--- a/clients/src/main/java/org/apache/kafka/common/Node.java
+++ b/clients/src/main/java/org/apache/kafka/common/Node.java
@@ -44,17 +44,16 @@ public class Node {
}
public Node(int id, String host, int port, String rack) {
- this.id = id;
- this.idString = Integer.toString(id);
- this.host = host;
- this.port = port;
- this.rack = rack;
- this.isFenced = false;
+ this(id, host, port, rack, false);
}
public Node(int id, String host, int port, String rack, boolean isFenced) {
+ this(id, host, port, rack, isFenced, Integer.toString(id));
+ }
+
+ protected Node(int id, String host, int port, String rack, boolean
isFenced, String idString) {
this.id = id;
- this.idString = Integer.toString(id);
+ this.idString = idString;
this.host = host;
this.port = port;
this.rack = rack;
@@ -83,7 +82,8 @@ public class Node {
/**
* String representation of the node id.
- * Typically the integer id is used to serialize over the wire, the string
representation is used as an identifier with NetworkClient code
+ * Typically, the integer id is used to serialize over the wire, the
string representation is used as an
+ * identifier with NetworkClient code
*/
public String idString() {
return idString;
@@ -161,5 +161,4 @@ public class Node {
public String toString() {
return host + ":" + port + " (id: " + idString + " rack: " + rack + "
isFenced: " + isFenced + ")";
}
-
}
diff --git a/clients/src/main/resources/common/message/ApiVersionsRequest.json
b/clients/src/main/resources/common/message/ApiVersionsRequest.json
index 22d120a04e9..311ae0f54b1 100644
--- a/clients/src/main/resources/common/message/ApiVersionsRequest.json
+++ b/clients/src/main/resources/common/message/ApiVersionsRequest.json
@@ -27,7 +27,6 @@
// Version 5 introduces ClusterId and NodeId checking and
REBOOTSTRAP_REQUIRED error (KIP-1242).
"validVersions": "0-5",
"flexibleVersions": "3+",
- "latestVersionUnstable": true,
"fields": [
{ "name": "ClientSoftwareName", "type": "string", "versions": "3+",
"ignorable": true, "about": "The name of the client." },
diff --git
a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
index 57610437012..364053aa3fc 100644
---
a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
@@ -916,21 +916,24 @@ public class KafkaAdminClientTest {
// which prevents AdminClient from being able to send the initial
metadata request
Cluster cluster = Cluster.bootstrap(singletonList(new
InetSocketAddress("localhost", 8121)));
- Map<Node, Long> unreachableNodes =
Collections.singletonMap(cluster.nodes().get(0), 200L);
+ Node bootstrapNode = cluster.nodes().get(0);
+ Map<Node, Long> unreachableNodes =
Collections.singletonMap(bootstrapNode, 200L);
try (final AdminClientUnitTestEnv env = new
AdminClientUnitTestEnv(Time.SYSTEM, cluster,
AdminClientUnitTestEnv.clientConfigs(AdminClientConfig.METADATA_RECOVERY_STRATEGY_CONFIG,
metadataRecoveryStrategy.name), unreachableNodes)) {
Cluster discoveredCluster = mockCluster(3, 0);
env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
- env.kafkaClient().prepareResponse(body -> body instanceof
MetadataRequest,
+ // Bind responses to specific destinations so MockClient delivery
does not depend on
+ // the iteration order of AdminClient's callsToSend map (which is
keyed by Node).
+ env.kafkaClient().prepareResponseFrom(body -> body instanceof
MetadataRequest,
RequestTestUtils.metadataResponse(discoveredCluster.nodes(),
discoveredCluster.clusterResource().clusterId(),
- 1, Collections.emptyList()));
+ 1, Collections.emptyList()), bootstrapNode);
if (metadataRecoveryStrategy ==
MetadataRecoveryStrategy.REBOOTSTRAP) {
- env.kafkaClient().prepareResponse(body -> body instanceof
MetadataRequest,
-
RequestTestUtils.metadataResponse(discoveredCluster.nodes(),
-
discoveredCluster.clusterResource().clusterId(), 1, Collections.emptyList()));
+ env.kafkaClient().prepareResponseFrom(body -> body instanceof
MetadataRequest,
+
RequestTestUtils.metadataResponse(discoveredCluster.nodes(),
discoveredCluster.clusterResource().clusterId(),
+ 1, Collections.emptyList()), bootstrapNode);
}
- env.kafkaClient().prepareResponse(body -> body instanceof
CreateTopicsRequest,
- prepareCreateTopicsResponse("myTopic", Errors.NONE));
+ env.kafkaClient().prepareResponseFrom(body -> body instanceof
CreateTopicsRequest,
+ prepareCreateTopicsResponse("myTopic", Errors.NONE),
discoveredCluster.nodeById(1));
KafkaFuture<Void> future = env.adminClient().createTopics(
singleton(new NewTopic("myTopic",
Collections.singletonMap(0, asList(0, 1, 2)))),
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
index 638868a29d3..aa596cf866c 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
@@ -28,6 +28,7 @@ import
org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy;
import org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer;
import org.apache.kafka.clients.consumer.internals.ConsumerMetadata;
import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.clients.consumer.internals.GroupCoordinatorNode;
import org.apache.kafka.clients.consumer.internals.MockRebalanceListener;
import org.apache.kafka.clients.consumer.internals.SubscriptionState;
import org.apache.kafka.common.Cluster;
@@ -1112,7 +1113,7 @@ public class KafkaConsumerTest {
// Since we would enable the heartbeat thread after received
join-response which could
// send the sync-group on behalf of the consumer if it is enqueued, we
may still complete
// the rebalance and send out the fetch; in order to avoid it we do
not prepare sync response here.
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
client.prepareResponseFrom(joinGroupFollowerResponse(assignor, 1,
memberId, leaderId, Errors.NONE), coordinator);
consumer.poll(Duration.ZERO);
@@ -1262,7 +1263,7 @@ public class KafkaConsumerTest {
Node node = metadata.fetch().nodes().get(0);
client.prepareResponseFrom(FindCoordinatorResponse.prepareResponse(Errors.NONE,
groupId, node), node);
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
client.prepareResponseFrom(offsetResponse(Map.of(tp0, -1L),
Errors.NONE), coordinator);
consumer = newConsumer(groupProtocol, time, client, subscription,
metadata, assignor,
@@ -1295,7 +1296,7 @@ public class KafkaConsumerTest {
true, groupId, groupInstanceId, false);
consumer.assign(List.of(tp0));
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
client.prepareResponseFrom(offsetResponse(Map.of(tp0, 539L),
Errors.NONE), coordinator);
consumer.poll(Duration.ZERO);
@@ -1331,7 +1332,7 @@ public class KafkaConsumerTest {
true, groupId, groupInstanceId, false);
consumer.assign(List.of(tp0));
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
client.prepareResponseFrom(offsetResponse(Map.of(tp0, -1L),
Errors.NONE), coordinator);
client.prepareResponse(listOffsetsResponse(Map.of(tp0, 50L)));
@@ -1373,7 +1374,7 @@ public class KafkaConsumerTest {
consumer.assign(List.of(tp0));
// lookup coordinator
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
// fetch offset for one topic
client.prepareResponseFrom(offsetResponse(Map.of(tp0, offset1),
Errors.NONE), coordinator);
@@ -1440,7 +1441,7 @@ public class KafkaConsumerTest {
groupProtocol, time, client, subscription, metadata, assignor,
true, groupId, groupInstanceId, true);
consumer.assign(List.of(tp0));
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
client.prepareResponseFrom(offsetResponse(
Map.of(tp0, offset1), Errors.NONE), coordinator);
@@ -1463,7 +1464,7 @@ public class KafkaConsumerTest {
consumer.assign(Arrays.asList(tp0, tp1));
// lookup coordinator
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
// fetch offset for one topic
client.prepareResponseFrom(offsetResponse(Map.of(tp0, offset1, tp1,
-1L), Errors.NONE), coordinator);
@@ -2012,7 +2013,7 @@ public class KafkaConsumerTest {
consumer = newConsumer(groupProtocol, time, client, subscription,
metadata, assignor, true, groupInstanceId);
// lookup coordinator
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
// manual assignment
consumer.assign(Set.of(tp0));
@@ -2069,7 +2070,7 @@ public class KafkaConsumerTest {
consumer = newConsumer(groupProtocol, time, client, subscription,
metadata, assignor, false, groupInstanceId);
// lookup coordinator
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
// manual assignment
consumer.assign(Set.of(tp0));
@@ -2124,7 +2125,7 @@ public class KafkaConsumerTest {
consumer = newConsumer(groupProtocol, time, client, subscription,
metadata, assignor, true, groupInstanceId);
// lookup coordinator
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
// manual assignment
Set<TopicPartition> partitions = Set.of(tp0, tp1);
@@ -2384,7 +2385,7 @@ public class KafkaConsumerTest {
consumer = newConsumer(groupProtocol, time, client, subscription,
metadata, assignor, false, groupInstanceId);
consumer.subscribe(Set.of(topic),
getConsumerRebalanceListener(consumer));
client.prepareResponseFrom(FindCoordinatorResponse.prepareResponse(Errors.NONE,
groupId, node), node);
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
client.prepareResponseFrom(joinGroupFollowerResponse(assignor, 1,
memberId, leaderId, Errors.NONE), coordinator);
@@ -2665,7 +2666,7 @@ public class KafkaConsumerTest {
client.prepareResponseFrom(
FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId,
node), node);
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
client.prepareResponseFrom(
offsetCommitResponse(Map.of(tp0, Errors.NONE)),
coordinator
@@ -2714,7 +2715,7 @@ public class KafkaConsumerTest {
// lookup coordinator
client.prepareResponseFrom(
FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId,
node), node);
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
// fetch offset for one topic
client.prepareResponseFrom(
@@ -2740,7 +2741,7 @@ public class KafkaConsumerTest {
KafkaConsumer<String, String> consumer = newConsumer(groupProtocol,
time, client, subscription, metadata, assignor, true, groupInstanceId);
consumer.subscribe(Set.of(topic),
getExceptionConsumerRebalanceListener());
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
client.prepareResponseFrom(FindCoordinatorResponse.prepareResponse(Errors.NONE,
groupId, node), node);
client.prepareResponseFrom(joinGroupFollowerResponse(assignor, 1,
memberId, leaderId, Errors.NONE), coordinator);
@@ -3285,7 +3286,7 @@ public class KafkaConsumerTest {
if (coordinator == null) {
// lookup coordinator
client.prepareResponseFrom(FindCoordinatorResponse.prepareResponse(Errors.NONE,
groupId, node), node);
- coordinator = new Node(Integer.MAX_VALUE - node.id(), node.host(),
node.port());
+ coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
}
// join group
@@ -3310,7 +3311,7 @@ public class KafkaConsumerTest {
if (coordinator == null) {
// lookup coordinator
client.prepareResponseFrom(FindCoordinatorResponse.prepareResponse(Errors.NONE,
groupId, node), node);
- coordinator = new Node(Integer.MAX_VALUE - node.id(), node.host(),
node.port());
+ coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
}
// join group
@@ -4118,7 +4119,7 @@ public void testPollIdleRatio(GroupProtocol
groupProtocol) {
consumer.assign(List.of(tp0));
// lookup coordinator
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
// try to get committed offsets for one topic-partition - but it is
disconnected so there's no response and it will time out
client.prepareResponseFrom(offsetResponse(Map.of(tp0, 0L),
Errors.NONE), coordinator, true);
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerTest.java
index 7741a8b58fd..45242f6d0b2 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerTest.java
@@ -19,6 +19,7 @@ package org.apache.kafka.clients.consumer;
import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.clients.MockClient;
import org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy;
+import org.apache.kafka.clients.consumer.internals.GroupCoordinatorNode;
import org.apache.kafka.clients.consumer.internals.ShareConsumerMetadata;
import org.apache.kafka.clients.consumer.internals.SubscriptionState;
import org.apache.kafka.common.Node;
@@ -317,7 +318,7 @@ public class KafkaShareConsumerTest {
private Node findCoordinator(MockClient client, Node node) {
client.prepareResponseFrom(FindCoordinatorResponse.prepareResponse(Errors.NONE,
groupId, node), node);
- return new Node(Integer.MAX_VALUE - node.id(), node.host(),
node.port());
+ return new GroupCoordinatorNode(node.id(), node.host(), node.port());
}
// This method generates a sequence of prepared SHARE_GROUP_HEARTBEAT
responses with increasing member epochs.
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
index a8137e3cab3..bf9527a4ade 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
@@ -154,7 +154,7 @@ public class AbstractCoordinatorTest {
mockClient.updateMetadata(RequestTestUtils.metadataUpdateWith(1,
emptyMap()));
this.node = metadata.fetch().nodes().get(0);
- this.coordinatorNode = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ this.coordinatorNode = new GroupCoordinatorNode(node.id(),
node.host(), node.port());
GroupRebalanceConfig rebalanceConfig = new
GroupRebalanceConfig(SESSION_TIMEOUT_MS,
rebalanceTimeoutMs,
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
index 41d9a9402b2..5ddca753702 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
@@ -2346,7 +2346,7 @@ public class AsyncKafkaConsumerTest {
new ConsumerGroupHeartbeatResponse(new
ConsumerGroupHeartbeatResponseData()
.setMemberId("")
.setMemberEpoch(0));
- Node coordinator = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ Node coordinator = new GroupCoordinatorNode(node.id(), node.host(),
node.port());
client.prepareResponseFrom(result, coordinator);
SubscriptionState subscriptionState = mock(SubscriptionState.class);
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
index e897683bb98..c9d050c161d 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
@@ -3022,7 +3022,7 @@ public abstract class ConsumerCoordinatorTest {
assertThrows(RebalanceInProgressException.class, () ->
coordinator.commitOffsetsSync(singletonMap(t1p,
new OffsetAndMetadata(100L, "metadata")),
time.timer(Long.MAX_VALUE)));
- final Node coordinatorNode = new Node(Integer.MAX_VALUE - node.id(),
node.host(), node.port());
+ final Node coordinatorNode = new GroupCoordinatorNode(node.id(),
node.host(), node.port());
client.respondFrom(joinGroupLeaderResponse(1, consumerId,
memberSubscriptions, Errors.NONE), coordinatorNode);
client.prepareResponse(body -> {
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManagerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManagerTest.java
index 91b3a7c2c33..955e60d1a2d 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManagerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManagerTest.java
@@ -73,7 +73,9 @@ public class CoordinatorRequestManagerTest {
Optional<Node> coordinatorOpt = coordinatorManager.coordinator();
assertTrue(coordinatorOpt.isPresent());
- assertEquals(Integer.MAX_VALUE - node.id(), coordinatorOpt.get().id());
+ assertEquals(node.id(), coordinatorOpt.get().id());
+ assertInstanceOf(GroupCoordinatorNode.class, coordinatorOpt.get());
+ assertEquals(node.id(),
Integer.parseInt(coordinatorOpt.get().idString()));
assertEquals(node.host(), coordinatorOpt.get().host());
assertEquals(node.port(), coordinatorOpt.get().port());
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/GroupCoordinatorNodeTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/GroupCoordinatorNodeTest.java
new file mode 100644
index 00000000000..41e5ab1c7f1
--- /dev/null
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/GroupCoordinatorNodeTest.java
@@ -0,0 +1,54 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.common.Node;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class GroupCoordinatorNodeTest {
+ @Test
+ public void testValidation() {
+ assertDoesNotThrow(() -> new Node(0, "localhost", 9092));
+ assertDoesNotThrow(() -> new Node(-1, "localhost", 9092));
+ assertDoesNotThrow(() -> new GroupCoordinatorNode(0, "localhost",
9092));
+ assertThrows(IllegalArgumentException.class, () -> new
GroupCoordinatorNode(-1, "localhost", 9092));
+ }
+
+ @Test
+ public void testIdString() {
+ Node node0 = new Node(0, "localhost", 9092);
+ Node gcNode0 = new GroupCoordinatorNode(0, "localhost", 9092);
+ Assertions.assertEquals(node0.id(), gcNode0.id());
+ assertNotEquals(node0.idString(), gcNode0.idString());
+ assertEquals(0, Integer.parseInt(gcNode0.idString()));
+ assertEquals(0, Integer.parseInt(node0.idString()));
+
+ Node node1 = new Node(1, "localhost", 9092);
+ Node gcNode1 = new GroupCoordinatorNode(1, "localhost", 9092);
+ assertEquals(node1.id(), gcNode1.id());
+ assertNotEquals(node1.idString(), gcNode1.idString());
+ assertEquals(1, Integer.parseInt(node1.idString()));
+ assertEquals(1, Integer.parseInt(gcNode1.idString()));
+ }
+}
\ No newline at end of file
diff --git
a/clients/src/testFixtures/java/org/apache/kafka/clients/MockClient.java
b/clients/src/testFixtures/java/org/apache/kafka/clients/MockClient.java
index fca0a9ca212..e24c9f4dc95 100644
--- a/clients/src/testFixtures/java/org/apache/kafka/clients/MockClient.java
+++ b/clients/src/testFixtures/java/org/apache/kafka/clients/MockClient.java
@@ -255,7 +255,6 @@ public class MockClient implements KafkaClient {
short version =
nodeApiVersions.latestUsableVersion(request.apiKey(),
builder.oldestAllowedVersion(),
builder.latestAllowedVersion());
-
AbstractRequest abstractRequest =
request.requestBuilder().build(version);
if (!futureResp.requestMatcher.matches(abstractRequest))
throw new IllegalStateException("Request matcher did not
match next-in-line request "
diff --git a/docs/getting-started/upgrade.md b/docs/getting-started/upgrade.md
index 50648c1fc22..f2594cdd3e8 100644
--- a/docs/getting-started/upgrade.md
+++ b/docs/getting-started/upgrade.md
@@ -43,6 +43,7 @@ type: docs
`kafka.server:type=group-coordinator-metrics,name=classic-group-count,state={PreparingRebalance|CompletingRebalance|Stable|Dead|Empty}`
instead of the
`kafka.coordinator.group:type=GroupMetadataManager,name=NumGroups{PreparingRebalance|CompletingRebalance|Stable|Dead|Empty}`
metrics.
For further details, please refer to
[KIP-1301](https://cwiki.apache.org/confluence/x/Z5U8G).
* The broker-side OAUTHBEARER JWT validator now fails fast at startup when a
JWKS endpoint (`sasl.oauthbearer.jwks.endpoint.url`) is configured but
`sasl.oauthbearer.expected.audience` or `sasl.oauthbearer.expected.issuer` is
not set. Brokers that previously started without these settings will now fail
to start until they are configured. To intentionally accept tokens regardless
of their audience or issuer, set the new
`sasl.oauthbearer.allow.unverified.audience` or `sasl.oauthbearer.a [...]
+ * When clients connect to the cluster, they now include cluster and node
information to enable detection and handling of misrouted connections. For
further details, please refer to
[KIP-1242](https://cwiki.apache.org/confluence/x/W4LMFw).
## Upgrading to 4.3.0
diff --git
a/server/src/test/java/org/apache/kafka/server/DefaultApiVersionManagerTest.java
b/server/src/test/java/org/apache/kafka/server/DefaultApiVersionManagerTest.java
index 99b826d9d55..b0c4cddc56f 100644
---
a/server/src/test/java/org/apache/kafka/server/DefaultApiVersionManagerTest.java
+++
b/server/src/test/java/org/apache/kafka/server/DefaultApiVersionManagerTest.java
@@ -92,15 +92,7 @@ public class DefaultApiVersionManagerTest {
);
for (ApiKeys apiKey : ApiKeys.apisForListener(apiScope)) {
- if (apiKey.id == ApiKeys.API_VERSIONS.id) {
- // ApiVersions API is a particular case. The client always
sends the highest version
- // that it supports and the server falls back to version 0 if
it does not know it.
- // See ApiKeys.isVersionEnabled for more information (KIP-511).
- // Because API_VERSIONS has an unstable version while KIP-1242
is under development,
- // we need a special case in this test. This assertion will
start failing when the
- // API is no longer unstable and the special case can be
removed.
- assertTrue(apiKey.messageType.latestVersionUnstable());
- } else if (apiKey.messageType.latestVersionUnstable()) {
+ if (apiKey.messageType.latestVersionUnstable()) {
assertFalse(
versionManager.isApiEnabled(apiKey,
apiKey.latestVersion()),
apiKey + " version " + apiKey.latestVersion() + " should
be disabled."
