[Improve] Change class name for RestJobExecutionEnvironment impleme…

…nt (#5671)
apache · Oct 25, 2023 · 83082b2 · 83082b2
1 parent 75b814b
commit 83082b2
Show file tree

Hide file tree

Showing 12 changed files with 52 additions and 52 deletions.
diff --git a/...c/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ClientExecuteCommand.java b/...c/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ClientExecuteCommand.java
@@ -25,8 +25,8 @@
 import org.apache.seatunnel.core.starter.seatunnel.args.ClientCommandArgs;
 import org.apache.seatunnel.core.starter.utils.FileUtils;
 import org.apache.seatunnel.engine.client.SeaTunnelClient;
+import org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment;
 import org.apache.seatunnel.engine.client.job.ClientJobProxy;
-import org.apache.seatunnel.engine.client.job.JobExecutionEnvironment;
 import org.apache.seatunnel.engine.client.job.JobMetricsRunner;
 import org.apache.seatunnel.engine.common.Constant;
 import org.apache.seatunnel.engine.common.config.ConfigProvider;
@@ -123,7 +123,7 @@ public void execute() throws CommandExecuteException {
                 Path configFile = FileUtils.getConfigPath(clientCommandArgs);
                 checkConfigExist(configFile);
                 JobConfig jobConfig = new JobConfig();
-                JobExecutionEnvironment jobExecutionEnv;
+                ClientJobExecutionEnvironment jobExecutionEnv;
                 jobConfig.setName(clientCommandArgs.getJobName());
                 if (null != clientCommandArgs.getRestoreJobId()) {
                     jobExecutionEnv =

diff --git a/...unnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/ClusterFaultToleranceIT.java b/...unnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/ClusterFaultToleranceIT.java
@@ -23,8 +23,8 @@
 import org.apache.seatunnel.common.utils.ExceptionUtils;
 import org.apache.seatunnel.common.utils.FileUtils;
 import org.apache.seatunnel.engine.client.SeaTunnelClient;
+import org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment;
 import org.apache.seatunnel.engine.client.job.ClientJobProxy;
-import org.apache.seatunnel.engine.client.job.JobExecutionEnvironment;
 import org.apache.seatunnel.engine.common.config.ConfigProvider;
 import org.apache.seatunnel.engine.common.config.JobConfig;
 import org.apache.seatunnel.engine.common.config.SeaTunnelConfig;
@@ -108,7 +108,7 @@ public void testBatchJobRunOkIn2Node() throws ExecutionException, InterruptedExc
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
             clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
             engineClient = new SeaTunnelClient(clientConfig);
-            JobExecutionEnvironment jobExecutionEnv =
+            ClientJobExecutionEnvironment jobExecutionEnv =
                     engineClient.createExecutionContext(testResources.getRight(), jobConfig);
             ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
 
@@ -230,7 +230,7 @@ public void testStreamJobRunOkIn2Node() throws ExecutionException, InterruptedEx
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
             clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
             engineClient = new SeaTunnelClient(clientConfig);
-            JobExecutionEnvironment jobExecutionEnv =
+            ClientJobExecutionEnvironment jobExecutionEnv =
                     engineClient.createExecutionContext(testResources.getRight(), jobConfig);
             ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
 
@@ -325,7 +325,7 @@ public void testBatchJobRestoreIn2NodeWorkerDown()
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
             clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
             engineClient = new SeaTunnelClient(clientConfig);
-            JobExecutionEnvironment jobExecutionEnv =
+            ClientJobExecutionEnvironment jobExecutionEnv =
                     engineClient.createExecutionContext(testResources.getRight(), jobConfig);
             ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
 
@@ -422,7 +422,7 @@ public void testStreamJobRestoreIn2NodeWorkerDown()
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
             clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
             engineClient = new SeaTunnelClient(clientConfig);
-            JobExecutionEnvironment jobExecutionEnv =
+            ClientJobExecutionEnvironment jobExecutionEnv =
                     engineClient.createExecutionContext(testResources.getRight(), jobConfig);
             ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
 
@@ -540,7 +540,7 @@ public void testBatchJobRestoreIn2NodeMasterDown()
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
             clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
             engineClient = new SeaTunnelClient(clientConfig);
-            JobExecutionEnvironment jobExecutionEnv =
+            ClientJobExecutionEnvironment jobExecutionEnv =
                     engineClient.createExecutionContext(testResources.getRight(), jobConfig);
             ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
 
@@ -642,7 +642,7 @@ public void testStreamJobRestoreIn2NodeMasterDown()
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
             clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
             engineClient = new SeaTunnelClient(clientConfig);
-            JobExecutionEnvironment jobExecutionEnv =
+            ClientJobExecutionEnvironment jobExecutionEnv =
                     engineClient.createExecutionContext(testResources.getRight(), jobConfig);
             ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
 
@@ -810,7 +810,7 @@ public void testStreamJobRestoreInAllNodeDown()
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
             clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
             engineClient = new SeaTunnelClient(clientConfig);
-            JobExecutionEnvironment jobExecutionEnv =
+            ClientJobExecutionEnvironment jobExecutionEnv =
                     engineClient.createExecutionContext(testResources.getRight(), jobConfig);
             ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
             Long jobId = clientJobProxy.getJobId();
@@ -1028,7 +1028,7 @@ public void testStreamJobRestoreFromOssInAllNodeDown()
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
             clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
             engineClient = new SeaTunnelClient(clientConfig);
-            JobExecutionEnvironment jobExecutionEnv =
+            ClientJobExecutionEnvironment jobExecutionEnv =
                     engineClient.createExecutionContext(testResources.getRight(), jobConfig);
             ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
             Long jobId = clientJobProxy.getJobId();

diff --git a/...ase/src/test/java/org/apache/seatunnel/engine/e2e/ClusterFaultToleranceTwoPipelineIT.java b/...ase/src/test/java/org/apache/seatunnel/engine/e2e/ClusterFaultToleranceTwoPipelineIT.java
@@ -22,8 +22,8 @@
 import org.apache.seatunnel.common.constants.JobMode;
 import org.apache.seatunnel.common.utils.FileUtils;
 import org.apache.seatunnel.engine.client.SeaTunnelClient;
+import org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment;
 import org.apache.seatunnel.engine.client.job.ClientJobProxy;
-import org.apache.seatunnel.engine.client.job.JobExecutionEnvironment;
 import org.apache.seatunnel.engine.common.config.ConfigProvider;
 import org.apache.seatunnel.engine.common.config.JobConfig;
 import org.apache.seatunnel.engine.common.config.SeaTunnelConfig;
@@ -115,7 +115,7 @@ public void testTwoPipelineBatchJobRunOkIn2Node()
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
             clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
             engineClient = new SeaTunnelClient(clientConfig);
-            JobExecutionEnvironment jobExecutionEnv =
+            ClientJobExecutionEnvironment jobExecutionEnv =
                     engineClient.createExecutionContext(testResources.getRight(), jobConfig);
             ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
 
@@ -243,7 +243,7 @@ public void testTwoPipelineStreamJobRunOkIn2Node()
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
             clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
             engineClient = new SeaTunnelClient(clientConfig);
-            JobExecutionEnvironment jobExecutionEnv =
+            ClientJobExecutionEnvironment jobExecutionEnv =
                     engineClient.createExecutionContext(testResources.getRight(), jobConfig);
             ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
 
@@ -341,7 +341,7 @@ public void testTwoPipelineBatchJobRestoreIn2NodeWorkerDown()
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
             clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
             engineClient = new SeaTunnelClient(clientConfig);
-            JobExecutionEnvironment jobExecutionEnv =
+            ClientJobExecutionEnvironment jobExecutionEnv =
                     engineClient.createExecutionContext(testResources.getRight(), jobConfig);
             ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
 
@@ -456,7 +456,7 @@ public void testTwoPipelineStreamJobRestoreIn2NodeWorkerDown()
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
             clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
             engineClient = new SeaTunnelClient(clientConfig);
-            JobExecutionEnvironment jobExecutionEnv =
+            ClientJobExecutionEnvironment jobExecutionEnv =
                     engineClient.createExecutionContext(testResources.getRight(), jobConfig);
             ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
 
@@ -586,7 +586,7 @@ public void testTwoPipelineBatchJobRestoreIn2NodeMasterDown()
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
             clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
             engineClient = new SeaTunnelClient(clientConfig);
-            JobExecutionEnvironment jobExecutionEnv =
+            ClientJobExecutionEnvironment jobExecutionEnv =
                     engineClient.createExecutionContext(testResources.getRight(), jobConfig);
             ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
 
@@ -696,7 +696,7 @@ public void testTwoPipelineStreamJobRestoreIn2NodeMasterDown()
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
             clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
             engineClient = new SeaTunnelClient(clientConfig);
-            JobExecutionEnvironment jobExecutionEnv =
+            ClientJobExecutionEnvironment jobExecutionEnv =
                     engineClient.createExecutionContext(testResources.getRight(), jobConfig);
             ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
 

diff --git a/...ctor-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/JobExecutionIT.java b/...ctor-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/JobExecutionIT.java
@@ -20,8 +20,8 @@
 import org.apache.seatunnel.common.config.Common;
 import org.apache.seatunnel.common.config.DeployMode;
 import org.apache.seatunnel.engine.client.SeaTunnelClient;
+import org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment;
 import org.apache.seatunnel.engine.client.job.ClientJobProxy;
-import org.apache.seatunnel.engine.client.job.JobExecutionEnvironment;
 import org.apache.seatunnel.engine.common.config.ConfigProvider;
 import org.apache.seatunnel.engine.common.config.JobConfig;
 import org.apache.seatunnel.engine.core.job.JobResult;
@@ -76,7 +76,7 @@ public void testExecuteJob() throws Exception {
         ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
         clientConfig.setClusterName(TestUtils.getClusterName("JobExecutionIT"));
         SeaTunnelClient engineClient = new SeaTunnelClient(clientConfig);
-        JobExecutionEnvironment jobExecutionEnv =
+        ClientJobExecutionEnvironment jobExecutionEnv =
                 engineClient.createExecutionContext(filePath, jobConfig);
 
         final ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
@@ -103,7 +103,7 @@ public void cancelJobTest() throws Exception {
         ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
         clientConfig.setClusterName(TestUtils.getClusterName("JobExecutionIT"));
         SeaTunnelClient engineClient = new SeaTunnelClient(clientConfig);
-        JobExecutionEnvironment jobExecutionEnv =
+        ClientJobExecutionEnvironment jobExecutionEnv =
                 engineClient.createExecutionContext(filePath, jobConfig);
 
         final ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
@@ -132,7 +132,7 @@ public void testGetErrorInfo() throws ExecutionException, InterruptedException {
         ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
         clientConfig.setClusterName(TestUtils.getClusterName("JobExecutionIT"));
         SeaTunnelClient engineClient = new SeaTunnelClient(clientConfig);
-        JobExecutionEnvironment jobExecutionEnv =
+        ClientJobExecutionEnvironment jobExecutionEnv =
                 engineClient.createExecutionContext(filePath, jobConfig);
         final ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
         CompletableFuture<JobStatus> completableFuture =
@@ -177,7 +177,7 @@ public void testExpiredJobWasDeleted() throws Exception {
         ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
         clientConfig.setClusterName(TestUtils.getClusterName("JobExecutionIT"));
         SeaTunnelClient engineClient = new SeaTunnelClient(clientConfig);
-        JobExecutionEnvironment jobExecutionEnv =
+        ClientJobExecutionEnvironment jobExecutionEnv =
                 engineClient.createExecutionContext(filePath, jobConfig);
 
         final ClientJobProxy clientJobProxy = jobExecutionEnv.execute();

diff --git a/...connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/RestApiIT.java b/...connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/RestApiIT.java
@@ -20,8 +20,8 @@
 import org.apache.seatunnel.common.config.Common;
 import org.apache.seatunnel.common.config.DeployMode;
 import org.apache.seatunnel.engine.client.SeaTunnelClient;
+import org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment;
 import org.apache.seatunnel.engine.client.job.ClientJobProxy;
-import org.apache.seatunnel.engine.client.job.JobExecutionEnvironment;
 import org.apache.seatunnel.engine.common.Constant;
 import org.apache.seatunnel.engine.common.config.ConfigProvider;
 import org.apache.seatunnel.engine.common.config.JobConfig;
@@ -71,7 +71,7 @@ void beforeClass() throws Exception {
         ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
         clientConfig.setClusterName(testClusterName);
         SeaTunnelClient engineClient = new SeaTunnelClient(clientConfig);
-        JobExecutionEnvironment jobExecutionEnv =
+        ClientJobExecutionEnvironment jobExecutionEnv =
                 engineClient.createExecutionContext(filePath, jobConfig);
 
         clientJobProxy = jobExecutionEnv.execute();

diff --git a/...nector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/TextHeaderIT.java b/...nector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/TextHeaderIT.java
@@ -22,8 +22,8 @@
 import org.apache.seatunnel.common.utils.FileUtils;
 import org.apache.seatunnel.connectors.seatunnel.file.config.BaseSinkConfig;
 import org.apache.seatunnel.engine.client.SeaTunnelClient;
+import org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment;
 import org.apache.seatunnel.engine.client.job.ClientJobProxy;
-import org.apache.seatunnel.engine.client.job.JobExecutionEnvironment;
 import org.apache.seatunnel.engine.common.config.ConfigProvider;
 import org.apache.seatunnel.engine.common.config.JobConfig;
 import org.apache.seatunnel.engine.common.config.SeaTunnelConfig;
@@ -130,7 +130,7 @@ public void enableWriteHeader(String file_format_type, String headerWrite, Strin
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
             clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
             engineClient = new SeaTunnelClient(clientConfig);
-            JobExecutionEnvironment jobExecutionEnv =
+            ClientJobExecutionEnvironment jobExecutionEnv =
                     engineClient.createExecutionContext(testResources.getRight(), jobConfig);
             ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
 

diff --git a/...unnel-engine-client/src/main/java/org/apache/seatunnel/engine/client/SeaTunnelClient.java b/...unnel-engine-client/src/main/java/org/apache/seatunnel/engine/client/SeaTunnelClient.java
@@ -18,8 +18,8 @@
 package org.apache.seatunnel.engine.client;
 
 import org.apache.seatunnel.common.utils.JsonUtils;
+import org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment;
 import org.apache.seatunnel.engine.client.job.JobClient;
-import org.apache.seatunnel.engine.client.job.JobExecutionEnvironment;
 import org.apache.seatunnel.engine.client.job.JobMetricsRunner.JobMetricsSummary;
 import org.apache.seatunnel.engine.common.config.JobConfig;
 import org.apache.seatunnel.engine.core.job.JobDAGInfo;
@@ -48,15 +48,15 @@ public SeaTunnelClient(@NonNull ClientConfig clientConfig) {
     }
 
     @Override
-    public JobExecutionEnvironment createExecutionContext(
+    public ClientJobExecutionEnvironment createExecutionContext(
             @NonNull String filePath, @NonNull JobConfig jobConfig) {
-        return new JobExecutionEnvironment(jobConfig, filePath, hazelcastClient);
+        return new ClientJobExecutionEnvironment(jobConfig, filePath, hazelcastClient);
     }
 
     @Override
-    public JobExecutionEnvironment restoreExecutionContext(
+    public ClientJobExecutionEnvironment restoreExecutionContext(
             @NonNull String filePath, @NonNull JobConfig jobConfig, @NonNull Long jobId) {
-        return new JobExecutionEnvironment(jobConfig, filePath, hazelcastClient, true, jobId);
+        return new ClientJobExecutionEnvironment(jobConfig, filePath, hazelcastClient, true, jobId);
     }
 
     @Override

diff --git a/...gine-client/src/main/java/org/apache/seatunnel/engine/client/SeaTunnelClientInstance.java b/...gine-client/src/main/java/org/apache/seatunnel/engine/client/SeaTunnelClientInstance.java
@@ -17,18 +17,18 @@
 
 package org.apache.seatunnel.engine.client;
 
+import org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment;
 import org.apache.seatunnel.engine.client.job.JobClient;
-import org.apache.seatunnel.engine.client.job.JobExecutionEnvironment;
 import org.apache.seatunnel.engine.common.config.JobConfig;
 
 import lombok.NonNull;
 
 public interface SeaTunnelClientInstance {
 
-    JobExecutionEnvironment createExecutionContext(
+    ClientJobExecutionEnvironment createExecutionContext(
             @NonNull String filePath, @NonNull JobConfig config);
 
-    JobExecutionEnvironment restoreExecutionContext(
+    ClientJobExecutionEnvironment restoreExecutionContext(
             @NonNull String filePath, @NonNull JobConfig config, @NonNull Long jobId);
 
     JobClient createJobClient();

diff --git a/...e/client/job/JobExecutionEnvironment.java → ...nt/job/ClientJobExecutionEnvironment.java b/...e/client/job/JobExecutionEnvironment.java → ...nt/job/ClientJobExecutionEnvironment.java
@@ -27,7 +27,7 @@
 import java.util.ArrayList;
 import java.util.concurrent.ExecutionException;
 
-public class JobExecutionEnvironment extends AbstractJobEnvironment {
+public class ClientJobExecutionEnvironment extends AbstractJobEnvironment {
 
     private final String jobFilePath;
 
@@ -36,7 +36,7 @@ public class JobExecutionEnvironment extends AbstractJobEnvironment {
     private final JobClient jobClient;
 
     /** If the JobId is not empty, it is used to restore job from savePoint */
-    public JobExecutionEnvironment(
+    public ClientJobExecutionEnvironment(
             JobConfig jobConfig,
             String jobFilePath,
             SeaTunnelHazelcastClient seaTunnelHazelcastClient,
@@ -50,7 +50,7 @@ public JobExecutionEnvironment(
                 new JobContext(isStartWithSavePoint ? jobId : jobClient.getNewJobId()));
     }
 
-    public JobExecutionEnvironment(
+    public ClientJobExecutionEnvironment(
             JobConfig jobConfig,
             String jobFilePath,
             SeaTunnelHazelcastClient seaTunnelHazelcastClient) {