You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "apoorvmittal10 (via GitHub)" <gi...@apache.org> on 2024/10/16 21:40:55 UTC
[PR] KAFKA-17813: Moving broker endpoint class and common server connection id [kafka]
apoorvmittal10 opened a new pull request, #17519:
URL: https://github.com/apache/kafka/pull/17519
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #17519:
URL: https://github.com/apache/kafka/pull/17519#issuecomment-2418628755
@apoorvmittal10 I love this cleanup and migration. Could you please rebase code to fix conflicts as #17473 is already merged
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805786774
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,143 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection on server for the client. The
+ * connection id is in the format of "localHost:localPort-remoteHost:remotePort-processorId-index".
+ * The processorId is the id of the processor that will handle this connection and the index is
+ * used to ensure uniqueness.
+ */
+public class ServerConnectionId {
+
+ private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");
+
+ private final String localHost;
+ private final int localPort;
+ private final String remoteHost;
+ private final int remotePort;
+ private final int processorId;
+ private final int index;
+
+ public ServerConnectionId(
+ BrokerEndPoint localEndpoint,
+ BrokerEndPoint remoteEndpoint,
+ int processorId,
+ int index
+ ) {
+ this(localEndpoint.host(), localEndpoint.port(), remoteEndpoint.host(), remoteEndpoint.port(), processorId, index);
+ }
+
+ public ServerConnectionId(
+ String localHost,
+ int localPort,
+ String remoteHost,
+ int remotePort,
+ int processorId,
+ int index
+ ) {
+ this.localHost = localHost;
+ this.localPort = localPort;
+ this.remoteHost = remoteHost;
+ this.remotePort = remotePort;
+ this.processorId = processorId;
+ this.index = index;
+ }
+
+ public String localHost() {
+ return localHost;
+ }
+
+ public int localPort() {
+ return localPort;
+ }
+
+ public String remoteHost() {
+ return remoteHost;
+ }
+
+ public int remotePort() {
+ return remotePort;
+ }
+
+ public int processorId() {
+ return processorId;
+ }
+
+ public int index() {
+ return index;
+ }
+
+ /**
+ * Returns an optional ServerConnectionId object from the given connection ID string.
+ *
+ * @param connectionIdString The connection ID string to parse.
+ * @return An optional ServerConnectionId object.
+ */
+ public static Optional<ServerConnectionId> fromString(String connectionIdString) {
+ String[] split = connectionIdString.split("-");
+ if (split.length != 4) {
+ return Optional.empty();
+ }
+
+ try {
+ return parseHostPort(split[0]).flatMap(localHost -> parseHostPort(split[1]).map(
+ remoteHost -> new ServerConnectionId(localHost, remoteHost, Integer.parseInt(split[2]), Integer.parseInt(split[3]))));
+ } catch (NumberFormatException e) {
+ return Optional.empty();
+ }
+ }
+
+ /**
+ * Generates a unique connection ID for the given socket.
+ *
+ * @param socket The socket for which the connection ID is to be generated.
+ * @param processorId The ID of the server processor that will handle this connection.
+ * @param connectionIndex The index to be used in the connection ID to ensure uniqueness.
+ * @return A string representing the unique connection ID.
+ */
+ public static String generateConnectionId(Socket socket, int processorId, int connectionIndex) {
+ String localHost = socket.getLocalAddress().getHostAddress();
+ int localPort = socket.getLocalPort();
+ String remoteHost = socket.getInetAddress().getHostAddress();
+ int remotePort = socket.getPort();
+ return localHost + ":" + localPort + "-" + remoteHost + ":" + remotePort + "-" + processorId + "-" + connectionIndex;
+ }
+
+ /**
+ * BrokerEndPoint URI is host:port or [ipv6_host]:port
+ * Note that unlike EndPoint (or listener) this URI has no security information.
+ */
+ // Visible for testing
+ static Optional<BrokerEndPoint> parseHostPort(String connectionString) {
Review Comment:
Could you please replace `BrokerEndPoint` by `Map.Entry`? we are under java 11, so it is easier to create `Entry` now.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805238289
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
Review Comment:
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?
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17519:
URL: https://github.com/apache/kafka/pull/17519#issuecomment-2419297950
Looking into the tests failure.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1806834183
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Optional;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection between a client and a server.
+ */
+public class ServerConnectionId {
Review Comment:
@chia7712 That we can't do, as we discussed that here: https://github.com/apache/kafka/pull/17519#discussion_r1805238289
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1809389043
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,143 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Map;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection on server for the client. The
+ * connection id is in the format of "localHost:localPort-remoteHost:remotePort-processorId-index".
+ * The processorId is the id of the processor that will handle this connection and the index is
+ * used to ensure uniqueness.
+ */
+public class ServerConnectionId {
+
+ private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");
Review Comment:
Also just trying to be consistent to the scala version of the previous code.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805119370
##########
clients/src/test/java/org/apache/kafka/common/network/BrokerEndPointTest.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.common.network;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class BrokerEndPointTest {
+
+ @Test
+ public void testParseHostPort() {
+ Optional<BrokerEndPoint> brokerEndPoint = BrokerEndPoint.parseHostPort("myhost:9092");
+ assertTrue(brokerEndPoint.isPresent());
+ assertEquals("myhost", brokerEndPoint.get().host());
+ assertEquals(9092, brokerEndPoint.get().port());
+
+ brokerEndPoint = BrokerEndPoint.parseHostPort("127.0.0.1:9092");
Review Comment:
Added a trst case.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17519:
URL: https://github.com/apache/kafka/pull/17519#issuecomment-2429697155
@junrao I have addressed the comment, removed brackets, corrected tests and comments. Please can you review.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805165552
##########
clients/src/main/java/org/apache/kafka/common/network/BrokerEndPoint.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.common.network;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * BrokerEndPoint is used to connect to specific host:port pair.
+ * It is typically used by clients (or brokers when connecting to other brokers)
+ * and contains no information about the security protocol used on the connection.
+ * Clients should know which security protocol to use from configuration.
+ * This allows us to keep the wire protocol with the clients unchanged where the protocol is not needed.
+ */
+public class BrokerEndPoint {
+
+ private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");
Review Comment:
Yeah, it's a good test file (BrokerEndPointTest.scala). I thought to port that too in java but it needs some zk specific scala classes to parse and uses additional scala classes which I cannot reference.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1806806625
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Optional;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection between a client and a server.
+ */
+public class ServerConnectionId {
Review Comment:
@apoorvmittal10 what do you think abort moving `ServerConnectionId` to server module?
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1807759687
##########
clients/src/test/java/org/apache/kafka/common/network/ServerConnectionIdTest.java:
##########
@@ -110,4 +110,38 @@ public void testGenerateConnectionId() throws IOException {
assertEquals("127.0.0.1:9092-127.0.0.1:9093-0-0", ServerConnectionId.generateConnectionId(socket, 0, 0));
assertEquals("127.0.0.1:9092-127.0.0.1:9093-1-2", ServerConnectionId.generateConnectionId(socket, 1, 2));
}
+
+ @Test
+ public void testParseHostPort() {
+ Optional<BrokerEndPoint> brokerEndPoint = ServerConnectionId.parseHostPort("myhost:9092");
+ assertTrue(brokerEndPoint.isPresent());
+ assertEquals("myhost", brokerEndPoint.get().host());
+ assertEquals(9092, brokerEndPoint.get().port());
+
+ brokerEndPoint = ServerConnectionId.parseHostPort("127.0.0.1:9092");
+ assertTrue(brokerEndPoint.isPresent());
+ assertEquals("127.0.0.1", brokerEndPoint.get().host());
+ assertEquals(9092, brokerEndPoint.get().port());
+
+ // IPv6 endpoint
+ brokerEndPoint = ServerConnectionId.parseHostPort("[2001:db8::1]:9092");
Review Comment:
Do I need to dix anything in this? I have just used the exisiting regex for parsing, from scala to java.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17519:
URL: https://github.com/apache/kafka/pull/17519#issuecomment-2418814655
@chia7712 Yeah, sorry https://github.com/apache/kafka/pull/17474 is not required for this PR. I have rebased it and marked for review. Please if you could take a look.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17519:
URL: https://github.com/apache/kafka/pull/17519#issuecomment-2422486410
I have resolved the comments, build passed.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1810634576
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,143 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Map;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection on server for the client. The
+ * connection id is in the format of "localHost:localPort-remoteHost:remotePort-processorId-index".
+ * The processorId is the id of the processor that will handle this connection and the index is
+ * used to ensure uniqueness.
+ */
+public class ServerConnectionId {
+
+ private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");
Review Comment:
Sure @junrao, I have made the change.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1809343716
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,143 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Map;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection on server for the client. The
+ * connection id is in the format of "localHost:localPort-remoteHost:remotePort-processorId-index".
+ * The processorId is the id of the processor that will handle this connection and the index is
+ * used to ensure uniqueness.
+ */
+public class ServerConnectionId {
+
+ private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");
Review Comment:
I have pushed a test case as well for ipv6.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805129240
##########
clients/src/main/java/org/apache/kafka/common/network/BrokerEndPoint.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.common.network;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * BrokerEndPoint is used to connect to specific host:port pair.
+ * It is typically used by clients (or brokers when connecting to other brokers)
+ * and contains no information about the security protocol used on the connection.
+ * Clients should know which security protocol to use from configuration.
+ * This allows us to keep the wire protocol with the clients unchanged where the protocol is not needed.
+ */
+public class BrokerEndPoint {
+
+ private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");
Review Comment:
Thanks for explanation ! I also notice that we already have a BrokerEndPointTest.scala which cover some tests... Anyway, thanks for your patience.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805240188
##########
clients/src/main/java/org/apache/kafka/common/network/BrokerEndPoint.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.common.network;
Review Comment:
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.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1806340078
##########
clients/src/main/java/org/apache/kafka/common/network/BrokerEndPoint.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.common.network;
Review Comment:
Moved to server-common as we have now moved to map entry as per @chia7712's comment.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17519:
URL: https://github.com/apache/kafka/pull/17519#issuecomment-2427306123
@junrao Thanks for the feedback and review. Please let me know if I need to address more. Build has passes for the 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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1809540887
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,143 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Map;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection on server for the client. The
+ * connection id is in the format of "localHost:localPort-remoteHost:remotePort-processorId-index".
+ * The processorId is the id of the processor that will handle this connection and the index is
+ * used to ensure uniqueness.
+ */
+public class ServerConnectionId {
+
+ private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");
Review Comment:
> 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.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805119370
##########
clients/src/test/java/org/apache/kafka/common/network/BrokerEndPointTest.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.common.network;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class BrokerEndPointTest {
+
+ @Test
+ public void testParseHostPort() {
+ Optional<BrokerEndPoint> brokerEndPoint = BrokerEndPoint.parseHostPort("myhost:9092");
+ assertTrue(brokerEndPoint.isPresent());
+ assertEquals("myhost", brokerEndPoint.get().host());
+ assertEquals(9092, brokerEndPoint.get().port());
+
+ brokerEndPoint = BrokerEndPoint.parseHostPort("127.0.0.1:9092");
Review Comment:
Added a test case.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805260456
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Optional;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection between a client and a server.
+ */
+public class ServerConnectionId {
Review Comment:
If we keep this class in the client module, it's better to give it a more accurate name. Perhaps ConnectionIdOnServer?
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805786334
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Optional;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection between a client and a server.
+ */
+public class ServerConnectionId {
Review Comment:
`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.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17519:
URL: https://github.com/apache/kafka/pull/17519#issuecomment-2418803774
> @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 https://github.com/apache/kafka/pull/17474 is also merged, as the changes are on top of that. Hence keeping the PR in draft mode till then.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1809342565
##########
clients/src/test/java/org/apache/kafka/common/network/ServerConnectionIdTest.java:
##########
@@ -110,4 +110,38 @@ public void testGenerateConnectionId() throws IOException {
assertEquals("127.0.0.1:9092-127.0.0.1:9093-0-0", ServerConnectionId.generateConnectionId(socket, 0, 0));
assertEquals("127.0.0.1:9092-127.0.0.1:9093-1-2", ServerConnectionId.generateConnectionId(socket, 1, 2));
}
+
+ @Test
+ public void testParseHostPort() {
+ Optional<BrokerEndPoint> brokerEndPoint = ServerConnectionId.parseHostPort("myhost:9092");
+ assertTrue(brokerEndPoint.isPresent());
+ assertEquals("myhost", brokerEndPoint.get().host());
+ assertEquals(9092, brokerEndPoint.get().port());
+
+ brokerEndPoint = ServerConnectionId.parseHostPort("127.0.0.1:9092");
+ assertTrue(brokerEndPoint.isPresent());
+ assertEquals("127.0.0.1", brokerEndPoint.get().host());
+ assertEquals(9092, brokerEndPoint.get().port());
+
+ // IPv6 endpoint
+ brokerEndPoint = ServerConnectionId.parseHostPort("[2001:db8::1]:9092");
Review Comment:
Yeah, this is not required. I have pushed an additional test for same.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1811045144
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Map;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection on server for the client. The
+ * connection id is in the format of "localHost:localPort-remoteHost:remotePort-processorId-index".
+ * The processorId is the id of the processor that will handle this connection and the index is
+ * used to ensure uniqueness.
+ */
+public class ServerConnectionId {
+
+ // 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]+)");
Review Comment:
Since this is not really a URI, should we rename URI_PARSE_EXP to sth like HOST_PORT_PARSE_EXP?
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1806888599
##########
clients/src/test/java/org/apache/kafka/common/network/ServerConnectionIdTest.java:
##########
@@ -110,4 +110,38 @@ public void testGenerateConnectionId() throws IOException {
assertEquals("127.0.0.1:9092-127.0.0.1:9093-0-0", ServerConnectionId.generateConnectionId(socket, 0, 0));
assertEquals("127.0.0.1:9092-127.0.0.1:9093-1-2", ServerConnectionId.generateConnectionId(socket, 1, 2));
}
+
+ @Test
+ public void testParseHostPort() {
+ Optional<BrokerEndPoint> brokerEndPoint = ServerConnectionId.parseHostPort("myhost:9092");
+ assertTrue(brokerEndPoint.isPresent());
+ assertEquals("myhost", brokerEndPoint.get().host());
+ assertEquals(9092, brokerEndPoint.get().port());
+
+ brokerEndPoint = ServerConnectionId.parseHostPort("127.0.0.1:9092");
+ assertTrue(brokerEndPoint.isPresent());
+ assertEquals("127.0.0.1", brokerEndPoint.get().host());
+ assertEquals(9092, brokerEndPoint.get().port());
+
+ // IPv6 endpoint
+ brokerEndPoint = ServerConnectionId.parseHostPort("[2001:db8::1]:9092");
Review Comment:
`ServerConnectionId` doesn't face ambiguity like a URL does, so in my opinion, the hostname in `ServerConnectionId` doesn't need brackets.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1807758132
##########
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}
Review Comment:
Missed it, thought spotlessApply will fix. Done.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1804734544
##########
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")
Review Comment:
I was trying to maintain least refactor while keeping original context. I ll look into this as well.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao merged PR #17519:
URL: https://github.com/apache/kafka/pull/17519
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805129240
##########
clients/src/main/java/org/apache/kafka/common/network/BrokerEndPoint.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.common.network;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * BrokerEndPoint is used to connect to specific host:port pair.
+ * It is typically used by clients (or brokers when connecting to other brokers)
+ * and contains no information about the security protocol used on the connection.
+ * Clients should know which security protocol to use from configuration.
+ * This allows us to keep the wire protocol with the clients unchanged where the protocol is not needed.
+ */
+public class BrokerEndPoint {
+
+ private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");
Review Comment:
Thanks for explanation ! I also notice that we already have a BrokerEndPointTest.scala which cover ipv6 tests... Anyway, thanks for your patience.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805378855
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Optional;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection between a client and a server.
+ */
+public class ServerConnectionId {
+
+ private final String localHost;
+ private final int localPort;
+ private final String remoteHost;
+ private final int remotePort;
+ private final int processorId;
+ private final int index;
+
+ public ServerConnectionId(
+ BrokerEndPoint localEndpoint,
+ BrokerEndPoint remoteEndpoint,
+ int processorId,
+ int index
+ ) {
+ this(localEndpoint.host(), localEndpoint.port(), remoteEndpoint.host(), remoteEndpoint.port(), processorId, index);
+ }
+
+ public ServerConnectionId(
+ String localHost,
+ int localPort,
+ String remoteHost,
+ int remotePort,
+ int processorId,
+ int index
+ ) {
+ this.localHost = localHost;
+ this.localPort = localPort;
+ this.remoteHost = remoteHost;
+ this.remotePort = remotePort;
+ this.processorId = processorId;
+ this.index = index;
+ }
+
+ public String localHost() {
+ return localHost;
+ }
+
+ public int localPort() {
+ return localPort;
+ }
+
+ public String remoteHost() {
+ return remoteHost;
+ }
+
+ public int remotePort() {
+ return remotePort;
+ }
+
+ public int processorId() {
+ return processorId;
+ }
+
+ public int index() {
+ return index;
+ }
+
+ /**
+ * Returns an optional ServerConnectionId object from the given connection ID string.
+ *
+ * @param connectionIdString The connection ID string to parse.
+ * @return An optional ServerConnectionId object.
+ */
+ public static Optional<ServerConnectionId> fromString(String connectionIdString) {
+ String[] split = connectionIdString.split("-");
+ if (split.length != 4) {
+ return Optional.empty();
+ }
+
+ 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]))));
Review Comment:
I have changed the way it's written now.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805379365
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Optional;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection between a client and a server.
+ */
+public class ServerConnectionId {
+
+ private final String localHost;
+ private final int localPort;
+ private final String remoteHost;
+ private final int remotePort;
+ private final int processorId;
+ private final int index;
+
+ public ServerConnectionId(
+ BrokerEndPoint localEndpoint,
+ BrokerEndPoint remoteEndpoint,
+ int processorId,
+ int index
+ ) {
+ this(localEndpoint.host(), localEndpoint.port(), remoteEndpoint.host(), remoteEndpoint.port(), processorId, index);
+ }
+
+ public ServerConnectionId(
+ String localHost,
+ int localPort,
+ String remoteHost,
+ int remotePort,
+ int processorId,
+ int index
+ ) {
+ this.localHost = localHost;
+ this.localPort = localPort;
+ this.remoteHost = remoteHost;
+ this.remotePort = remotePort;
+ this.processorId = processorId;
+ this.index = index;
+ }
+
+ public String localHost() {
+ return localHost;
+ }
+
+ public int localPort() {
+ return localPort;
+ }
+
+ public String remoteHost() {
+ return remoteHost;
+ }
+
+ public int remotePort() {
+ return remotePort;
+ }
+
+ public int processorId() {
+ return processorId;
+ }
+
+ public int index() {
+ return index;
+ }
+
+ /**
+ * Returns an optional ServerConnectionId object from the given connection ID string.
+ *
+ * @param connectionIdString The connection ID string to parse.
+ * @return An optional ServerConnectionId object.
+ */
+ public static Optional<ServerConnectionId> fromString(String connectionIdString) {
+ String[] split = connectionIdString.split("-");
+ if (split.length != 4) {
+ return Optional.empty();
+ }
+
+ try {
+ return BrokerEndPoint.parseHostPort(split[0])
Review Comment:
I have moved it same class now.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17519:
URL: https://github.com/apache/kafka/pull/17519#issuecomment-2420478408
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.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805120617
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Optional;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection between a client and a server.
+ */
+public class ServerConnectionId {
+
+ private final String localHost;
+ private final int localPort;
+ private final String remoteHost;
+ private final int remotePort;
+ private final int processorId;
+ private final int index;
+
+ public ServerConnectionId(
+ BrokerEndPoint localEndpoint,
+ BrokerEndPoint remoteEndpoint,
+ int processorId,
+ int index
+ ) {
+ this(localEndpoint.host(), localEndpoint.port(), remoteEndpoint.host(), remoteEndpoint.port(), processorId, index);
+ }
+
+ public ServerConnectionId(
+ String localHost,
+ int localPort,
+ String remoteHost,
+ int remotePort,
+ int processorId,
+ int index
+ ) {
+ this.localHost = localHost;
+ this.localPort = localPort;
+ this.remoteHost = remoteHost;
+ this.remotePort = remotePort;
+ this.processorId = processorId;
+ this.index = index;
+ }
+
+ public String localHost() {
+ return localHost;
+ }
+
+ public int localPort() {
+ return localPort;
+ }
+
+ public String remoteHost() {
+ return remoteHost;
+ }
+
+ public int remotePort() {
+ return remotePort;
+ }
+
+ public int processorId() {
+ return processorId;
+ }
+
+ public int index() {
+ return index;
+ }
+
+ /**
+ * Returns an optional ServerConnectionId object from the given connection ID string.
+ *
+ * @param connectionIdString The connection ID string to parse.
+ * @return An optional ServerConnectionId object.
+ */
+ public static Optional<ServerConnectionId> fromString(String connectionIdString) {
+ String[] split = connectionIdString.split("-");
+ if (split.length != 4) {
+ return Optional.empty();
+ }
+
+ try {
+ return BrokerEndPoint.parseHostPort(split[0])
Review Comment:
Thanks for the suggestion, I have addressed the first comment. Sorry I didn't understand the `inline` comment.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17519:
URL: https://github.com/apache/kafka/pull/17519#issuecomment-2420055066
@chia7712 @brandboat Thanks for the review, can you please re-look.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1806816662
##########
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}
Review Comment:
We can remove the brackets since there is a single import.
##########
clients/src/test/java/org/apache/kafka/common/network/ServerConnectionIdTest.java:
##########
@@ -110,4 +110,38 @@ public void testGenerateConnectionId() throws IOException {
assertEquals("127.0.0.1:9092-127.0.0.1:9093-0-0", ServerConnectionId.generateConnectionId(socket, 0, 0));
assertEquals("127.0.0.1:9092-127.0.0.1:9093-1-2", ServerConnectionId.generateConnectionId(socket, 1, 2));
}
+
+ @Test
+ public void testParseHostPort() {
+ Optional<BrokerEndPoint> brokerEndPoint = ServerConnectionId.parseHostPort("myhost:9092");
+ assertTrue(brokerEndPoint.isPresent());
+ assertEquals("myhost", brokerEndPoint.get().host());
+ assertEquals(9092, brokerEndPoint.get().port());
+
+ brokerEndPoint = ServerConnectionId.parseHostPort("127.0.0.1:9092");
+ assertTrue(brokerEndPoint.isPresent());
+ assertEquals("127.0.0.1", brokerEndPoint.get().host());
+ assertEquals(9092, brokerEndPoint.get().port());
+
+ // IPv6 endpoint
+ brokerEndPoint = ServerConnectionId.parseHostPort("[2001:db8::1]:9092");
Review Comment:
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.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1809383914
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,143 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Map;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection on server for the client. The
+ * connection id is in the format of "localHost:localPort-remoteHost:remotePort-processorId-index".
+ * The processorId is the id of the processor that will handle this connection and the index is
+ * used to ensure uniqueness.
+ */
+public class ServerConnectionId {
+
+ private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");
Review Comment:
>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.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1809211995
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,143 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Map;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection on server for the client. The
+ * connection id is in the format of "localHost:localPort-remoteHost:remotePort-processorId-index".
+ * The processorId is the id of the processor that will handle this connection and the index is
+ * used to ensure uniqueness.
+ */
+public class ServerConnectionId {
+
+ private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");
Review Comment:
A couple of things.
1. `Utils.HOST_PORT_PATTERN` is more general than this. We could just reuse the former instead.
2. 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.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1809325469
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,143 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Map;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection on server for the client. The
+ * connection id is in the format of "localHost:localPort-remoteHost:remotePort-processorId-index".
+ * The processorId is the id of the processor that will handle this connection and the index is
+ * used to ensure uniqueness.
+ */
+public class ServerConnectionId {
+
+ private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");
Review Comment:
> 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.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1811075335
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Map;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection on server for the client. The
+ * connection id is in the format of "localHost:localPort-remoteHost:remotePort-processorId-index".
+ * The processorId is the id of the processor that will handle this connection and the index is
+ * used to ensure uniqueness.
+ */
+public class ServerConnectionId {
+
+ // 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]+)");
Review Comment:
Done.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1809351498
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,143 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Map;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection on server for the client. The
+ * connection id is in the format of "localHost:localPort-remoteHost:remotePort-processorId-index".
+ * The processorId is the id of the processor that will handle this connection and the index is
+ * used to ensure uniqueness.
+ */
+public class ServerConnectionId {
+
+ private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");
Review Comment:
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.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1804730052
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Optional;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection between a client and a server.
+ */
+public class ServerConnectionId {
+
+ private final String localHost;
+ private final int localPort;
+ private final String remoteHost;
+ private final int remotePort;
+ private final int processorId;
+ private final int index;
+
+ public ServerConnectionId(
+ BrokerEndPoint localEndpoint,
+ BrokerEndPoint remoteEndpoint,
+ int processorId,
+ int index
+ ) {
+ this(localEndpoint.host(), localEndpoint.port(), remoteEndpoint.host(), remoteEndpoint.port(), processorId, index);
+ }
+
+ public ServerConnectionId(
+ String localHost,
+ int localPort,
+ String remoteHost,
+ int remotePort,
+ int processorId,
+ int index
+ ) {
+ this.localHost = localHost;
+ this.localPort = localPort;
+ this.remoteHost = remoteHost;
+ this.remotePort = remotePort;
+ this.processorId = processorId;
+ this.index = index;
+ }
+
+ public String localHost() {
+ return localHost;
+ }
+
+ public int localPort() {
+ return localPort;
+ }
+
+ public String remoteHost() {
+ return remoteHost;
+ }
+
+ public int remotePort() {
+ return remotePort;
+ }
+
+ public int processorId() {
+ return processorId;
+ }
+
+ public int index() {
+ return index;
+ }
+
+ /**
+ * Returns an optional ServerConnectionId object from the given connection ID string.
+ *
+ * @param connectionIdString The connection ID string to parse.
+ * @return An optional ServerConnectionId object.
+ */
+ public static Optional<ServerConnectionId> fromString(String connectionIdString) {
+ String[] split = connectionIdString.split("-");
+ if (split.length != 4) {
+ return Optional.empty();
+ }
+
+ try {
+ return BrokerEndPoint.parseHostPort(split[0])
Review Comment:
Additionally, `BrokerEndPoint.parseHostPort` is used by `ServerConnectionId` only, so maybe we can inline it?
##########
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")
Review Comment:
Maybe `ServerConnectionId.fromString` can throw `IllegalStateException` instead of returning `Optional`?
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Optional;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection between a client and a server.
+ */
+public class ServerConnectionId {
+
+ private final String localHost;
+ private final int localPort;
+ private final String remoteHost;
+ private final int remotePort;
+ private final int processorId;
+ private final int index;
+
+ public ServerConnectionId(
+ BrokerEndPoint localEndpoint,
+ BrokerEndPoint remoteEndpoint,
+ int processorId,
+ int index
+ ) {
+ this(localEndpoint.host(), localEndpoint.port(), remoteEndpoint.host(), remoteEndpoint.port(), processorId, index);
+ }
+
+ public ServerConnectionId(
+ String localHost,
+ int localPort,
+ String remoteHost,
+ int remotePort,
+ int processorId,
+ int index
+ ) {
+ this.localHost = localHost;
+ this.localPort = localPort;
+ this.remoteHost = remoteHost;
+ this.remotePort = remotePort;
+ this.processorId = processorId;
+ this.index = index;
+ }
+
+ public String localHost() {
+ return localHost;
+ }
+
+ public int localPort() {
+ return localPort;
+ }
+
+ public String remoteHost() {
+ return remoteHost;
+ }
+
+ public int remotePort() {
+ return remotePort;
+ }
+
+ public int processorId() {
+ return processorId;
+ }
+
+ public int index() {
+ return index;
+ }
+
+ /**
+ * Returns an optional ServerConnectionId object from the given connection ID string.
+ *
+ * @param connectionIdString The connection ID string to parse.
+ * @return An optional ServerConnectionId object.
+ */
+ public static Optional<ServerConnectionId> fromString(String connectionIdString) {
+ String[] split = connectionIdString.split("-");
+ if (split.length != 4) {
+ return Optional.empty();
+ }
+
+ try {
+ return BrokerEndPoint.parseHostPort(split[0])
Review Comment:
```java
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`
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805119858
##########
clients/src/main/java/org/apache/kafka/common/network/BrokerEndPoint.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.common.network;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * BrokerEndPoint is used to connect to specific host:port pair.
+ * It is typically used by clients (or brokers when connecting to other brokers)
+ * and contains no information about the security protocol used on the connection.
+ * Clients should know which security protocol to use from configuration.
+ * This allows us to keep the wire protocol with the clients unchanged where the protocol is not needed.
+ */
+public class BrokerEndPoint {
+
+ private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");
Review Comment:
This is not needed, additional escape for `]` is not needed in java.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805181683
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Optional;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection between a client and a server.
+ */
+public class ServerConnectionId {
+
+ private final String localHost;
+ private final int localPort;
+ private final String remoteHost;
+ private final int remotePort;
+ private final int processorId;
+ private final int index;
+
+ public ServerConnectionId(
+ BrokerEndPoint localEndpoint,
+ BrokerEndPoint remoteEndpoint,
+ int processorId,
+ int index
+ ) {
+ this(localEndpoint.host(), localEndpoint.port(), remoteEndpoint.host(), remoteEndpoint.port(), processorId, index);
+ }
+
+ public ServerConnectionId(
+ String localHost,
+ int localPort,
+ String remoteHost,
+ int remotePort,
+ int processorId,
+ int index
+ ) {
+ this.localHost = localHost;
+ this.localPort = localPort;
+ this.remoteHost = remoteHost;
+ this.remotePort = remotePort;
+ this.processorId = processorId;
+ this.index = index;
+ }
+
+ public String localHost() {
+ return localHost;
+ }
+
+ public int localPort() {
+ return localPort;
+ }
+
+ public String remoteHost() {
+ return remoteHost;
+ }
+
+ public int remotePort() {
+ return remotePort;
+ }
+
+ public int processorId() {
+ return processorId;
+ }
+
+ public int index() {
+ return index;
+ }
+
+ /**
+ * Returns an optional ServerConnectionId object from the given connection ID string.
+ *
+ * @param connectionIdString The connection ID string to parse.
+ * @return An optional ServerConnectionId object.
+ */
+ public static Optional<ServerConnectionId> fromString(String connectionIdString) {
+ String[] split = connectionIdString.split("-");
+ if (split.length != 4) {
+ return Optional.empty();
+ }
+
+ 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]))));
Review Comment:
indentation
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
Review Comment:
This is only used on the server. Could we put it under the server-common module?
##########
clients/src/main/java/org/apache/kafka/common/network/BrokerEndPoint.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.common.network;
Review Comment:
This is only used on the server. Could we put it under the server-common module?
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Optional;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection between a client and a server.
+ */
+public class ServerConnectionId {
Review Comment:
ServerConnectionId => ConnectionId ?
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Optional;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection between a client and a server.
+ */
+public class ServerConnectionId {
+
+ private final String localHost;
+ private final int localPort;
+ private final String remoteHost;
+ private final int remotePort;
+ private final int processorId;
+ private final int index;
+
+ public ServerConnectionId(
+ BrokerEndPoint localEndpoint,
+ BrokerEndPoint remoteEndpoint,
+ int processorId,
+ int index
+ ) {
+ this(localEndpoint.host(), localEndpoint.port(), remoteEndpoint.host(), remoteEndpoint.port(), processorId, index);
+ }
+
+ public ServerConnectionId(
+ String localHost,
+ int localPort,
+ String remoteHost,
+ int remotePort,
+ int processorId,
+ int index
+ ) {
+ this.localHost = localHost;
+ this.localPort = localPort;
+ this.remoteHost = remoteHost;
+ this.remotePort = remotePort;
+ this.processorId = processorId;
+ this.index = index;
+ }
+
+ public String localHost() {
+ return localHost;
+ }
+
+ public int localPort() {
+ return localPort;
+ }
+
+ public String remoteHost() {
+ return remoteHost;
+ }
+
+ public int remotePort() {
+ return remotePort;
+ }
+
+ public int processorId() {
+ return processorId;
+ }
+
+ public int index() {
+ return index;
+ }
+
+ /**
+ * Returns an optional ServerConnectionId object from the given connection ID string.
+ *
+ * @param connectionIdString The connection ID string to parse.
+ * @return An optional ServerConnectionId object.
+ */
+ public static Optional<ServerConnectionId> fromString(String connectionIdString) {
+ String[] split = connectionIdString.split("-");
+ if (split.length != 4) {
+ return Optional.empty();
+ }
+
+ try {
+ return BrokerEndPoint.parseHostPort(split[0])
Review 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`.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805257747
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
Review Comment:
Got it. We can keep it in the client module then.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1805354371
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Optional;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection between a client and a server.
+ */
+public class ServerConnectionId {
Review Comment:
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.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1804616740
##########
clients/src/main/java/org/apache/kafka/common/network/BrokerEndPoint.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.common.network;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * BrokerEndPoint is used to connect to specific host:port pair.
+ * It is typically used by clients (or brokers when connecting to other brokers)
+ * and contains no information about the security protocol used on the connection.
+ * Clients should know which security protocol to use from configuration.
+ * This allows us to keep the wire protocol with the clients unchanged where the protocol is not needed.
+ */
+public class BrokerEndPoint {
+
+ private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");
Review Comment:
This seems slightly different from the original Scala version, perhaps this should be `\\[?([0-9a-zA-Z\\-%._:]*)\\]?:([0-9]+)` ?
##########
clients/src/test/java/org/apache/kafka/common/network/BrokerEndPointTest.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.common.network;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class BrokerEndPointTest {
+
+ @Test
+ public void testParseHostPort() {
+ Optional<BrokerEndPoint> brokerEndPoint = BrokerEndPoint.parseHostPort("myhost:9092");
+ assertTrue(brokerEndPoint.isPresent());
+ assertEquals("myhost", brokerEndPoint.get().host());
+ assertEquals(9092, brokerEndPoint.get().port());
+
+ brokerEndPoint = BrokerEndPoint.parseHostPort("127.0.0.1:9092");
Review Comment:
It would be great if we can add one more `[ipv6_host]:port` test case.
--
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-17813: Moving broker endpoint class and common server connection id [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17519:
URL: https://github.com/apache/kafka/pull/17519#discussion_r1806340848
##########
clients/src/main/java/org/apache/kafka/common/network/ServerConnectionId.java:
##########
@@ -0,0 +1,143 @@
+/*
+ * 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.common.network;
+
+import java.net.Socket;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * ServerConnectionId is used to uniquely identify a connection on server for the client. The
+ * connection id is in the format of "localHost:localPort-remoteHost:remotePort-processorId-index".
+ * The processorId is the id of the processor that will handle this connection and the index is
+ * used to ensure uniqueness.
+ */
+public class ServerConnectionId {
+
+ private static final Pattern URI_PARSE_EXP = Pattern.compile("\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)");
+
+ private final String localHost;
+ private final int localPort;
+ private final String remoteHost;
+ private final int remotePort;
+ private final int processorId;
+ private final int index;
+
+ public ServerConnectionId(
+ BrokerEndPoint localEndpoint,
+ BrokerEndPoint remoteEndpoint,
+ int processorId,
+ int index
+ ) {
+ this(localEndpoint.host(), localEndpoint.port(), remoteEndpoint.host(), remoteEndpoint.port(), processorId, index);
+ }
+
+ public ServerConnectionId(
+ String localHost,
+ int localPort,
+ String remoteHost,
+ int remotePort,
+ int processorId,
+ int index
+ ) {
+ this.localHost = localHost;
+ this.localPort = localPort;
+ this.remoteHost = remoteHost;
+ this.remotePort = remotePort;
+ this.processorId = processorId;
+ this.index = index;
+ }
+
+ public String localHost() {
+ return localHost;
+ }
+
+ public int localPort() {
+ return localPort;
+ }
+
+ public String remoteHost() {
+ return remoteHost;
+ }
+
+ public int remotePort() {
+ return remotePort;
+ }
+
+ public int processorId() {
+ return processorId;
+ }
+
+ public int index() {
+ return index;
+ }
+
+ /**
+ * Returns an optional ServerConnectionId object from the given connection ID string.
+ *
+ * @param connectionIdString The connection ID string to parse.
+ * @return An optional ServerConnectionId object.
+ */
+ public static Optional<ServerConnectionId> fromString(String connectionIdString) {
+ String[] split = connectionIdString.split("-");
+ if (split.length != 4) {
+ return Optional.empty();
+ }
+
+ try {
+ return parseHostPort(split[0]).flatMap(localHost -> parseHostPort(split[1]).map(
+ remoteHost -> new ServerConnectionId(localHost, remoteHost, Integer.parseInt(split[2]), Integer.parseInt(split[3]))));
+ } catch (NumberFormatException e) {
+ return Optional.empty();
+ }
+ }
+
+ /**
+ * Generates a unique connection ID for the given socket.
+ *
+ * @param socket The socket for which the connection ID is to be generated.
+ * @param processorId The ID of the server processor that will handle this connection.
+ * @param connectionIndex The index to be used in the connection ID to ensure uniqueness.
+ * @return A string representing the unique connection ID.
+ */
+ public static String generateConnectionId(Socket socket, int processorId, int connectionIndex) {
+ String localHost = socket.getLocalAddress().getHostAddress();
+ int localPort = socket.getLocalPort();
+ String remoteHost = socket.getInetAddress().getHostAddress();
+ int remotePort = socket.getPort();
+ return localHost + ":" + localPort + "-" + remoteHost + ":" + remotePort + "-" + processorId + "-" + connectionIndex;
+ }
+
+ /**
+ * BrokerEndPoint URI is host:port or [ipv6_host]:port
+ * Note that unlike EndPoint (or listener) this URI has no security information.
+ */
+ // Visible for testing
+ static Optional<BrokerEndPoint> parseHostPort(String connectionString) {
Review Comment:
Done, though not a big fan of Map.Entry but seems sensible here as there was no broker id for BrokerEndpoint.
--
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