KAFKA-19239 Rewrite IntegrationTestUtils by java #19776

jim0987795064 · 2025-05-20T17:57:41Z

This PR rewrites the IntegrationTestUtils.java from Scala to Java.

Changes:

Converted all the existing Scala code in IntegrationTestUtils.scala
into Java in IntegrationTestUtils.java.
Preserved the original logic and functionality to ensure backward
compatibility.
Updated relevant imports and dependencies accordingly.

Motivation:

The rewrite aims to standardize the codebase in Java, which aligns
better with the rest of the project and facilitates easier maintenance
by the Java-centric team.

TaiJuWu · 2025-05-20T23:21:33Z

core/src/test/java/integration/kafka/server/IntegrationTestUtils.java

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.server;


should this package be put into server-common module?

chia7712 · 2025-05-22T20:35:35Z

@jim0987795064 please fix the conflicts

….java

jim0987795064 · 2025-05-23T06:07:34Z

@jim0987795064 please fix the conflicts
Hello @chia7712
Thank you, I have addressed this comment.

m1a2st

Thanks @jim0987795064 for this patch, left some comments

m1a2st · 2025-05-24T08:10:47Z

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

-                        ApiError apiError = new ApiError(error, response.data().errorMessage());
-                        handleFailure(apiError.exception());
+                        handleFailure(error.exception(response.data().errorMessage()));


Just wondering, how is this change related to the main scope of this PR?

Hi @m1a2st,

Thanks for pointing this out! You're absolutely right — this change is unrelated to the main scope of this PR. I'll move it to a separate pull request to keep things clean and focused.

Thanks for your feedback!

m1a2st · 2025-05-24T08:14:07Z

core/src/test/java/integration/kafka/server/IntegrationTestUtils.java

+        Socket socket = connect(destination, listenerName);
+        try {
+            return sendAndReceive(request, socket);
+        } finally {
+            socket.close();
+        }


Suggested change

Socket socket = connect(destination, listenerName);

try {

return sendAndReceive(request, socket);

} finally {

socket.close();

}

try (Socket socket = connect(destination, listenerName)) {

return sendAndReceive(request, socket);

}

m1a2st · 2025-05-24T08:18:26Z

core/src/test/java/integration/kafka/server/IntegrationTestUtils.java

+            Integer correlationId
+    ) throws IOException {
+        send(request, socket, clientId, correlationId);
+        return (T) receive(socket, request.apiKey(), request.version());


Suggested change

return (T) receive(socket, request.apiKey(), request.version());

return receive(socket, request.apiKey(), request.version());

m1a2st · 2025-05-24T08:19:19Z

core/src/test/java/integration/kafka/server/IntegrationTestUtils.java

+        sendWithHeader(request, header, socket);
+    }
+
+    public static <T extends AbstractResponse> T receive(Socket socket, ApiKeys apiKey, short version) throws IOException, ClassCastException {


Please add the @SuppressWarnings("unchecked")

m1a2st · 2025-05-24T08:21:23Z

core/src/test/java/integration/kafka/server/IntegrationTestUtils.java

+    public static void send(AbstractRequest request, Socket socket) throws IOException {
+        RequestHeader header = nextRequestHeader(request.apiKey(), request.version(), "client-id", 0);
+        sendWithHeader(request, header, socket);
+    }


This method is unused in project, I think we should remove it.

m1a2st · 2025-05-24T08:22:14Z

core/src/test/java/integration/kafka/server/IntegrationTestUtils.java

+    }
+
+    public static RequestHeader nextRequestHeader(ApiKeys apiKey, short apiVersion) {
+


Extra blank line.

chia7712 · 2025-05-29T11:18:25Z

server-common/src/test/java/integration/kafka/server/IntegrationTestUtils.java

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server;


please move this file to the kafka/server-common/src/test/java/org/apache/kafka/server

chia7712 · 2025-05-29T11:18:57Z

server-common/src/test/java/integration/kafka/server/IntegrationTestUtils.java

+            AbstractRequest request,
+            int port
+    ) throws IOException {
+        Socket socket = connect(port);


try (Socket socket = connect(port)) { return sendAndReceive(request, socket); }

chia7712 · 2025-05-29T11:22:18Z

core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala

  }

  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 socketServer = cluster.brokerSocketServers().asScala.head


Could you please add following helper to ClusterInstance?

default Collection<Integer> boundPorts() { return brokers().values().stream() .map(KafkaBroker::socketServer) .map(s -> s.boundPort(clientListener())) .collect(Collectors.toList()); }

this helper can streamline this PR I believe

Hello, @chia7712 I've addressed these 3 issues mentioned above.
Could you please review these?

with that helper, we can use val socket = IntegrationTestUtils.connect(cluster.boundPorts().asScala.head) to streamline code, right?

…of IntegrationTestUtils, and helper to class ClusterInstance

…ationTestUtils

chia7712 · 2025-05-30T18:05:24Z

core/src/test/scala/integration/kafka/coordinator/transaction/ProducerIntegrationTest.scala

@@ -205,11 +205,12 @@ class ProducerIntegrationTest {
        .setProducerId(RecordBatch.NO_PRODUCER_ID)
        .setTransactionalId(null)
        .setTransactionTimeoutMs(10)
-      val request = new InitProducerIdRequest.Builder(data).build()
+    val request = new InitProducerIdRequest.Builder(data).build()


val request = new InitProducerIdRequest.Builder(data).build() val port = broker.boundPort(listener) response = IntegrationTestUtils.connectAndReceive[InitProducerIdResponse](request, port) shouldRetry = response.data.errorCode == Errors.COORDINATOR_LOAD_IN_PROGRESS.code

chia7712 · 2025-05-30T18:07:28Z

core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala

-    )
+                                                    listenerName: ListenerName): DescribeClusterResponse = {
+
+      val port = destination.boundPort(listenerName)


connectAndReceive[DescribeClusterResponse](new DescribeClusterRequest.Builder(new DescribeClusterRequestData()).build(), destination.boundPort(listenerName))

chia7712 · 2025-05-30T18:07:53Z

core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala

@@ -41,12 +42,18 @@ abstract class AbstractApiVersionsRequestTest(cluster: ClusterInstance) {
    } else {
      cluster.brokerSocketServers().asScala.head
    }
-    IntegrationTestUtils.connectAndReceive[ApiVersionsResponse](request, socket, listenerName)
+
+    val port = socket.boundPort(listenerName)


IntegrationTestUtils.connectAndReceive[ApiVersionsResponse](request, socket.boundPort(listenerName))

chia7712 · 2025-05-30T18:09:12Z

core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala

  }

  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 socketServer = cluster.brokerSocketServers().asScala.head


with that helper, we can use val socket = IntegrationTestUtils.connect(cluster.boundPorts().asScala.head) to streamline code, right?

chia7712 · 2025-05-30T18:10:11Z

core/src/test/scala/unit/kafka/server/AllocateProducerIdsRequestTest.scala

+
+    val listenerName = cluster.controllerListenerName
+    val port = controllerSocketServer.boundPort(listenerName)
+
    IntegrationTestUtils.connectAndReceive[AllocateProducerIdsResponse](


IntegrationTestUtils.connectAndReceive[AllocateProducerIdsResponse]( request, controllerSocketServer.boundPort(cluster.controllerListenerName) )

github-actions bot added triage PRs from the community core Kafka Broker clients labels May 20, 2025

TaiJuWu reviewed May 20, 2025

View reviewed changes

github-actions bot removed the triage PRs from the community label May 21, 2025

jim0987795064 changed the title ~~(WIP)KAFKA-19239 Rewrite IntegrationTestUtils by java~~ KAFKA-19239 Rewrite IntegrationTestUtils by java May 21, 2025

m1a2st added the ci-approved label May 21, 2025

jim0987795064 added 2 commits May 22, 2025 23:59

test

7fdbabd

refactor: Refactor IntegrationTestUtils.scala to IntegrationTestUtils…

c2318db

….java

jim0987795064 force-pushed the KAFKA-19239 branch from a2d6561 to c2318db Compare May 22, 2025 22:47

m1a2st reviewed May 24, 2025

View reviewed changes

Revert accidental changes to KafkaAdminClient.java

196881e

github-actions bot added the tests Test fixes (including flaky tests) label May 27, 2025

refactor: Move IntegrationTestUtils.java to server-common module

cbb7188

chia7712 reviewed May 29, 2025

View reviewed changes

jim0987795064 added 2 commits May 29, 2025 14:32

refactor: Move IntegrationTestUtils.java to new path, simplify logic …

e66b48d

…of IntegrationTestUtils, and helper to class ClusterInstance

refactor: Simplify the logic of connectAndReceive in the class Integr…

bb4462e

…ationTestUtils

chia7712 reviewed May 30, 2025

View reviewed changes

	return (T) receive(socket, request.apiKey(), request.version());
	return receive(socket, request.apiKey(), request.version());

		}

		public static RequestHeader nextRequestHeader(ApiKeys apiKey, short apiVersion) {

KAFKA-19239 Rewrite IntegrationTestUtils by java #19776

Are you sure you want to change the base?

KAFKA-19239 Rewrite IntegrationTestUtils by java #19776

Uh oh!

Conversation

jim0987795064 commented May 20, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Changes:

Uh oh!

TaiJuWu May 20, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chia7712 commented May 22, 2025

Uh oh!

jim0987795064 commented May 23, 2025

Uh oh!

m1a2st left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jim0987795064 May 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

jim0987795064 commented May 20, 2025 •

edited by github-actions bot

Loading

TaiJuWu May 20, 2025 •

edited

Loading

jim0987795064 May 29, 2025 •

edited

Loading