You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "AndrewJSchofield (via GitHub)" <gi...@apache.org> on 2026/06/08 17:22:51 UTC
[PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
AndrewJSchofield opened a new pull request, #22512:
URL: https://github.com/apache/kafka/pull/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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3437465376
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1140,8 +1140,11 @@ private void handleInitiateApiVersionRequests(long now) {
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) {
+ // We can get the real node ID by subtracting from Integer.MAX_VALUE.
+ if (nodeId > Integer.MAX_VALUE / 2) {
Review Comment:
Yes, overriding `idString` for coordinator connections does work.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3529542901
##########
clients/src/main/java/org/apache/kafka/common/Node.java:
##########
@@ -31,30 +31,47 @@ public class Node {
private final int port;
private final String rack;
private final boolean isFenced;
+ private final boolean isCoordinator;
// Cache hashCode as it is called in performance sensitive parts of the code (e.g. RecordAccumulator.ready)
private Integer hash;
public Node(int id, String host, int port) {
- this(id, host, port, null, false);
+ this(id, host, port, null, false, false);
}
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, false);
}
public Node(int id, String host, int port, String rack, boolean isFenced) {
+ this(id, host, port, rack, isFenced, false);
+ }
+
+ public Node(int id, String host, int port, String rack, boolean isFenced, boolean isCoordinator) {
Review Comment:
opened https://issues.apache.org/jira/browse/KAFKA-20775
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3412851110
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1140,8 +1140,11 @@ private void handleInitiateApiVersionRequests(long now) {
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) {
+ // We can get the real node ID by subtracting from Integer.MAX_VALUE.
+ if (nodeId > Integer.MAX_VALUE / 2) {
Review Comment:
OK, but in that case, wouldn't the entire principle of using large node IDs for the coordinator connections be entirely broken then? We could end up a broker using a very high node ID, and another one which coincidentally has the same node ID when it is being connected as a coordinator. @dajac What do you think? What's a safe path here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3421552752
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1140,8 +1140,11 @@ private void handleInitiateApiVersionRequests(long now) {
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) {
+ // We can get the real node ID by subtracting from Integer.MAX_VALUE.
+ if (nodeId > Integer.MAX_VALUE / 2) {
Review Comment:
The coordinator id is just the broker id of the broker hosting the coordinator. Hence, it should usually be low, say < 100000. However, I agree with @rajinisivaram that this is not documented anywhere. Honestly, I don't like the `MAX - id` hack at all. I wonder if we should rather fix the underlying issue which is that we limit the number of connections by node id. Could you have a way to tell the network client that we want a connection which is not limited by the maximum number of connections allowed? I haven't check how feasible it it though.
My understanding is also that the connection to the group coordinator won't have the nodeId set which is also weird from a design perspective.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3467460009
##########
clients/src/main/java/org/apache/kafka/common/Node.java:
##########
@@ -31,30 +31,47 @@ public class Node {
private final int port;
private final String rack;
private final boolean isFenced;
+ private final boolean isCoordinator;
// Cache hashCode as it is called in performance sensitive parts of the code (e.g. RecordAccumulator.ready)
private Integer hash;
public Node(int id, String host, int port) {
- this(id, host, port, null, false);
+ this(id, host, port, null, false, false);
}
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, false);
}
public Node(int id, String host, int port, String rack, boolean isFenced) {
+ this(id, host, port, rack, isFenced, false);
+ }
+
+ public Node(int id, String host, int port, String rack, boolean isFenced, boolean isCoordinator) {
this.id = id;
- this.idString = Integer.toString(id);
+ // The contract here is that the node's id can be extracted from the node's idString by parsing as
+ // an integer. We want coordinator and non-coordinator nodes with the same id to have different
+ // idString values which parse to the same numeric id value. The Kafka network code depends upon having
+ // separate coordinator connections to brokers, and uses idString to differentiate.
+ //
+ // There are three cases:
+ // * Negative node ID - used for bootstrap, will not be a coordinator, idString will be a negative integer string
+ // * Non-negative node ID, not coordinator - idString will be a non-negative integer string with no sign
+ // * Non-negative node ID, coordinator - idString will be a non-negative integer string, with a leading '+'
+ if (isCoordinator) {
+ if (id < 0) {
+ throw new IllegalArgumentException("Node id for coordinator node cannot be negative");
+ }
+ this.idString = "+" + id;
Review Comment:
Because I want integer parsing to work. All we need is a difference. There's no need to make this more difficult I feel. I did start with `coordinator-[%d}` but then I needed a parsing method in `Node` and I decided the complexity wasn't really worth it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3467466725
##########
clients/src/main/java/org/apache/kafka/common/Node.java:
##########
@@ -31,30 +31,47 @@ public class Node {
private final int port;
private final String rack;
private final boolean isFenced;
+ private final boolean isCoordinator;
// Cache hashCode as it is called in performance sensitive parts of the code (e.g. RecordAccumulator.ready)
private Integer hash;
public Node(int id, String host, int port) {
- this(id, host, port, null, false);
+ this(id, host, port, null, false, false);
}
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, false);
}
public Node(int id, String host, int port, String rack, boolean isFenced) {
+ this(id, host, port, rack, isFenced, false);
+ }
+
+ public Node(int id, String host, int port, String rack, boolean isFenced, boolean isCoordinator) {
Review Comment:
Hmm. That's a thought. I will take a look. This is one of those PRs where I'm gradually learning about areas I've not really dug into before.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "rajinisivaram (via GitHub)" <gi...@apache.org>.
rajinisivaram commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3413314959
##########
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,
Review Comment:
Makes sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3504438590
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1138,10 +1138,7 @@ private void handleInitiateApiVersionRequests(long now) {
if (metadataRecoveryStrategy != MetadataRecoveryStrategy.NONE && metadataClusterCheckEnable) {
String clusterId = this.metadataUpdater.clusterId();
int nodeId = Integer.parseInt(node);
Review Comment:
Can't we just use `node.id` here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3414283681
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1140,8 +1140,11 @@ private void handleInitiateApiVersionRequests(long now) {
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) {
+ // We can get the real node ID by subtracting from Integer.MAX_VALUE.
+ if (nodeId > Integer.MAX_VALUE / 2) {
Review Comment:
Yes, let's. I can think of a way of improving this area if simple arithmetic is not enough.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #22512:
URL: https://github.com/apache/kafka/pull/22512
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3421610572
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1140,8 +1140,11 @@ private void handleInitiateApiVersionRequests(long now) {
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) {
+ // We can get the real node ID by subtracting from Integer.MAX_VALUE.
+ if (nodeId > Integer.MAX_VALUE / 2) {
Review Comment:
Could we override `idString` in `Node` to be `coordinator-{id}` to force the creating of a new connection without hacking the id?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "rajinisivaram (via GitHub)" <gi...@apache.org>.
rajinisivaram commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3413313565
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1140,8 +1140,11 @@ private void handleInitiateApiVersionRequests(long now) {
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) {
+ // We can get the real node ID by subtracting from Integer.MAX_VALUE.
+ if (nodeId > Integer.MAX_VALUE / 2) {
Review Comment:
I may be wrong, but I thought for the coordinator, we weren't relying on node ids being low, we were just relying on the node ids being close together, to avoid conflicts between broker node ids and the coordinator ids. Let's see what @dajac says.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3468189633
##########
clients/src/main/java/org/apache/kafka/common/Node.java:
##########
@@ -31,30 +31,47 @@ public class Node {
private final int port;
private final String rack;
private final boolean isFenced;
+ private final boolean isCoordinator;
// Cache hashCode as it is called in performance sensitive parts of the code (e.g. RecordAccumulator.ready)
private Integer hash;
public Node(int id, String host, int port) {
- this(id, host, port, null, false);
+ this(id, host, port, null, false, false);
}
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, false);
}
public Node(int id, String host, int port, String rack, boolean isFenced) {
+ this(id, host, port, rack, isFenced, false);
+ }
+
+ public Node(int id, String host, int port, String rack, boolean isFenced, boolean isCoordinator) {
Review Comment:
I've taken a look and this is a bigger piece of work. It's also not necessary in this PR because the way TransactionManager handles the Nodes is much simpler. I propose not to do this enhancement as part of this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3527975031
##########
clients/src/main/java/org/apache/kafka/common/Node.java:
##########
@@ -31,30 +31,47 @@ public class Node {
private final int port;
private final String rack;
private final boolean isFenced;
+ private final boolean isCoordinator;
// Cache hashCode as it is called in performance sensitive parts of the code (e.g. RecordAccumulator.ready)
private Integer hash;
public Node(int id, String host, int port) {
- this(id, host, port, null, false);
+ this(id, host, port, null, false, false);
}
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, false);
}
public Node(int id, String host, int port, String rack, boolean isFenced) {
+ this(id, host, port, rack, isFenced, false);
+ }
+
+ public Node(int id, String host, int port, String rack, boolean isFenced, boolean isCoordinator) {
Review Comment:
I actually like the "+" workaround. it's a very clever trick! :laughing:
Since Node is a public API, the current contract of `idString` must be maintained for backward compatibility. However, have we considered deprecating `idString` in favor of a new method that clearly reflects this new routing behavior? For example: `connectionId()`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3412835451
##########
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,
Review Comment:
While the implementation was incomplete, I marked the latest version as unstable. This change removes the unstable indication so that new RPC version is now considered stable. That's my intent anyway.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3467393646
##########
clients/src/main/java/org/apache/kafka/common/Node.java:
##########
@@ -31,30 +31,47 @@ public class Node {
private final int port;
private final String rack;
private final boolean isFenced;
+ private final boolean isCoordinator;
// Cache hashCode as it is called in performance sensitive parts of the code (e.g. RecordAccumulator.ready)
private Integer hash;
public Node(int id, String host, int port) {
- this(id, host, port, null, false);
+ this(id, host, port, null, false, false);
}
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, false);
}
public Node(int id, String host, int port, String rack, boolean isFenced) {
+ this(id, host, port, rack, isFenced, false);
+ }
+
+ public Node(int id, String host, int port, String rack, boolean isFenced, boolean isCoordinator) {
Review Comment:
I wonder if we should generalize it a bit more so we could also use it to have dedicated connection to group coordinator and transaction coordinator in the producer.
##########
clients/src/main/java/org/apache/kafka/common/Node.java:
##########
@@ -31,30 +31,47 @@ public class Node {
private final int port;
private final String rack;
private final boolean isFenced;
+ private final boolean isCoordinator;
// Cache hashCode as it is called in performance sensitive parts of the code (e.g. RecordAccumulator.ready)
private Integer hash;
public Node(int id, String host, int port) {
- this(id, host, port, null, false);
+ this(id, host, port, null, false, false);
}
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, false);
}
public Node(int id, String host, int port, String rack, boolean isFenced) {
+ this(id, host, port, rack, isFenced, false);
+ }
+
+ public Node(int id, String host, int port, String rack, boolean isFenced, boolean isCoordinator) {
this.id = id;
- this.idString = Integer.toString(id);
+ // The contract here is that the node's id can be extracted from the node's idString by parsing as
+ // an integer. We want coordinator and non-coordinator nodes with the same id to have different
+ // idString values which parse to the same numeric id value. The Kafka network code depends upon having
+ // separate coordinator connections to brokers, and uses idString to differentiate.
+ //
+ // There are three cases:
+ // * Negative node ID - used for bootstrap, will not be a coordinator, idString will be a negative integer string
+ // * Non-negative node ID, not coordinator - idString will be a non-negative integer string with no sign
+ // * Non-negative node ID, coordinator - idString will be a non-negative integer string, with a leading '+'
+ if (isCoordinator) {
+ if (id < 0) {
+ throw new IllegalArgumentException("Node id for coordinator node cannot be negative");
+ }
+ this.idString = "+" + id;
Review Comment:
why `+` vs using e.g. `coordinator-`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3421632498
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1140,8 +1140,11 @@ private void handleInitiateApiVersionRequests(long now) {
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) {
+ // We can get the real node ID by subtracting from Integer.MAX_VALUE.
+ if (nodeId > Integer.MAX_VALUE / 2) {
Review Comment:
It does not work everywhere as we either use the Node or the id as the key...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3421632498
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1140,8 +1140,11 @@ private void handleInitiateApiVersionRequests(long now) {
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) {
+ // We can get the real node ID by subtracting from Integer.MAX_VALUE.
+ if (nodeId > Integer.MAX_VALUE / 2) {
Review Comment:
It does not work everywhere as we either use the Node or the id as the key...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "rajinisivaram (via GitHub)" <gi...@apache.org>.
rajinisivaram commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3412301788
##########
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,
Review Comment:
This change is not related to the KIP?
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1140,8 +1140,11 @@ private void handleInitiateApiVersionRequests(long now) {
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) {
+ // We can get the real node ID by subtracting from Integer.MAX_VALUE.
+ if (nodeId > Integer.MAX_VALUE / 2) {
Review Comment:
There is an assumption here that node ids are less than `Integer.MAX_VALUE / 2`. But that is not a documented limitation for broker ids, right? Isn't it possible that some deployments use larger broker ids, in which case, the larger number is the actual broker id and the smaller one is the coordinator id?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #22512:
URL: https://github.com/apache/kafka/pull/22512#issuecomment-4767568443
The failure of `KafkaAdminClientTest.testUnreachableBootstrapNode` is not caused by this PR, because there's a latent bug in that test. I'll put in a fix for that too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3520256185
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1138,10 +1138,7 @@ private void handleInitiateApiVersionRequests(long now) {
if (metadataRecoveryStrategy != MetadataRecoveryStrategy.NONE && metadataClusterCheckEnable) {
String clusterId = this.metadataUpdater.clusterId();
int nodeId = Integer.parseInt(node);
Review Comment:
My bad. I thought that `node` was a `Node` :).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3529104180
##########
clients/src/main/java/org/apache/kafka/common/Node.java:
##########
@@ -31,30 +31,47 @@ public class Node {
private final int port;
private final String rack;
private final boolean isFenced;
+ private final boolean isCoordinator;
// Cache hashCode as it is called in performance sensitive parts of the code (e.g. RecordAccumulator.ready)
private Integer hash;
public Node(int id, String host, int port) {
- this(id, host, port, null, false);
+ this(id, host, port, null, false, false);
}
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, false);
}
public Node(int id, String host, int port, String rack, boolean isFenced) {
+ this(id, host, port, rack, isFenced, false);
+ }
+
+ public Node(int id, String host, int port, String rack, boolean isFenced, boolean isCoordinator) {
Review Comment:
That is a sensible idea. Since this trick was a late addition to the implementation of this KIP, I didn't do that. I wonder if you have a volunteer in mind for the writing the KIP.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20246: Add clusterId and nodeId to ApiVersionsRequest (3/N) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22512:
URL: https://github.com/apache/kafka/pull/22512#discussion_r3529307790
##########
clients/src/main/java/org/apache/kafka/common/Node.java:
##########
@@ -31,30 +31,47 @@ public class Node {
private final int port;
private final String rack;
private final boolean isFenced;
+ private final boolean isCoordinator;
// Cache hashCode as it is called in performance sensitive parts of the code (e.g. RecordAccumulator.ready)
private Integer hash;
public Node(int id, String host, int port) {
- this(id, host, port, null, false);
+ this(id, host, port, null, false, false);
}
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, false);
}
public Node(int id, String host, int port, String rack, boolean isFenced) {
+ this(id, host, port, rack, isFenced, false);
+ }
+
+ public Node(int id, String host, int port, String rack, boolean isFenced, boolean isCoordinator) {
Review Comment:
> if you have a volunteer in mind for the writing the KIP.
Yes, we definitely have folks who are interested in contributing to Kafka.
Regarding the approach, besides deprecating the old and adding the new, I do have some alternative ideas, but I think it would be best to detail and discuss them directly in a comprehensive KIP first.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org