1

apache · Mar 12, 2024 · 387e23b · 387e23b
1 parent fe3435f
commit 387e23b
Showing 1 changed file with 10 additions and 5 deletions.
diff --git a/...r/src/main/java/org/apache/seatunnel/engine/server/rest/RestHttpPostCommandProcessor.java b/...r/src/main/java/org/apache/seatunnel/engine/server/rest/RestHttpPostCommandProcessor.java
@@ -97,10 +97,15 @@ public void handle(HttpPostCommand httpPostCommand) {
         this.textCommandService.sendResponse(httpPostCommand);
     }
 
-    private SeaTunnelServer getSeaTunnelServer() {
+    private SeaTunnelServer getSeaTunnelServer(boolean shouldBeMaster) {
         Map<String, Object> extensionServices =
                 this.textCommandService.getNode().getNodeExtension().createExtensionServices();
-        return (SeaTunnelServer) extensionServices.get(Constant.SEATUNNEL_SERVICE_NAME);
+        SeaTunnelServer seaTunnelServer =
+                (SeaTunnelServer) extensionServices.get(Constant.SEATUNNEL_SERVICE_NAME);
+        if (!seaTunnelServer.isMasterNode() && shouldBeMaster) {
+            return null;
+        }
+        return seaTunnelServer;
     }
 
     private void handleSubmitJob(HttpPostCommand httpPostCommand, String uri)
@@ -119,7 +124,7 @@ private void handleSubmitJob(HttpPostCommand httpPostCommand, String uri)
 
         boolean startWithSavePoint =
                 Boolean.parseBoolean(requestParams.get(RestConstant.IS_START_WITH_SAVE_POINT));
-        SeaTunnelServer seaTunnelServer = getSeaTunnelServer();
+        SeaTunnelServer seaTunnelServer = getSeaTunnelServer(false);
         RestJobExecutionEnvironment restJobExecutionEnvironment =
                 new RestJobExecutionEnvironment(
                         seaTunnelServer,
@@ -165,8 +170,8 @@ private void handleStopJob(HttpPostCommand httpPostCommand, String uri) {
                     Boolean.parseBoolean(map.get(RestConstant.IS_STOP_WITH_SAVE_POINT).toString());
         }
 
-        SeaTunnelServer seaTunnelServer = getSeaTunnelServer();
-        if (!seaTunnelServer.isMasterNode()) {
+        SeaTunnelServer seaTunnelServer = getSeaTunnelServer(true);
+        if (seaTunnelServer == null) {
             if (isStopWithSavePoint) {
                 NodeEngineUtil.sendOperationToMasterNode(
                                 getNode().nodeEngine, new SavePointJobOperation(jobId))