AndrewJSchofield commented on code in PR #23544:
URL: https://github.com/apache/kafka/pull/23544#discussion_r4097683153
##########
clients/src/main/java/org/apache/kafka/clients/ClientRequest.java:
##########
@@ -52,6 +53,29 @@ public ClientRequest(String destination,
boolean expectResponse,
int requestTimeoutMs,
RequestCompletionHandler callback) {
+ this(destination, requestBuilder, correlationId, clientId,
createdTimeMs, expectResponse,
+ requestTimeoutMs, callback, null);
+ }
+
+ /**
+ * @param destination The brokerId to send the request to
+ * @param requestBuilder The builder for the request to make
+ * @param correlationId The correlation id for this client request
+ * @param clientId The client ID to use for the header
+ * @param createdTimeMs The unix timestamp in milliseconds for the time at
which this request was created.
+ * @param expectResponse Should we expect a response message or is this
request complete once it is sent?
+ * @param callback A callback to execute when the response has been
received (or null if no callback is necessary)
Review Comment:
`requestTImeoutMs` is missing from the javadoc.
##########
clients/src/main/java/org/apache/kafka/clients/ClientRequest.java:
##########
@@ -52,6 +53,29 @@ public ClientRequest(String destination,
boolean expectResponse,
int requestTimeoutMs,
RequestCompletionHandler callback) {
+ this(destination, requestBuilder, correlationId, clientId,
createdTimeMs, expectResponse,
+ requestTimeoutMs, callback, null);
+ }
+
+ /**
+ * @param destination The brokerId to send the request to
+ * @param requestBuilder The builder for the request to make
+ * @param correlationId The correlation id for this client request
+ * @param clientId The client ID to use for the header
+ * @param createdTimeMs The unix timestamp in milliseconds for the time at
which this request was created.
+ * @param expectResponse Should we expect a response message or is this
request complete once it is sent?
+ * @param callback A callback to execute when the response has been
received (or null if no callback is necessary)
+ * @param clientInstanceId The client instance ID to use for the header,
or null if this client does not have one
+ */
+ public ClientRequest(String destination,
+ AbstractRequest.Builder<?> requestBuilder,
+ int correlationId,
+ String clientId,
Review Comment:
And let's put the `clientInstanceId` just after the `clientId` for
consistency across the classes.
##########
clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java:
##########
@@ -188,6 +191,15 @@ connectionSetupTimeoutMsTest,
connectionSetupTimeoutMaxMsTest, time, false, new
MetadataRecoveryStrategy.NONE, bootstrapConfiguration, false);
}
+ private NetworkClient createNetworkClientWithClientInstanceId(Uuid
clientInstanceId) {
Review Comment:
I'd prefer to remove this method and provide a non-null client instance ID
in most cases in this test file. We can have one which provides a null instead
specifically to test that. I think we will demand non-null once the KIP
implementation is complete.
##########
clients/src/main/java/org/apache/kafka/clients/ClientRequest.java:
##########
@@ -52,6 +53,29 @@ public ClientRequest(String destination,
boolean expectResponse,
int requestTimeoutMs,
RequestCompletionHandler callback) {
+ this(destination, requestBuilder, correlationId, clientId,
createdTimeMs, expectResponse,
+ requestTimeoutMs, callback, null);
+ }
+
+ /**
+ * @param destination The brokerId to send the request to
+ * @param requestBuilder The builder for the request to make
+ * @param correlationId The correlation id for this client request
+ * @param clientId The client ID to use for the header
+ * @param createdTimeMs The unix timestamp in milliseconds for the time at
which this request was created.
+ * @param expectResponse Should we expect a response message or is this
request complete once it is sent?
+ * @param callback A callback to execute when the response has been
received (or null if no callback is necessary)
+ * @param clientInstanceId The client instance ID to use for the header,
or null if this client does not have one
+ */
+ public ClientRequest(String destination,
Review Comment:
There aren't really very many instances of `new ClientRequest` as far as I
can see, so wouldn't it be better to have one constructor, rather than adding a
new one.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -4488,4 +4491,28 @@ public static void resetCounters() {
CLOSE_COUNT.set(0);
}
}
+
+ private Map<String, Object> clientInstanceIdConfigs(GroupProtocol
groupProtocol) {
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ configs.put(ConsumerConfig.GROUP_ID_CONFIG, "group");
+ configs.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG,
groupProtocol.name().toLowerCase(Locale.ROOT));
+ return configs;
+ }
+
+ @Test
+ public void
testClassicConsumerPassesTheClientInstanceIdToTheNetworkClient() {
+ Map<String, Object> configs =
clientInstanceIdConfigs(GroupProtocol.CLASSIC);
+ ClientInstanceIdCapture.assertGenerated(
+ () -> new KafkaConsumer<>(configs, new StringDeserializer(), new
StringDeserializer()));
+ }
+
+ @Test
+ public void testAsyncConsumerPassesTheClientInstanceIdToTheNetworkClient()
{
+ Map<String, Object> configs =
clientInstanceIdConfigs(GroupProtocol.CONSUMER);
+ ClientInstanceIdCapture.assertGenerated(NetworkClientDelegate.class,
Review Comment:
This could be done in the `ClientInstanceIdCapture` class more neatly. There
are a few other instances too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
