chia7712 commented on code in PR #19776:
URL: https://github.com/apache/kafka/pull/19776#discussion_r2153459442
##########
core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala:
##########
@@ -556,9 +557,11 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
private def sendDescribeClientQuotasRequest(filter: ClientQuotaFilter):
DescribeClientQuotasResponse = {
val request = new DescribeClientQuotasRequest.Builder(filter).build()
-
IntegrationTestUtils.connectAndReceive[DescribeClientQuotasResponse](request,
- destination = cluster.anyBrokerSocketServer(),
- listenerName = cluster.clientListener())
+ val destination = cluster.anyBrokerSocketServer()
Review Comment:
Could you please consider using `cluster.boundPorts()` to streamline it?
##########
core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala:
##########
@@ -41,12 +42,12 @@ abstract class AbstractApiVersionsRequestTest(cluster:
ClusterInstance) {
} else {
cluster.brokerSocketServers().asScala.head
}
- IntegrationTestUtils.connectAndReceive[ApiVersionsResponse](request,
socket, listenerName)
+ IntegrationTestUtils.connectAndReceive[ApiVersionsResponse](request,
socket.boundPort(listenerName))
}
def sendUnsupportedApiVersionRequest(request: ApiVersionsRequest):
ApiVersionsResponse = {
val overrideHeader =
IntegrationTestUtils.nextRequestHeader(ApiKeys.API_VERSIONS, Short.MaxValue)
- val socket =
IntegrationTestUtils.connect(cluster.brokerSocketServers().asScala.head,
cluster.clientListener())
+ val socket =
IntegrationTestUtils.connect(cluster.boundPorts().asScala.head)
Review Comment:
If `boundPorts` returns `List` instead of `Collection`, the scala conversion
could be removed.
##########
test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java:
##########
@@ -406,4 +406,11 @@ default int getLeaderBrokerId(TopicPartition
topicPartition) throws ExecutionExc
.orElseThrow(() -> new RuntimeException("Leader not found
for tp " + topicPartition));
}
}
+
+ default Collection<Integer> boundPorts() {
+ return brokers().values().stream()
+ .map(KafkaBroker::socketServer)
+ .map(s -> s.boundPort(clientListener()))
+ .collect(Collectors.toList());
Review Comment:
`toList`
##########
test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java:
##########
@@ -406,4 +406,11 @@ default int getLeaderBrokerId(TopicPartition
topicPartition) throws ExecutionExc
.orElseThrow(() -> new RuntimeException("Leader not found
for tp " + topicPartition));
}
}
+
+ default Collection<Integer> boundPorts() {
Review Comment:
`Collection` -> `List`
`List` has many helpful functions
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
