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-11868-4f70bc55d1fe78b7cb2a32a7cddac95d4560bedc in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit fd2562251d29cbf99b64f17ff5f8f9da7c448a5c Author: JeremyXin <[email protected]> AuthorDate: Mon Oct 26 16:09:38 2026 +0000 [Fix][Zeta] Align Hazelcast join port try count with port range (#11868) Co-authored-by: David Zollo <[email protected]> --- .../src/test/resources/hazelcast.yaml | 2 +- .../src/main/resources/hazelcast.yaml | 4 +- .../src/test/resources/hazelcast.yaml | 4 +- .../engine/server/AbstractSeaTunnelServerTest.java | 30 ++++++- .../server/resourcemanager/WorkerTagTest.java | 2 +- .../engine/server/rest/OptionRulesApiTest.java | 97 +++++++++++++++++++++- .../RestApiSubmitJobConfigShadeDecryptTest.java | 1 + .../RestApiSubmitJobStartWithSavePointTest.java | 2 +- .../src/test/resources/hazelcast.yaml | 1 + 9 files changed, 130 insertions(+), 13 deletions(-) diff --git a/seatunnel-engine/seatunnel-engine-client/src/test/resources/hazelcast.yaml b/seatunnel-engine/seatunnel-engine-client/src/test/resources/hazelcast.yaml index 453538bcc4..6cc192a42f 100644 --- a/seatunnel-engine/seatunnel-engine-client/src/test/resources/hazelcast.yaml +++ b/seatunnel-engine/seatunnel-engine-client/src/test/resources/hazelcast.yaml @@ -38,6 +38,6 @@ hazelcast: properties: hazelcast.invocation.max.retry.count: 20 - hazelcast.tcp.join.port.try.count: 30 + hazelcast.tcp.join.port.try.count: 10 hazelcast.logging.type: log4j2 hazelcast.operation.generic.thread.count: 200 diff --git a/seatunnel-engine/seatunnel-engine-common/src/main/resources/hazelcast.yaml b/seatunnel-engine/seatunnel-engine-common/src/main/resources/hazelcast.yaml index 0b48069c3e..fd5a9cb1b5 100644 --- a/seatunnel-engine/seatunnel-engine-common/src/main/resources/hazelcast.yaml +++ b/seatunnel-engine/seatunnel-engine-common/src/main/resources/hazelcast.yaml @@ -36,6 +36,6 @@ hazelcast: port: 5801 properties: hazelcast.invocation.max.retry.count: 20 - hazelcast.tcp.join.port.try.count: 30 + hazelcast.tcp.join.port.try.count: 100 hazelcast.logging.type: log4j2 - hazelcast.operation.generic.thread.count: 50 \ No newline at end of file + hazelcast.operation.generic.thread.count: 50 diff --git a/seatunnel-engine/seatunnel-engine-common/src/test/resources/hazelcast.yaml b/seatunnel-engine/seatunnel-engine-common/src/test/resources/hazelcast.yaml index 21f4d544d8..80c394b828 100644 --- a/seatunnel-engine/seatunnel-engine-common/src/test/resources/hazelcast.yaml +++ b/seatunnel-engine/seatunnel-engine-common/src/test/resources/hazelcast.yaml @@ -27,6 +27,8 @@ hazelcast: auto-increment: true port-count: 100 port: 5801 + properties: + hazelcast.tcp.join.port.try.count: 100 map: map-name-template: map-store: @@ -34,4 +36,4 @@ hazelcast: initial-mode: EAGER class-name: org.apache.seatunnel.engine.server.persistence.FileMapStore properties: - path: /tmp/file-store-map \ No newline at end of file + path: /tmp/file-store-map diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/AbstractSeaTunnelServerTest.java b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/AbstractSeaTunnelServerTest.java index d95e7278ac..521e733253 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/AbstractSeaTunnelServerTest.java +++ b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/AbstractSeaTunnelServerTest.java @@ -48,6 +48,9 @@ import java.util.Collections; @TestInstance(TestInstance.Lifecycle.PER_CLASS) public abstract class AbstractSeaTunnelServerTest<T extends AbstractSeaTunnelServerTest> { + private static final int HAZELCAST_PORT_COUNT = 100; + private static final int HAZELCAST_TCP_JOIN_PORT_TRY_COUNT = HAZELCAST_PORT_COUNT; + protected SeaTunnelServer server; protected NodeEngine nodeEngine; @@ -56,7 +59,7 @@ public abstract class AbstractSeaTunnelServerTest<T extends AbstractSeaTunnelSer protected static ILogger LOGGER; - private final int hazelcastPort = TestUtils.getAvailablePort(100); + private final int hazelcastPort = TestUtils.getAvailablePort(HAZELCAST_PORT_COUNT); @BeforeAll public void before() { @@ -74,6 +77,7 @@ public abstract class AbstractSeaTunnelServerTest<T extends AbstractSeaTunnelSer } protected String getHazelcastConfig() { + assertHazelcastJoinPortTryCount(HAZELCAST_PORT_COUNT, HAZELCAST_TCP_JOIN_PORT_TRY_COUNT); // Use the IPv4 loopback address to avoid dual-stack localhost resolution on Windows CI. return "hazelcast:\n" + " cluster-name: seatunnel\n" @@ -90,20 +94,40 @@ public abstract class AbstractSeaTunnelServerTest<T extends AbstractSeaTunnelSer + " - 127.0.0.1\n" + " port:\n" + " auto-increment: true\n" - + " port-count: 100\n" + + " port-count: " + + HAZELCAST_PORT_COUNT + + "\n" + " port: " + getHazelcastPort() + "\n" + "\n" + " properties:\n" + " hazelcast.invocation.max.retry.count: 200\n" - + " hazelcast.tcp.join.port.try.count: 30\n" + + " hazelcast.tcp.join.port.try.count: " + + HAZELCAST_TCP_JOIN_PORT_TRY_COUNT + + "\n" + " hazelcast.invocation.retry.pause.millis: 2000\n" + " hazelcast.slow.operation.detector.stacktrace.logging.enabled: true\n" + " hazelcast.logging.type: log4j2\n" + " hazelcast.operation.generic.thread.count: 200\n"; } + /** + * Ensures TCP discovery can find a test member on every permitted auto-increment port. + * + * @param portCount number of candidate member ports + * @param joinPortTryCount number of ports probed by TCP discovery + * @throws IllegalArgumentException when discovery cannot cover the member port range + */ + public static void assertHazelcastJoinPortTryCount(int portCount, int joinPortTryCount) { + if (joinPortTryCount < portCount) { + throw new IllegalArgumentException( + String.format( + "hazelcast.tcp.join.port.try.count (%d) must be >= port-count (%d)", + joinPortTryCount, portCount)); + } + } + protected int getHazelcastPort() { return hazelcastPort; } diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/resourcemanager/WorkerTagTest.java b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/resourcemanager/WorkerTagTest.java index 6ba193bddc..022b2625cd 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/resourcemanager/WorkerTagTest.java +++ b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/resourcemanager/WorkerTagTest.java @@ -71,7 +71,7 @@ public class WorkerTagTest extends AbstractSeaTunnelServerTest<WorkerTagTest> { + "\n" + " properties:\n" + " hazelcast.invocation.max.retry.count: 200\n" - + " hazelcast.tcp.join.port.try.count: 30\n" + + " hazelcast.tcp.join.port.try.count: 100\n" + " hazelcast.invocation.retry.pause.millis: 2000\n" + " hazelcast.slow.operation.detector.stacktrace.logging.enabled: true\n" + " hazelcast.logging.type: log4j2\n" diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/OptionRulesApiTest.java b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/OptionRulesApiTest.java index 5737f4722e..d5517b3137 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/OptionRulesApiTest.java +++ b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/OptionRulesApiTest.java @@ -25,6 +25,7 @@ import org.apache.seatunnel.engine.server.SeaTunnelServerStarter; import org.apache.seatunnel.engine.server.TestUtils; import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.core.LoggerContext; import org.junit.jupiter.api.AfterAll; @@ -34,6 +35,11 @@ import org.junit.jupiter.api.Test; import com.hazelcast.config.Config; import com.hazelcast.instance.impl.HazelcastInstanceImpl; +import java.io.IOException; +import java.net.DatagramSocket; +import java.net.ServerSocket; +import java.util.Map; + import static io.restassured.RestAssured.given; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.equalTo; @@ -41,6 +47,7 @@ import static org.hamcrest.Matchers.hasItem; class OptionRulesApiTest { + private static final Logger LOG = LogManager.getLogger(OptionRulesApiTest.class); private static final int HTTP_PORT = 18082; private static final int HAZELCAST_PORT = TestUtils.getAvailablePort(100); @@ -61,8 +68,10 @@ class OptionRulesApiTest { originalEnableDynamicPort = httpConfig.isEnableDynamicPort(); originalExecutionMode = seaTunnelConfig.getEngineConfig().getMode(); - Config hazelcastConfig = Config.loadFromString(getHazelcastConfig()); - hazelcastConfig.setClusterName(TestUtils.getClusterName("OptionRulesApiTest")); + String hazelcastYaml = getHazelcastConfig(); + Config hazelcastConfig = Config.loadFromString(hazelcastYaml); + String clusterName = TestUtils.getClusterName("OptionRulesApiTest"); + hazelcastConfig.setClusterName(clusterName); seaTunnelConfig.setHazelcastConfig(hazelcastConfig); seaTunnelConfig.getEngineConfig().setMode(ExecutionMode.LOCAL); @@ -70,7 +79,31 @@ class OptionRulesApiTest { httpConfig.setPort(HTTP_PORT); httpConfig.setEnableDynamicPort(false); - instance = SeaTunnelServerStarter.createHazelcastInstance(seaTunnelConfig); + logStartupContext(clusterName, hazelcastYaml, httpConfig); + + try { + instance = SeaTunnelServerStarter.createHazelcastInstance(seaTunnelConfig); + LOG.info( + "OptionRulesApiTest started successfully. clusterName={}, localMemberAddress={}," + + " actualHazelcastPort={}, clusterSize={}", + clusterName, + instance.getCluster().getLocalMember().getAddress(), + instance.getCluster().getLocalMember().getAddress().getPort(), + instance.getCluster().getMembers().size()); + } catch (Throwable t) { + LOG.error( + "OptionRulesApiTest failed to start. clusterName={}, hazelcastBasePort={}, httpPort={}," + + " hazelcastPortInUse={}, httpPortInUse={}", + clusterName, + HAZELCAST_PORT, + HTTP_PORT, + isPortInUse(HAZELCAST_PORT), + isPortInUse(HTTP_PORT), + t); + logCauseChain(t); + logThreadDump(); + throw t; + } } @Test @@ -171,10 +204,66 @@ class OptionRulesApiTest { + "\n" + " properties:\n" + " hazelcast.invocation.max.retry.count: 200\n" - + " hazelcast.tcp.join.port.try.count: 30\n" + + " hazelcast.tcp.join.port.try.count: 100\n" + " hazelcast.invocation.retry.pause.millis: 2000\n" + " hazelcast.slow.operation.detector.stacktrace.logging.enabled: true\n" + " hazelcast.logging.type: log4j2\n" + " hazelcast.operation.generic.thread.count: 200\n"; } + + private static void logStartupContext( + String clusterName, String hazelcastYaml, HttpConfig httpConfig) { + LOG.info( + "OptionRulesApiTest startup context: clusterName={}, hazelcastBasePort={}," + + " hazelcastPortCount=100, hazelcastJoinPortTryCount=100, httpPort={}," + + " httpDynamicPortEnabled={}, hazelcastBasePortInUse={}, httpPortInUse={}", + clusterName, + HAZELCAST_PORT, + HTTP_PORT, + httpConfig.isEnableDynamicPort(), + isPortInUse(HAZELCAST_PORT), + isPortInUse(HTTP_PORT)); + LOG.info("OptionRulesApiTest Hazelcast YAML:\n{}", hazelcastYaml); + } + + private static void logCauseChain(Throwable throwable) { + Throwable current = throwable; + int depth = 0; + while (current != null) { + LOG.error( + "OptionRulesApiTest startup failure cause[{}]: {}: {}", + depth, + current.getClass().getName(), + current.getMessage()); + current = current.getCause(); + depth++; + } + } + + private static void logThreadDump() { + LOG.error("OptionRulesApiTest thread dump start"); + for (Map.Entry<Thread, StackTraceElement[]> entry : Thread.getAllStackTraces().entrySet()) { + Thread thread = entry.getKey(); + StackTraceElement[] stackTrace = entry.getValue(); + LOG.error( + "Thread dump: name='{}', state={}, daemon={}, interrupted={}", + thread.getName(), + thread.getState(), + thread.isDaemon(), + thread.isInterrupted()); + for (StackTraceElement stackTraceElement : stackTrace) { + LOG.error(" at {}", stackTraceElement); + } + } + LOG.error("OptionRulesApiTest thread dump end"); + } + + private static boolean isPortInUse(int port) { + try (ServerSocket serverSocket = new ServerSocket(port); + DatagramSocket datagramSocket = new DatagramSocket(port)) { + return false; + } catch (IOException e) { + return true; + } + } } diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/RestApiSubmitJobConfigShadeDecryptTest.java b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/RestApiSubmitJobConfigShadeDecryptTest.java index 1ac746eb0b..9d6d33eef3 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/RestApiSubmitJobConfigShadeDecryptTest.java +++ b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/RestApiSubmitJobConfigShadeDecryptTest.java @@ -388,6 +388,7 @@ public class RestApiSubmitJobConfigShadeDecryptTest { + "\n" + " properties:\n" + " hazelcast.invocation.max.retry.count: 200\n" + + " hazelcast.tcp.join.port.try.count: 100\n" + " hazelcast.invocation.retry.pause.millis: 2000\n" + " hazelcast.slow.operation.detector.stacktrace.logging.enabled: true\n" + " hazelcast.logging.type: log4j2\n" diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/RestApiSubmitJobStartWithSavePointTest.java b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/RestApiSubmitJobStartWithSavePointTest.java index e6ce345c08..fae9809a02 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/RestApiSubmitJobStartWithSavePointTest.java +++ b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/RestApiSubmitJobStartWithSavePointTest.java @@ -531,7 +531,7 @@ public class RestApiSubmitJobStartWithSavePointTest { + "\n" + " properties:\n" + " hazelcast.invocation.max.retry.count: 200\n" - + " hazelcast.tcp.join.port.try.count: 30\n" + + " hazelcast.tcp.join.port.try.count: 100\n" + " hazelcast.invocation.retry.pause.millis: 2000\n" + " hazelcast.slow.operation.detector.stacktrace.logging.enabled: true\n" + " hazelcast.logging.type: log4j2\n" diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/resources/hazelcast.yaml b/seatunnel-engine/seatunnel-engine-server/src/test/resources/hazelcast.yaml index af81ee98a8..7a63d86bed 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/test/resources/hazelcast.yaml +++ b/seatunnel-engine/seatunnel-engine-server/src/test/resources/hazelcast.yaml @@ -36,5 +36,6 @@ hazelcast: properties: path: /tmp/file-store-map properties: + hazelcast.tcp.join.port.try.count: 100 hazelcast.slow.operation.detector.stacktrace.logging.enabled: true hazelcast.slow.operation.detector.logging.enabled: true
