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/01/14 00:30:51 UTC

[PR] KAFKA-15811: Enhance request context with client socket port information (KIP-714) [kafka]

apoorvmittal10 opened a new pull request, #15190:
URL: https://github.com/apache/kafka/pull/15190

   PR adds support to capture client socket port information in Request Context. The port from request context is used as matching criteria in filtering clients and shall be used by metrics plugin to fetch port from request context.
   
   ### 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-15811: Enhance request context with client socket port information (KIP-714) [kafka]

Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #15190:
URL: https://github.com/apache/kafka/pull/15190#discussion_r1451631637


##########
clients/src/main/java/org/apache/kafka/common/requests/RequestContext.java:
##########
@@ -37,6 +37,7 @@ public class RequestContext implements AuthorizableRequestContext {
     public final RequestHeader header;
     public final String connectionId;
     public final InetAddress clientAddress;
+    public final Optional<Integer> clientPort;

Review Comment:
   I have kept the client port as Optional because the data should not be required for forwarded requests in context of KIP-714.



-- 
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-15811: Enhance request context with client socket port information (KIP-714) [kafka]

Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #15190:
URL: https://github.com/apache/kafka/pull/15190#discussion_r1458055273


##########
server/src/test/java/org/apache/kafka/server/metrics/ClientMetricsTestUtils.java:
##########
@@ -59,6 +60,7 @@ public static RequestContext requestContext() throws UnknownHostException {
             new RequestHeader(ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS, (short) 0, "producer-1", 0),
             "1",
             InetAddress.getLocalHost(),
+            Optional.of(56078),

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-15811: Enhance request context with client socket port information (KIP-714) [kafka]

Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #15190:
URL: https://github.com/apache/kafka/pull/15190#discussion_r1456303419


##########
core/src/main/scala/kafka/network/SocketServer.scala:
##########
@@ -1140,9 +1141,9 @@ private[kafka] class Processor(
                 expiredConnectionsKilledCount.record(null, 1, 0)
               } else {
                 val connectionId = receive.source
-                val context = new RequestContext(header, connectionId, channel.socketAddress,
-                  channel.principal, listenerName, securityProtocol,
-                  channel.channelMetadataRegistry.clientInformation, isPrivilegedListener, channel.principalSerde)
+                val context = new RequestContext(header, connectionId, channel.socketAddress, Optional.of(channel.socketPort()),

Review Comment:
   Hmm, channel.socketPort() is never null. Why do we need to convert it to an Optional?



##########
server/src/test/java/org/apache/kafka/server/metrics/ClientMetricsTestUtils.java:
##########
@@ -59,6 +60,7 @@ public static RequestContext requestContext() throws UnknownHostException {
             new RequestHeader(ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS, (short) 0, "producer-1", 0),
             "1",
             InetAddress.getLocalHost(),
+            Optional.of(56078),

Review Comment:
   Could we define a constant for 56078 and reuse?



-- 
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-15811: Enhance request context with client socket port information (KIP-714) [kafka]

Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #15190:
URL: https://github.com/apache/kafka/pull/15190#discussion_r1453676869


##########
clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java:
##########
@@ -369,6 +369,16 @@ public InetAddress socketAddress() {
         return transportLayer.socketChannel().socket().getInetAddress();
     }
 
+    /**
+     * Returns the port to which this channel's socket is connected or `null` if the socket has never been connected.

Review Comment:
   Thanks for pointing out, it should be 0 instead of null. 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-15811: Enhance request context with client socket port information (KIP-714) [kafka]

Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on PR #15190:
URL: https://github.com/apache/kafka/pull/15190#issuecomment-1900901612

   Thanks for triaging the test failures, @apoorvmittal10. Merging the PR 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-15811: Enhance request context with client socket port information (KIP-714) [kafka]

Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #15190:
URL: https://github.com/apache/kafka/pull/15190#issuecomment-1890806724

   @junrao @AndrewJSchofield Please 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-15811: Enhance request context with client socket port information (KIP-714) [kafka]

Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #15190:
URL: https://github.com/apache/kafka/pull/15190#discussion_r1457900006


##########
core/src/main/scala/kafka/network/SocketServer.scala:
##########
@@ -1140,9 +1141,9 @@ private[kafka] class Processor(
                 expiredConnectionsKilledCount.record(null, 1, 0)
               } else {
                 val connectionId = receive.source
-                val context = new RequestContext(header, connectionId, channel.socketAddress,
-                  channel.principal, listenerName, securityProtocol,
-                  channel.channelMetadataRegistry.clientInformation, isPrivilegedListener, channel.principalSerde)
+                val context = new RequestContext(header, connectionId, channel.socketAddress, Optional.of(channel.socketPort()),

Review Comment:
   I find RequestContext is either created in SocketServer or during forwarding requests. We currently don't require client port information outside KIP-714, push telemetry request is not marked forwardable, wiring the client port information elsewhere seems not useful at this point of time hence I marked it as Optional.



-- 
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-15811: Enhance request context with client socket port information (KIP-714) [kafka]

Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #15190:
URL: https://github.com/apache/kafka/pull/15190#issuecomment-1900762271

   > @apoorvmittal10 : Thanks for the updated PR. LGTM. Are the test failures related? Are they all being tracked?
   
   @junrao The tests failures are unrelated and below are the details regarding failed tests. Most of the were existing and I have created jira for couple of missing ones.
   
   `Existing`: Build / JDK 8 and Scala 2.12 / testWithGroupId(String).quorum=zk – kafka.api.TransactionsBounceTest
   14s - https://issues.apache.org/jira/browse/KAFKA-15930
   `Existing`: Build / JDK 8 and Scala 2.12 / "testUnrecoverableErrors(Errors).error=UNKNOWN_SERVER_ERROR" – kafka.coordinator.transaction.ProducerIdManagerTest
   <1s - https://issues.apache.org/jira/browse/KAFKA-15915
   `Existing`: Build / JDK 8 and Scala 2.12 / testDescribeUnderReplicatedPartitionsWhenReassignmentIsInProgress(String).kraft – org.apache.kafka.tools.TopicCommandIntegrationTest
   2s - https://issues.apache.org/jira/browse/KAFKA-16071
   `Existing`: Build / JDK 8 and Scala 2.12 / testDescribeUnderReplicatedPartitionsWhenReassignmentIsInProgress(String).kraft – org.apache.kafka.tools.TopicCommandIntegrationTest
   1s - https://issues.apache.org/jira/browse/KAFKA-16071
   `Existing`: Build / JDK 21 and Scala 2.13 / testReplicateSourceDefault() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationSSLTest
   1m 39s - https://issues.apache.org/jira/browse/KAFKA-15926
   `Existing`: Build / JDK 21 and Scala 2.13 / testBumpTransactionalEpoch(String).quorum=kraft+kip848 – kafka.api.TransactionsTest
   13s - https://issues.apache.org/jira/browse/KAFKA-15786
   `Created`: Build / JDK 21 and Scala 2.13 / testProducerConsumerOverrideLowerQuota(String).quorum=kraft – kafka.api.UserQuotaTest
   41s - https://issues.apache.org/jira/browse/KAFKA-16172
   `Created`: Build / JDK 21 and Scala 2.13 / testTimeoutMetrics() – org.apache.kafka.controller.QuorumControllerMetricsIntegrationTest
   <1s - https://issues.apache.org/jira/browse/KAFKA-16173
   `Existing`: Build / JDK 21 and Scala 2.13 / testTaskRequestWithOldStartMsGetsUpdated() – org.apache.kafka.trogdor.coordinator.CoordinatorTest
   4m 0s - https://issues.apache.org/jira/browse/KAFKA-8115
   `Existing`: Build / JDK 17 and Scala 2.13 / testReplicateSourceDefault() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationSSLTest
   2m 23s - https://issues.apache.org/jira/browse/KAFKA-15927
   `Existing`: Build / JDK 17 and Scala 2.13 / testBumpTransactionalEpoch(String).quorum=kraft – org.apache.kafka.tiered.storage.integration.TransactionsWithTieredStoreTest
   40s - https://issues.apache.org/jira/browse/KAFKA-15786
   `Created`: Build / JDK 17 and Scala 2.13 / testDescribeQuorumStatusSuccessful [6] Type=Raft-Isolated, MetadataVersion=3.8-IV0, Security=PLAINTEXT – org.apache.kafka.tools.MetadataQuorumCommandTest
   1m 7s - https://issues.apache.org/jira/browse/KAFKA-16174
   `Created`: Build / JDK 17 and Scala 2.13 / testDescribeQuorumStatusSuccessful [1] Type=Raft-Combined, MetadataVersion=3.8-IV0, Security=PLAINTEXT – org.apache.kafka.tools.MetadataQuorumCommandTest
   1m 9s - https://issues.apache.org/jira/browse/KAFKA-16174
   `Existing`: Build / JDK 11 and Scala 2.13 / testSeparateOffsetsTopic – org.apache.kafka.connect.integration.ExactlyOnceSourceIntegrationTest
   2m 32s - https://issues.apache.org/jira/browse/KAFKA-14089
   `Existing`: Build / JDK 11 and Scala 2.13 / testConsumptionWithBrokerFailures() – kafka.api.ConsumerBounceTest
   34s - https://issues.apache.org/jira/browse/KAFKA-15146
   `Created`: Build / JDK 11 and Scala 2.13 / "testSendToPartitionWithFollowerShutdownShouldNotTimeout(String).quorum=kraft" – kafka.api.PlaintextProducerSendTest
   1m 7s - https://issues.apache.org/jira/browse/KAFKA-16176
   `Created`: Build / JDK 11 and Scala 2.13 / testBatchSizeZeroNoPartitionNoRecordKey(String).quorum=kraft – kafka.api.PlaintextProducerSendTest - https://issues.apache.org/jira/browse/KAFKA-16177
   1m 13s
   `Created`: Build / JDK 11 and Scala 2.13 / testAsynchronousAuthorizerAclUpdatesDontBlockRequestThreads() – kafka.api.SslAdminIntegrationTest
   5s - https://issues.apache.org/jira/browse/KAFKA-16175
   `Existing`: Build / JDK 11 and Scala 2.13 / testSynchronousAuthorizerAclUpdatesBlockRequestThreads() – kafka.api.SslAdminIntegrationTest
   6s - https://issues.apache.org/jira/browse/KAFKA-9496
   `Existing`: Build / JDK 11 and Scala 2.13 / testThreadPoolResize() – kafka.server.DynamicBrokerReconfigurationTest
   30s Related to: https://issues.apache.org/jira/browse/KAFKA-7957
   `Existing`: Build / JDK 11 and Scala 2.13 / testFenceMultipleBrokers() – org.apache.kafka.controller.QuorumControllerTest
   48s - https://issues.apache.org/jira/browse/KAFKA-15898
   `Existing`: Build / JDK 11 and Scala 2.13 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest - https://issues.apache.org/jira/browse/KAFKA-15928


-- 
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-15811: Enhance request context with client socket port information (KIP-714) [kafka]

Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #15190:
URL: https://github.com/apache/kafka/pull/15190#discussion_r1453668720


##########
clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java:
##########
@@ -369,6 +369,16 @@ public InetAddress socketAddress() {
         return transportLayer.socketChannel().socket().getInetAddress();
     }
 
+    /**
+     * Returns the port to which this channel's socket is connected or `null` if the socket has never been connected.

Review Comment:
   nit: This cannot return `null` if the socket has never been connected.



##########
clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java:
##########
@@ -433,7 +437,7 @@ private void handleSaslToken(byte[] clientToken) throws IOException {
             RequestHeader header = RequestHeader.parse(requestBuffer);
             ApiKeys apiKey = header.apiKey();
             short version = header.apiVersion();
-            RequestContext requestContext = new RequestContext(header, connectionId, clientAddress(),
+            RequestContext requestContext = new RequestContext(header, connectionId, clientAddress(), Optional.of(clientPort()),

Review Comment:
   I'm not quite sure the optionality is right here. Shouldn't there be the option of `Optional.empty()` if there's no data?



-- 
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-15811: Enhance request context with client socket port information (KIP-714) [kafka]

Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #15190:
URL: https://github.com/apache/kafka/pull/15190#discussion_r1453684188


##########
clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java:
##########
@@ -433,7 +437,7 @@ private void handleSaslToken(byte[] clientToken) throws IOException {
             RequestHeader header = RequestHeader.parse(requestBuffer);
             ApiKeys apiKey = header.apiKey();
             short version = header.apiVersion();
-            RequestContext requestContext = new RequestContext(header, connectionId, clientAddress(),
+            RequestContext requestContext = new RequestContext(header, connectionId, clientAddress(), Optional.of(clientPort()),

Review Comment:
   Hmmm, the reason I kept clientPort Optional in RequestContext is because the data is not required in case of forwarded requests but wherever it can be determined by explicit value then that has been wrapped. I am not of an opinion to make a decision of returning Optional.empty in case connection is closed or socket exception occurred as that scenario ll anyways not make the channel usable.  



-- 
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-15811: Enhance request context with client socket port information (KIP-714) [kafka]

Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #15190:
URL: https://github.com/apache/kafka/pull/15190#issuecomment-1899343398

   > @apoorvmittal10 : Thanks for the PR. A couple of minor comments.
   
   Thanks for reviewing @junrao. I have addressed and replied to the 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-15811: Enhance request context with client socket port information (KIP-714) [kafka]

Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao merged PR #15190:
URL: https://github.com/apache/kafka/pull/15190


-- 
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