This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12778-fd2562251d29cbf99b64f17ff5f8f9da7c448a5c in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit cb3351ed124f68eeac6304ac4c1d328edafbed8f Author: JeremyXin <[email protected]> AuthorDate: Mon Oct 26 16:30:05 2026 +0000 [Fix][CI] Isolate Hazelcast test port ranges in additional E2E tests (#12778) Co-authored-by: JeremyXin <--golbal> --- .../seatunnel/engine/e2e/RecordSerializerIT.java | 17 +++++--- .../org/apache/seatunnel/engine/e2e/TestUtils.java | 44 +++++++++++++++++++ .../apache/seatunnel/engine/e2e/TestUtilsTest.java | 49 ++++++++++++++++++++++ .../resourceIsolation/WorkerTagClusterTest.java | 15 ++++--- 4 files changed, 113 insertions(+), 12 deletions(-) diff --git a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/RecordSerializerIT.java b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/RecordSerializerIT.java index cbad1083f2..4e3abd6ca9 100644 --- a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/RecordSerializerIT.java +++ b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/RecordSerializerIT.java @@ -56,8 +56,9 @@ public class RecordSerializerIT { @BeforeAll static void setUp() { String clusterName = TestUtils.getClusterName("RecordSerializerIT_hzSerializationTest"); - instance1 = createHazelcastInstance(clusterName); - instance2 = createHazelcastInstance(clusterName); + int hazelcastPort = TestUtils.getAvailablePort(100); + instance1 = createHazelcastInstance(clusterName, hazelcastPort); + instance2 = createHazelcastInstance(clusterName, hazelcastPort); await().atMost(30, TimeUnit.SECONDS) .until(() -> instance1.getCluster().getMembers().size() == 2); } @@ -314,15 +315,17 @@ public class RecordSerializerIT { writerMap.remove(key); } - private static HazelcastInstanceImpl createHazelcastInstance(String clusterName) { + private static HazelcastInstanceImpl createHazelcastInstance( + String clusterName, int hazelcastPort) { SeaTunnelConfig seaTunnelConfig = ConfigProvider.locateAndGetSeaTunnelConfig(); seaTunnelConfig.getEngineConfig().getHttpConfig().setEnabled(false); - Config hazelcastConfig = Config.loadFromString(buildHazelcastConfig(clusterName)); + Config hazelcastConfig = + Config.loadFromString(buildHazelcastConfig(clusterName, hazelcastPort)); seaTunnelConfig.setHazelcastConfig(hazelcastConfig); return SeaTunnelServerStarter.createHazelcastInstance(seaTunnelConfig); } - private static String buildHazelcastConfig(String clusterName) { + private static String buildHazelcastConfig(String clusterName, int hazelcastPort) { return "hazelcast:\n" + " cluster-name: " + clusterName @@ -336,6 +339,8 @@ public class RecordSerializerIT { + " port:\n" + " auto-increment: true\n" + " port-count: 100\n" - + " port: 5901\n"; + + " port: " + + hazelcastPort + + "\n"; } } diff --git a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/TestUtils.java b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/TestUtils.java index d88c7c5ee8..f3aad036c5 100644 --- a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/TestUtils.java +++ b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/TestUtils.java @@ -25,11 +25,55 @@ import lombok.extern.slf4j.Slf4j; import java.io.File; import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.ServerSocket; import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.List; import java.util.Map; +import java.util.concurrent.ThreadLocalRandom; @Slf4j public class TestUtils { + private static final int MIN_TEST_PORT = 10000; + private static final int MAX_TEST_PORT = 30000; + private static final int MAX_PORT_ALLOCATION_ATTEMPTS = 1000; + private static final List<int[]> ALLOCATED_PORT_RANGES = new ArrayList<>(); + + public static synchronized int getAvailablePort(int portCount) { + if (portCount <= 0 || portCount > MAX_TEST_PORT - MIN_TEST_PORT + 1) { + throw new IllegalArgumentException("Invalid port count: " + portCount); + } + for (int i = 0; i < MAX_PORT_ALLOCATION_ATTEMPTS; i++) { + int port = + ThreadLocalRandom.current() + .nextInt(MIN_TEST_PORT, MAX_TEST_PORT - portCount + 2); + if (isAvailablePortRange(port, portCount)) { + ALLOCATED_PORT_RANGES.add(new int[] {port, port + portCount - 1}); + return port; + } + } + throw new IllegalStateException("Failed to allocate an available port range"); + } + + private static boolean isAvailablePortRange(int port, int portCount) { + int rangeEnd = port + portCount - 1; + for (int[] allocatedPortRange : ALLOCATED_PORT_RANGES) { + if (port <= allocatedPortRange[1] && rangeEnd >= allocatedPortRange[0]) { + return false; + } + } + for (int currentPort = port; currentPort <= rangeEnd; currentPort++) { + try (ServerSocket socket = new ServerSocket()) { + socket.setReuseAddress(false); + socket.bind(new InetSocketAddress(currentPort)); + } catch (IOException e) { + return false; + } + } + return true; + } + public static String getResource(String confFile) { return System.getProperty("user.dir") + File.separator diff --git a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/TestUtilsTest.java b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/TestUtilsTest.java new file mode 100644 index 0000000000..d1d1b0b7be --- /dev/null +++ b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/TestUtilsTest.java @@ -0,0 +1,49 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.engine.e2e; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.net.ServerSocket; + +class TestUtilsTest { + + @Test + void allocatesBindablePort() throws Exception { + int port = TestUtils.getAvailablePort(1); + + try (ServerSocket ignored = new ServerSocket(port)) { + Assertions.assertEquals(port, ignored.getLocalPort()); + } + } + + @Test + void allocatesNonOverlappingPortRanges() { + int firstPort = TestUtils.getAvailablePort(10); + int secondPort = TestUtils.getAvailablePort(10); + + Assertions.assertTrue(firstPort + 9 < secondPort || secondPort + 9 < firstPort); + } + + @Test + void rejectsInvalidPortRangeSize() { + Assertions.assertThrows( + IllegalArgumentException.class, () -> TestUtils.getAvailablePort(0)); + } +} diff --git a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/resourceIsolation/WorkerTagClusterTest.java b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/resourceIsolation/WorkerTagClusterTest.java index 58ada293eb..6a2e692ac0 100644 --- a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/resourceIsolation/WorkerTagClusterTest.java +++ b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/resourceIsolation/WorkerTagClusterTest.java @@ -48,8 +48,9 @@ public class WorkerTagClusterTest { @BeforeEach public void before() { - SeaTunnelConfig masterNode1Config = getSeaTunnelConfig(testClusterName); - SeaTunnelConfig workerNode1Config = getSeaTunnelConfig(testClusterName); + int hazelcastPort = TestUtils.getAvailablePort(100); + SeaTunnelConfig masterNode1Config = getSeaTunnelConfig(testClusterName, hazelcastPort); + SeaTunnelConfig workerNode1Config = getSeaTunnelConfig(testClusterName, hazelcastPort); masterNode1 = SeaTunnelServerStarter.createMasterHazelcastInstance(masterNode1Config); workerNode1 = SeaTunnelServerStarter.createWorkerHazelcastInstance(workerNode1Config); } @@ -114,8 +115,8 @@ public class WorkerTagClusterTest { }); } - private static SeaTunnelConfig getSeaTunnelConfig(String testClusterName) { - Config hazelcastConfig = Config.loadFromString(getHazelcastConfig()); + private static SeaTunnelConfig getSeaTunnelConfig(String testClusterName, int hazelcastPort) { + Config hazelcastConfig = Config.loadFromString(getHazelcastConfig(hazelcastPort)); hazelcastConfig.setClusterName(TestUtils.getClusterName(testClusterName)); SeaTunnelConfig seaTunnelConfig = ConfigProvider.locateAndGetSeaTunnelConfig(); seaTunnelConfig.setHazelcastConfig(hazelcastConfig); @@ -123,7 +124,7 @@ public class WorkerTagClusterTest { return seaTunnelConfig; } - protected static String getHazelcastConfig() { + protected static String getHazelcastConfig(int hazelcastPort) { return "hazelcast:\n" + " cluster-name: seatunnel\n" + " network:\n" @@ -140,7 +141,9 @@ public class WorkerTagClusterTest { + " port:\n" + " auto-increment: true\n" + " port-count: 100\n" - + " port: 5801\n" + + " port: " + + hazelcastPort + + "\n" + "\n" + " properties:\n" + " hazelcast.invocation.max.retry.count: 200\n"
