Merge pull request #343 from Aiven-Open/anatolii/flaky-test-fix

Flaky integration tests fix

.github/workflows/main_push_workflow.yml

@@ -34,4 +34,4 @@ jobs:
         run: ./gradlew build test
       - name: Build in Linux
         if: runner.os == 'Linux'
-        run: ./gradlew build check test integrationTest
+        run: ./gradlew build check test integrationTest -i

s3-source-connector/src/integration-test/java/io/aiven/kafka/connect/s3/source/IntegrationBase.java

@@ -28,6 +28,7 @@
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -116,9 +117,16 @@ static LocalStackContainer createS3Container() {
                 .withServices(LocalStackContainer.Service.S3);
     }
 
-    static int getRandomPort() throws IOException {
-        try (ServerSocket socket = new ServerSocket(0)) {
-            return socket.getLocalPort();
+    /**
+     * Finds 2 simultaneously free port for Kafka listeners
+     *
+     * @return list of 2 ports
+     * @throws IOException
+     *             when port allocation failure happens
+     */
+    static List<Integer> getKafkaListenerPorts() throws IOException {
+        try (ServerSocket socket = new ServerSocket(0); ServerSocket socket2 = new ServerSocket(0)) {
+            return Arrays.asList(socket.getLocalPort(), socket2.getLocalPort());
         } catch (IOException e) {
             throw new IOException("Failed to allocate port for test", e);
         }

s3-source-connector/src/integration-test/java/io/aiven/kafka/connect/s3/source/IntegrationTest.java

@@ -124,8 +124,9 @@ void setUp(final TestInfo testInfo) throws Exception {
         testBucketAccessor.createBucket();
 
         connectRunner = new ConnectRunner(OFFSET_FLUSH_INTERVAL_MS);
-        final int localListenerPort = IntegrationBase.getRandomPort();
-        final int containerListenerPort = IntegrationBase.getRandomPort();
+        final List<Integer> ports = IntegrationBase.getKafkaListenerPorts();
+        final int localListenerPort = ports.get(0);
+        final int containerListenerPort = ports.get(1);
         connectRunner.startConnectCluster(CONNECTOR_NAME, localListenerPort, containerListenerPort);
 
         adminClient = newAdminClient(connectRunner.getBootstrapServers());