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-12963-c875a87bd6fb238585dd6856c2c4233b6a8ece46 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 8b93585ef7c06a0af3db96e50451d193340d093c Author: WPNG <[email protected]> AuthorDate: Tue Oct 27 10:11:58 2026 +0000 [Fix][Core] Fix ServerExecuteCommand rejects explicit master_and_worker cluster role (#12963) Co-authored-by: wpngwang <[email protected]> Co-authored-by: zhiweiniu <[email protected]> --- .../seatunnel/command/ServerExecuteCommand.java | 38 ++++++++++++++------ .../command/ServerExecuteCommandTest.java | 40 ++++++++++++++++++++++ 2 files changed, 68 insertions(+), 10 deletions(-) diff --git a/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ServerExecuteCommand.java b/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ServerExecuteCommand.java index 0eceb20e56..b31095258f 100644 --- a/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ServerExecuteCommand.java +++ b/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ServerExecuteCommand.java @@ -62,16 +62,7 @@ public class ServerExecuteCommand implements Command<ServerCommandArgs> { String clusterRole = this.serverCommandArgs.getClusterRole(); if (StringUtils.isNotBlank(clusterRole)) { - if (EngineConfig.ClusterRole.MASTER.toString().equalsIgnoreCase(clusterRole)) { - seaTunnelConfig.getEngineConfig().setClusterRole(EngineConfig.ClusterRole.MASTER); - } else if (EngineConfig.ClusterRole.WORKER.toString().equalsIgnoreCase(clusterRole)) { - seaTunnelConfig.getEngineConfig().setClusterRole(EngineConfig.ClusterRole.WORKER); - - // in hazelcast lite node will not store IMap data. - seaTunnelConfig.getHazelcastConfig().setLiteMember(true); - } else { - throw new SeaTunnelEngineException("Not supported cluster role: " + clusterRole); - } + applyClusterRole(seaTunnelConfig, clusterRole); } else { seaTunnelConfig .getEngineConfig() @@ -82,6 +73,33 @@ public class ServerExecuteCommand implements Command<ServerCommandArgs> { seaTunnelConfig, Thread.currentThread().getName()); } + /** + * Applies the cluster role to the SeaTunnel config based on the provided role string. Supports + * master, worker and master_and_worker (case-insensitive). + * + * @param seaTunnelConfig the SeaTunnel config to configure + * @param clusterRole the cluster role string + */ + @VisibleForTesting + static void applyClusterRole(SeaTunnelConfig seaTunnelConfig, String clusterRole) { + if (EngineConfig.ClusterRole.MASTER.toString().equalsIgnoreCase(clusterRole)) { + seaTunnelConfig.getEngineConfig().setClusterRole(EngineConfig.ClusterRole.MASTER); + } else if (EngineConfig.ClusterRole.WORKER.toString().equalsIgnoreCase(clusterRole)) { + seaTunnelConfig.getEngineConfig().setClusterRole(EngineConfig.ClusterRole.WORKER); + + // in hazelcast lite node will not store IMap data. + seaTunnelConfig.getHazelcastConfig().setLiteMember(true); + } else if (EngineConfig.ClusterRole.MASTER_AND_WORKER + .toString() + .equalsIgnoreCase(clusterRole)) { + seaTunnelConfig + .getEngineConfig() + .setClusterRole(EngineConfig.ClusterRole.MASTER_AND_WORKER); + } else { + throw new SeaTunnelEngineException("Not supported cluster role: " + clusterRole); + } + } + private void checkEnvironment() { if (isUnsupportedJavaVersion()) { log.warn( diff --git a/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/command/ServerExecuteCommandTest.java b/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/command/ServerExecuteCommandTest.java index d32d942e14..65712735ad 100644 --- a/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/command/ServerExecuteCommandTest.java +++ b/seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/command/ServerExecuteCommandTest.java @@ -19,7 +19,9 @@ package org.apache.seatunnel.core.starter.seatunnel.command; import org.apache.seatunnel.core.starter.seatunnel.args.ServerCommandArgs; import org.apache.seatunnel.engine.common.config.ConfigProvider; +import org.apache.seatunnel.engine.common.config.EngineConfig; import org.apache.seatunnel.engine.common.config.SeaTunnelConfig; +import org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException; import org.apache.seatunnel.engine.server.SeaTunnelServerStarter; import org.junit.jupiter.api.Assertions; @@ -42,6 +44,44 @@ public class ServerExecuteCommandTest { Assertions.assertFalse(ServerExecuteCommand.isUnsupportedJavaVersion()); } + @Test + public void testApplyClusterRole() { + // master + SeaTunnelConfig masterConfig = new SeaTunnelConfig(); + ServerExecuteCommand.applyClusterRole(masterConfig, "master"); + Assertions.assertEquals( + EngineConfig.ClusterRole.MASTER, masterConfig.getEngineConfig().getClusterRole()); + Assertions.assertFalse(masterConfig.getHazelcastConfig().isLiteMember()); + + // worker + SeaTunnelConfig workerConfig = new SeaTunnelConfig(); + ServerExecuteCommand.applyClusterRole(workerConfig, "worker"); + Assertions.assertEquals( + EngineConfig.ClusterRole.WORKER, workerConfig.getEngineConfig().getClusterRole()); + Assertions.assertTrue(workerConfig.getHazelcastConfig().isLiteMember()); + + // master_and_worker + SeaTunnelConfig masterAndWorkerConfig = new SeaTunnelConfig(); + ServerExecuteCommand.applyClusterRole(masterAndWorkerConfig, "master_and_worker"); + Assertions.assertEquals( + EngineConfig.ClusterRole.MASTER_AND_WORKER, + masterAndWorkerConfig.getEngineConfig().getClusterRole()); + Assertions.assertFalse(masterAndWorkerConfig.getHazelcastConfig().isLiteMember()); + + // case-insensitive + SeaTunnelConfig caseInsensitiveConfig = new SeaTunnelConfig(); + ServerExecuteCommand.applyClusterRole(caseInsensitiveConfig, "Master_And_Worker"); + Assertions.assertEquals( + EngineConfig.ClusterRole.MASTER_AND_WORKER, + caseInsensitiveConfig.getEngineConfig().getClusterRole()); + + // invalid value + SeaTunnelConfig invalidConfig = new SeaTunnelConfig(); + Assertions.assertThrows( + SeaTunnelEngineException.class, + () -> ServerExecuteCommand.applyClusterRole(invalidConfig, "invalid_role")); + } + @Test public void testMemberList() throws InterruptedException { String clusterName = getClusterName("ServerExecuteCommandTest");
