aliehsaeedii commented on code in PR #22778:
URL: https://github.com/apache/kafka/pull/22778#discussion_r3640194830
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2EndpointToPartitionsIntegrationTest.java:
##########
@@ -110,6 +110,7 @@ public void shouldGetCorrectHostPartitionInformation(final
String groupProtocolC
streamOneProperties.put(StreamsConfig.STATE_DIR_CONFIG,
TestUtils.tempDirectory(appId).getPath() + "-ks1");
streamOneProperties.put(StreamsConfig.CLIENT_ID_CONFIG, appId +
"-ks1");
streamOneProperties.put(StreamsConfig.APPLICATION_SERVER_CONFIG,
"localhost:2020");
+ streamOneProperties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG,
3);
Review Comment:
Adding 3 threads to streamOne alone makes the whole test fail
deterministically (all 4 cases): the assertion at line 159 expects one thread
to hold 2 active tasks, but with 3 threads streamOne's 2 active tasks end up
spread one-per-thread, so `metadataForLocalThreads().iterator().next()` never
has 2. As Matthias suggested, use 2 threads on both instances and 4 input
partitions, and update the per-thread assertions to match.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -834,8 +834,21 @@ private static Map<StreamsRebalanceData.HostInfo,
StreamsRebalanceData.EndpointP
List<TopicPartition> activeTopicPartitions =
getTopicPartitionList(endpoint.activePartitions());
List<TopicPartition> standbyTopicPartitions =
getTopicPartitionList(endpoint.standbyPartitions());
StreamsGroupHeartbeatResponseData.Endpoint userEndpoint =
endpoint.userEndpoint();
- StreamsRebalanceData.EndpointPartitions endpointPartitions = new
StreamsRebalanceData.EndpointPartitions(activeTopicPartitions,
standbyTopicPartitions);
- partitionsByHost.put(new
StreamsRebalanceData.HostInfo(userEndpoint.host(), userEndpoint.port()),
endpointPartitions);
+ StreamsRebalanceData.HostInfo hostInfo = new
StreamsRebalanceData.HostInfo(userEndpoint.host(), userEndpoint.port());
+
+ partitionsByHost.merge(
Review Comment:
No unit test covers this merge branch.
`StreamsGroupHeartbeatRequestManagerTest`'s fixture has a single endpoint, so
the union path across duplicate host:port entries never runs. Add a case with
two entries sharing a host:port and assert the partitions are unioned.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -834,8 +834,21 @@ private static Map<StreamsRebalanceData.HostInfo,
StreamsRebalanceData.EndpointP
List<TopicPartition> activeTopicPartitions =
getTopicPartitionList(endpoint.activePartitions());
List<TopicPartition> standbyTopicPartitions =
getTopicPartitionList(endpoint.standbyPartitions());
StreamsGroupHeartbeatResponseData.Endpoint userEndpoint =
endpoint.userEndpoint();
- StreamsRebalanceData.EndpointPartitions endpointPartitions = new
StreamsRebalanceData.EndpointPartitions(activeTopicPartitions,
standbyTopicPartitions);
- partitionsByHost.put(new
StreamsRebalanceData.HostInfo(userEndpoint.host(), userEndpoint.port()),
endpointPartitions);
+ StreamsRebalanceData.HostInfo hostInfo = new
StreamsRebalanceData.HostInfo(userEndpoint.host(), userEndpoint.port());
+
Review Comment:
These blank lines (838, 845, 848) have trailing whitespace — please strip
them.
--
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]
