This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 8b93585ef7 [Fix][Core] Fix ServerExecuteCommand rejects explicit
master_and_worker cluster role (#12963)
8b93585ef7 is described below
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");
