You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "bbejeck (via GitHub)" <gi...@apache.org> on 2024/11/01 21:19:22 UTC
[PR] MINOR: Update clientInstanceIds from EOS_V2 refactor [kafka]
bbejeck opened a new pull request, #17664:
URL: https://github.com/apache/kafka/pull/17664
Updates `KafkaStreams.clientInstanceIds` method to correctly populate the client-id -> clientInstanceId map that was altered in a previous refactoring.
Added a test that confirms `ClientInstanceIds` is correctly storing consumer and producer instance ids
### 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] MINOR: Update clientInstanceIds from EOS_V2 refactor [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck merged PR #17664:
URL: https://github.com/apache/kafka/pull/17664
--
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] MINOR: Update clientInstanceIds from EOS_V2 refactor [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #17664:
URL: https://github.com/apache/kafka/pull/17664#discussion_r1826568967
##########
streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java:
##########
@@ -1668,6 +1669,33 @@ public void shouldReturnAdminInstanceID() {
}
}
+ @Test
+ public void shouldReturnProducerAndConsumerInstanceIds() {
+ prepareStreams();
+ prepareStreamThread(streamThreadOne, 1);
+ prepareStreamThread(streamThreadTwo, 2);
+ final Uuid mainConsumerInstanceId = Uuid.randomUuid();
+ final Uuid producerInstanceId = Uuid.randomUuid();
+ final KafkaFutureImpl<Uuid> consumerFuture = new KafkaFutureImpl<>();
+ final KafkaFutureImpl<Uuid> producerFuture = new KafkaFutureImpl<>();
+ consumerFuture.complete(mainConsumerInstanceId);
+ producerFuture.complete(producerInstanceId);
+ final Uuid adminInstanceId = Uuid.randomUuid();
+ adminClient.setClientInstanceId(adminInstanceId);
+
+ final Map<String, KafkaFuture<Uuid>> expectedClientIds = Map.of("main-consumer", consumerFuture, "some-thread-producer-id", producerFuture);
+ when(streamThreadOne.clientInstanceIds(any())).thenReturn(expectedClientIds);
Review Comment:
Fixed, unfamiliarity with test.
--
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] MINOR: Update clientInstanceIds from EOS_V2 refactor [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #17664:
URL: https://github.com/apache/kafka/pull/17664#discussion_r1826569227
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -1929,10 +1930,18 @@ public synchronized ClientInstanceIds clientInstanceIds(final Duration timeout)
// could be `null` if telemetry is disabled on the consumer itself
if (instanceId != null) {
- clientInstanceIds.addConsumerInstanceId(
- clientFuture.getKey(),
- instanceId
- );
+ final String clientFutureKey = clientFuture.getKey();
+ if (clientFutureKey.toLowerCase(Locale.getDefault()).contains("producer")) {
Review Comment:
I was aiming for something more generic if the thread names were to change, but that's unlikely, so I'm updating it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Update clientInstanceIds from EOS_V2 refactor [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #17664:
URL: https://github.com/apache/kafka/pull/17664#issuecomment-2453010760
Merged #17664 into trunk
--
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] MINOR: Update clientInstanceIds from EOS_V2 refactor [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17664:
URL: https://github.com/apache/kafka/pull/17664#discussion_r1826430616
##########
streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java:
##########
@@ -1668,6 +1669,33 @@ public void shouldReturnAdminInstanceID() {
}
}
+ @Test
+ public void shouldReturnProducerAndConsumerInstanceIds() {
+ prepareStreams();
+ prepareStreamThread(streamThreadOne, 1);
+ prepareStreamThread(streamThreadTwo, 2);
+ final Uuid mainConsumerInstanceId = Uuid.randomUuid();
+ final Uuid producerInstanceId = Uuid.randomUuid();
+ final KafkaFutureImpl<Uuid> consumerFuture = new KafkaFutureImpl<>();
+ final KafkaFutureImpl<Uuid> producerFuture = new KafkaFutureImpl<>();
+ consumerFuture.complete(mainConsumerInstanceId);
+ producerFuture.complete(producerInstanceId);
+ final Uuid adminInstanceId = Uuid.randomUuid();
+ adminClient.setClientInstanceId(adminInstanceId);
Review Comment:
Why do we need `adminClient` and `adminInstanceId` - seems both are unused below.
##########
streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java:
##########
@@ -1668,6 +1669,33 @@ public void shouldReturnAdminInstanceID() {
}
}
+ @Test
+ public void shouldReturnProducerAndConsumerInstanceIds() {
+ prepareStreams();
+ prepareStreamThread(streamThreadOne, 1);
+ prepareStreamThread(streamThreadTwo, 2);
+ final Uuid mainConsumerInstanceId = Uuid.randomUuid();
+ final Uuid producerInstanceId = Uuid.randomUuid();
+ final KafkaFutureImpl<Uuid> consumerFuture = new KafkaFutureImpl<>();
+ final KafkaFutureImpl<Uuid> producerFuture = new KafkaFutureImpl<>();
+ consumerFuture.complete(mainConsumerInstanceId);
+ producerFuture.complete(producerInstanceId);
+ final Uuid adminInstanceId = Uuid.randomUuid();
+ adminClient.setClientInstanceId(adminInstanceId);
+
+ final Map<String, KafkaFuture<Uuid>> expectedClientIds = Map.of("main-consumer", consumerFuture, "some-thread-producer-id", producerFuture);
+ when(streamThreadOne.clientInstanceIds(any())).thenReturn(expectedClientIds);
Review Comment:
Seems we are only using `streamThreadOne` here? Why do we `prepareStreamThread(streamThreadTwo, 2)` above?
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -1929,10 +1930,18 @@ public synchronized ClientInstanceIds clientInstanceIds(final Duration timeout)
// could be `null` if telemetry is disabled on the consumer itself
if (instanceId != null) {
- clientInstanceIds.addConsumerInstanceId(
- clientFuture.getKey(),
- instanceId
- );
+ final String clientFutureKey = clientFuture.getKey();
+ if (clientFutureKey.toLowerCase(Locale.getDefault()).contains("producer")) {
Review Comment:
Should we look for `.endsWith("-producer")` instead of using `contains()`?
--
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] MINOR: Update clientInstanceIds from EOS_V2 refactor [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #17664:
URL: https://github.com/apache/kafka/pull/17664#discussion_r1826568522
##########
streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java:
##########
@@ -1668,6 +1669,33 @@ public void shouldReturnAdminInstanceID() {
}
}
+ @Test
+ public void shouldReturnProducerAndConsumerInstanceIds() {
+ prepareStreams();
+ prepareStreamThread(streamThreadOne, 1);
+ prepareStreamThread(streamThreadTwo, 2);
+ final Uuid mainConsumerInstanceId = Uuid.randomUuid();
+ final Uuid producerInstanceId = Uuid.randomUuid();
+ final KafkaFutureImpl<Uuid> consumerFuture = new KafkaFutureImpl<>();
+ final KafkaFutureImpl<Uuid> producerFuture = new KafkaFutureImpl<>();
+ consumerFuture.complete(mainConsumerInstanceId);
+ producerFuture.complete(producerInstanceId);
+ final Uuid adminInstanceId = Uuid.randomUuid();
+ adminClient.setClientInstanceId(adminInstanceId);
Review Comment:
Test throws an exception when not set, I've added it to the assertions
--
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