KAFKA-17813: Moving broker endpoint class and common server connection id #17519

apoorvmittal10 · 2024-10-16T21:40:53Z

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

chia7712 · 2024-10-17T06:23:37Z

@apoorvmittal10 I love this cleanup and migration. Could you please rebase code to fix conflicts as #17473 is already merged

…n id

apoorvmittal10 · 2024-10-17T07:44:17Z

@apoorvmittal10 I love this cleanup and migration. Could you please rebase code to fix conflicts as #17473 is already merged

Thanks @chia7712, sure. I ll force push the branch once #17474 is also merged, as the changes are on top of that. Hence keeping the PR in draft mode till then.

apoorvmittal10 · 2024-10-17T07:50:06Z

@chia7712 Yeah, sorry #17474 is not required for this PR. I have rebased it and marked for review. Please if you could take a look.

apoorvmittal10 · 2024-10-17T11:42:51Z

Looking into the tests failure.

brandboat

Thanks for the refactor !!!

clients/src/test/java/org/apache/kafka/common/network/BrokerEndPointTest.java

clients/src/main/java/org/apache/kafka/common/network/BrokerEndPoint.java

chia7712

@apoorvmittal10 thanks for this refactor

chia7712 · 2024-10-17T12:31:38Z

clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java

+        }
+
+        try {
+            return BrokerEndPoint.parseHostPort(split[0])


return BrokerEndPoint.parseHostPort(split[0]) .flatMap(localHost -> BrokerEndPoint.parseHostPort(split[1]) .map(remoteHost -> new ServerConnectionId(localHost, remoteHost, Integer.parseInt(split[2]), Integer.parseInt(split[3]))));

we don't need to wrap it by Optional.of

Additionally, BrokerEndPoint.parseHostPort is used by ServerConnectionId only, so maybe we can inline it?

Thanks for the suggestion, I have addressed the first comment. Sorry I didn't understand the inline comment.

I think Chia is suggesting to move BrokerEndPoint.parseHostPort as a static method in this class. Since the connection string doesn't contain brokerId. It doesn't quite fit into BrokerEndPoint.

I have moved it same class now.

chia7712 · 2024-10-17T12:53:02Z

core/src/main/scala/kafka/network/SocketServer.scala

@@ -1221,7 +1206,7 @@ private[kafka] class Processor(
  private def processDisconnected(): Unit = {
    selector.disconnected.keySet.forEach { connectionId =>
      try {
-        val remoteHost = ConnectionId.fromString(connectionId).getOrElse {
+        val remoteHost = ServerConnectionId.fromString(connectionId).orElseThrow { () =>
          throw new IllegalStateException(s"connectionId has unexpected format: $connectionId")


Maybe ServerConnectionId.fromString can throw IllegalStateException instead of returning Optional?

I was trying to maintain least refactor while keeping original context. I ll look into this as well.

apoorvmittal10 · 2024-10-17T17:00:26Z

@chia7712 @brandboat Thanks for the review, can you please re-look.

brandboat

LGTM, thanks !

junrao

@apoorvmittal10 : Thanks for the PR. Left a few comments.

junrao · 2024-10-17T17:38:09Z

clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java

+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.network;


This is only used on the server. Could we put it under the server-common module?

Make sense, but then we have to remove the generateConnectionId method from this class as that's used by NioEchoServer. And if we remove that then there is no consolidation between Selector.java and SocketServer.scala related connectionId methods. Wdyt?

Got it. We can keep it in the client module then.

junrao · 2024-10-17T17:41:09Z

clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java

+/**
+ * ServerConnectionId is used to uniquely identify a connection between a client and a server.
+ */
+public class ServerConnectionId {


ServerConnectionId => ConnectionId ?

If we keep this class in the client module, it's better to give it a more accurate name. Perhaps ConnectionIdOnServer?

Hmmm, I somehow felt ServerConnectionId is better as it prefixes with server itself. I can write the class description appropraitely. If you still think it's not appropriate then I ll move it to ConnectionIdOnServer. Please let me know.

ServerConnectionId needs to remain in the clients module since it's also used by the clients' test code. Perhaps we could rewrite NioEchoServer to use a different connection ID scheme to decouple it. In #16460, we wanted to add a more meaningful string, but that's not necessary for aligning with the core server code.

@apoorvmittal10 what do you think abort moving ServerConnectionId to server module?

@chia7712 That we can't do, as we discussed that here: #17519 (comment)

junrao · 2024-10-17T17:53:25Z

clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java

+        try {
+            return BrokerEndPoint.parseHostPort(split[0])
+                .flatMap(localHost -> BrokerEndPoint.parseHostPort(split[1])
+                    .map(remoteHost -> new ServerConnectionId(localHost, remoteHost, Integer.parseInt(split[2]), Integer.parseInt(split[3]))));


indentation

I have changed the way it's written now.

junrao · 2024-10-17T17:54:29Z

clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java

+        }
+
+        try {
+            return BrokerEndPoint.parseHostPort(split[0])


I think Chia is suggesting to move BrokerEndPoint.parseHostPort as a static method in this class. Since the connection string doesn't contain brokerId. It doesn't quite fit into BrokerEndPoint.

junrao · 2024-10-17T18:06:13Z

clients/src/main/java/org/apache/kafka/common/network/BrokerEndPoint.java

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.network;


This is only used on the server. Could we put it under the server-common module?

As the other class is in clients uses BrokerEnpoint now henc kept it in clients itself, under network. As that's the only place I found to aggregate where clients and server related classes can go.

Moved to server-common as we have now moved to map entry as per @chia7712's comment.

apoorvmittal10 · 2024-10-17T20:19:42Z

I have addressed the comments, 1 with the name is pending. If the name doesn't seems right for ServerConnectionId then I ll change it. Though I have added the class description.

chia7712 · 2024-10-18T03:26:17Z

clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java

+     * Note that unlike EndPoint (or listener) this URI has no security information.
+     */
+    // Visible for testing
+    static Optional<BrokerEndPoint> parseHostPort(String connectionString) {


Could you please replace BrokerEndPoint by Map.Entry? we are under java 11, so it is easier to create Entry now.

Done, though not a big fan of Map.Entry but seems sensible here as there was no broker id for BrokerEndpoint.

apoorvmittal10 · 2024-10-18T13:31:29Z

I have resolved the comments, build passed.

junrao

@apoorvmittal10 : Thanks for the updated PR. A couple of more comments.

junrao · 2024-10-18T17:30:17Z

core/src/main/scala/kafka/network/SocketServer.scala

@@ -25,7 +25,7 @@ import java.util
 import java.util.Optional
 import java.util.concurrent._
 import java.util.concurrent.atomic._
-import kafka.cluster.{BrokerEndPoint, EndPoint}
+import kafka.cluster.{EndPoint}


We can remove the brackets since there is a single import.

Missed it, thought spotlessApply will fix. Done.

junrao · 2024-10-18T17:59:56Z

clients/src/test/java/org/apache/kafka/common/network/ServerConnectionIdTest.java

+        assertEquals(9092, brokerEndPoint.get().port());
+
+        // IPv6 endpoint
+        brokerEndPoint = ServerConnectionId.parseHostPort("[2001:db8::1]:9092");


Hmm, an interesting thing is that if we take a ServerConnectionId with host name of [2001:db8::1] and call ServerConnectionId.fromString, we will get a slightly different ServerConnectionId since the hostname will not have the bracket. It's not an issue now, but it may cause confusion in the future.

ServerConnectionId doesn't face ambiguity like a URL does, so in my opinion, the hostname in ServerConnectionId doesn't need brackets.

Do I need to dix anything in this? I have just used the exisiting regex for parsing, from scala to java.

Yeah, this is not required. I have pushed an additional test for same.

chia7712

LGTM

apoorvmittal10 · 2024-10-21T17:22:44Z

@junrao Thanks for the feedback and review. Please let me know if I need to address more. Build has passes for the PR.

junrao

@apoorvmittal10 : Thanks for the updated PR. A couple of more comments.

junrao · 2024-10-21T17:20:43Z

clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java

+ */
+public class ServerConnectionId {
+
+    private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");


A couple of things.

Utils.HOST_PORT_PATTERN is more general than this. We could just reuse the former instead.

It seems that we have an existing bug. InetAddress.getHostAddress() doesn't contain brackets. When we generate a parsable string for host:port, we need to add brackets for the host part since it can contain colon. For example, we need to do that for bootstrap.servers. However, in Processor, when generating a connection string, we just concatenate the host and the port without adding the brackets. This means that when we try to convert the connection string back to ConnectionId, we can't parse the hostname properly if it's IPV6. To fix this, it seems that we need to add brackets to the hostname when generating the connection string.

Utils.HOST_PORT_PATTERN is more general than this. We could just reuse the former instead.

I initially looked at it but didn't use that as it also has protocol parsing which we don't require. And moreover I thought to port same scala version to java.

This means that when we try to convert the connection string back to ConnectionId, we can't parse the hostname properly if it's IPV6. To fix this, it seems that we need to add brackets to the hostname when generating the connection string.

I don't think it's a bug, I am writing a test case for same as well. The regex splits on the end : hence the host and port will be constructed correctly.

I have pushed a test case as well for ipv6.

Thanks for the explanation. It's fine to have a separate regex just for ServerConnectionId. If so, the host name in the connection string will never have brackets in it, right? Then, it would be useful to change the regex, the comment and the tests to remove the support on brackets accordingly.

If so, the host name in the connection string will never have brackets in it, right?

I am not sure about this but the code handles with or without brackets. I think it's the right behaviour. The test cases I wrote additionally verifies that.

Also just trying to be consistent to the scala version of the previous code.

I am not sure about this but the code handles with or without brackets. I think it's the right behaviour.

Well, we create the connection string in two paths. The first one gets the hostname from InetAddress.getHostAddress(), which shouldn't include brackets according to https://groups.google.com/g/golang-dev/c/AiC3MffM4-U. The second one gets the hostname from ServerConnectionId.parseHostPort() which stripes out the brackets if present.

Sure @junrao, I have made the change.

apoorvmittal10 · 2024-10-22T16:09:23Z

@junrao I have addressed the comment, removed brackets, corrected tests and comments. Please can you review.

junrao

@apoorvmittal10 : Thanks for the updated PR. Just a minor comment.

junrao · 2024-10-22T16:28:30Z

clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java

+    // The regex for parsing the URI string. The URI string should be in the format of "host:port",
+    //  where host can be an IPv4 address or an IPv6 address.
+    //  Note: The IPv6 address should not be enclosed in square brackets.
+    private static final Pattern URI_PARSE_EXP = Pattern.compile("([0-9a-zA-Z\\-%._:]*):([0-9]+)");


Since this is not really a URI, should we rename URI_PARSE_EXP to sth like HOST_PORT_PARSE_EXP?

junrao

@apoorvmittal10 : Thanks for the updated PR. LGTM

…n id (apache#17519) Reviewers: Chia-Ping Tsai <[email protected]>, Kuan-Po Tseng <[email protected]>, Jun Rao <[email protected]>

github-actions bot added core Kafka Broker performance clients labels Oct 16, 2024

KAFKA-17813: Moving broker endpoint class and common server connectio…

7526c60

…n id

apoorvmittal10 force-pushed the KAFKA-17813 branch from 94ac636 to 7526c60 Compare October 17, 2024 07:48

apoorvmittal10 marked this pull request as ready for review October 17, 2024 07:50

apoorvmittal10 requested review from junrao and chia7712 October 17, 2024 07:50

brandboat reviewed Oct 17, 2024

View reviewed changes

clients/src/test/java/org/apache/kafka/common/network/BrokerEndPointTest.java Outdated Show resolved Hide resolved

clients/src/main/java/org/apache/kafka/common/network/BrokerEndPoint.java Outdated Show resolved Hide resolved

chia7712 reviewed Oct 17, 2024

View reviewed changes

apoorvmittal10 added 2 commits October 17, 2024 17:57

Addressing review comments

9cdc316

Merge remote-tracking branch 'upstream/trunk' into KAFKA-17813

57b5a67

apoorvmittal10 requested a review from brandboat October 17, 2024 17:00

apoorvmittal10 requested a review from chia7712 October 17, 2024 17:24

brandboat approved these changes Oct 17, 2024

View reviewed changes

junrao reviewed Oct 17, 2024

View reviewed changes

Addressing review comments

3ad638a

apoorvmittal10 requested a review from junrao October 17, 2024 20:18

chia7712 reviewed Oct 18, 2024

View reviewed changes

Moving to map entry

2efc9d2

apoorvmittal10 requested a review from chia7712 October 18, 2024 13:31

junrao reviewed Oct 18, 2024

View reviewed changes

Correcting imports

13ac770

apoorvmittal10 requested a review from junrao October 20, 2024 10:22

chia7712 approved these changes Oct 20, 2024

View reviewed changes

junrao reviewed Oct 21, 2024

View reviewed changes

Adding ipv6 additional test

7702b58

apoorvmittal10 requested a review from junrao October 21, 2024 19:03

apoorvmittal10 added 2 commits October 22, 2024 13:29

Removing brackets from regex

09ba603

Correcting comment

b2ce627

junrao reviewed Oct 22, 2024

View reviewed changes

Addressing minor comments

b419758

junrao approved these changes Oct 22, 2024

View reviewed changes

junrao merged commit 25a3590 into apache:trunk Oct 22, 2024
6 checks passed

KAFKA-17813: Moving broker endpoint class and common server connection id #17519

KAFKA-17813: Moving broker endpoint class and common server connection id #17519

Conversation

apoorvmittal10 commented Oct 16, 2024

Committer Checklist (excluded from commit message)

chia7712 commented Oct 17, 2024

apoorvmittal10 commented Oct 17, 2024

apoorvmittal10 commented Oct 17, 2024

apoorvmittal10 commented Oct 17, 2024

brandboat left a comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apoorvmittal10 commented Oct 17, 2024

brandboat left a comment

Choose a reason for hiding this comment

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apoorvmittal10 commented Oct 17, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apoorvmittal10 commented Oct 18, 2024

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

apoorvmittal10 commented Oct 21, 2024

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apoorvmittal10 commented Oct 22, 2024

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

junrao left a comment

Choose a reason for hiding this comment