[DEV][Api] Replace SeaTunnelContext with JobContext and remove singleton pattern (#2706)

* remove singleton pattern for SeaTunnelContext

* rename SeaTunnelContext to JobContext and fix checkstyle

* fix checkstyle

* put the constructor on top

Author: ic4y

seatunnel-api/src/main/java/org/apache/seatunnel/api/common/SeaTunnelContext.java renamed to seatunnel-api/src/main/java/org/apache/seatunnel/api/common/JobContext.java

@@ -28,23 +28,21 @@
 import java.util.concurrent.ConcurrentHashMap;
 
 /**
- * This class is used to store the context of the application. e.g. the table schema, catalog...etc.
+ * This class is used to store the context of the job. e.g. the table schema, catalog...etc.
  */
-public final class SeaTunnelContext implements Serializable {
+public final class JobContext implements Serializable {
 
     private static final long serialVersionUID = -1L;
 
-    private static final SeaTunnelContext INSTANCE = new SeaTunnelContext();
-
     // tableName -> tableSchema
     private final Map<String, TableSchema> tableSchemaMap = new ConcurrentHashMap<>(Common.COLLECTION_SIZE);
 
     private JobMode jobMode;
 
     private final String jobId;
 
-    public static SeaTunnelContext getContext() {
-        return INSTANCE;
+    public JobContext() {
+        this.jobId = UUID.randomUUID().toString().replace("-", "");
     }
 
     /**
@@ -67,7 +65,7 @@ public Optional<TableSchema> getSchema(String tableName) {
         return Optional.ofNullable(tableSchemaMap.get(tableName));
     }
 
-    public SeaTunnelContext setJobMode(JobMode jobMode) {
+    public JobContext setJobMode(JobMode jobMode) {
         this.jobMode = jobMode;
         return this;
     }
@@ -80,8 +78,4 @@ public String getJobId() {
         return this.jobId;
     }
 
-    private SeaTunnelContext() {
-        this.jobId = UUID.randomUUID().toString().replace("-", "");
-    }
-
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/SeaTunnelSink.java

@@ -20,7 +20,7 @@
 import org.apache.seatunnel.api.common.PluginIdentifierInterface;
 import org.apache.seatunnel.api.common.SeaTunnelPluginLifeCycle;
 import org.apache.seatunnel.api.serialization.Serializer;
-import org.apache.seatunnel.api.source.SeaTunnelContextAware;
+import org.apache.seatunnel.api.source.SeaTunnelJobAware;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 
@@ -43,7 +43,7 @@
  *                                {@link SinkAggregatedCommitter} handle it, this class should implement interface {@link Serializable}.
  */
 public interface SeaTunnelSink<IN, StateT, CommitInfoT, AggregatedCommitInfoT>
-    extends Serializable, PluginIdentifierInterface, SeaTunnelPluginLifeCycle, SeaTunnelContextAware {
+    extends Serializable, PluginIdentifierInterface, SeaTunnelPluginLifeCycle, SeaTunnelJobAware {
 
     /**
      * Set the row type info of sink row data. This method will be automatically called by translation.

seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SeaTunnelContextAware.java renamed to seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SeaTunnelJobAware.java

@@ -17,14 +17,14 @@
 
 package org.apache.seatunnel.api.source;
 
-import org.apache.seatunnel.api.common.SeaTunnelContext;
+import org.apache.seatunnel.api.common.JobContext;
 
 /**
- * This interface defines the runtime environment of the SeaTunnel application.
+ * This interface defines the runtime environment of the SeaTunnel job.
  */
-public interface SeaTunnelContextAware {
+public interface SeaTunnelJobAware {
 
-    default void setSeaTunnelContext(SeaTunnelContext seaTunnelContext){
+    default void setJobContext(JobContext jobContext){
         // nothing
     }
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SeaTunnelSource.java

@@ -34,7 +34,7 @@
  * @param <StateT> The type of checkpoint states.
  */
 public interface SeaTunnelSource<T, SplitT extends SourceSplit, StateT extends Serializable>
-    extends Serializable, PluginIdentifierInterface, SeaTunnelPluginLifeCycle, SeaTunnelContextAware {
+    extends Serializable, PluginIdentifierInterface, SeaTunnelPluginLifeCycle, SeaTunnelJobAware {
 
     /**
      * Get the boundedness of this source.

seatunnel-connectors-v2/connector-assert/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertSink.java

@@ -17,7 +17,6 @@
 
 package org.apache.seatunnel.connectors.seatunnel.assertion.sink;
 
-import org.apache.seatunnel.api.common.SeaTunnelContext;
 import org.apache.seatunnel.api.sink.SeaTunnelSink;
 import org.apache.seatunnel.api.sink.SinkWriter;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
@@ -41,7 +40,6 @@
 @AutoService(SeaTunnelSink.class)
 public class AssertSink extends AbstractSimpleSink<SeaTunnelRow, Void> {
     private static final String RULES = "rules";
-    private SeaTunnelContext seaTunnelContext;
     private SeaTunnelRowType seaTunnelRowType;
     private List<AssertFieldRule> assertFieldRules;
 
@@ -73,11 +71,6 @@ public void prepare(Config pluginConfig) {
         assertFieldRules = new AssertRuleParser().parseRules(configList);
     }
 
-    @Override
-    public void setSeaTunnelContext(SeaTunnelContext seaTunnelContext) {
-        this.seaTunnelContext = seaTunnelContext;
-    }
-
     @Override
     public String getPluginName() {
         return "Assert";

seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ClickhouseSink.java

@@ -29,7 +29,6 @@
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.Config.USERNAME;
 
 import org.apache.seatunnel.api.common.PrepareFailException;
-import org.apache.seatunnel.api.common.SeaTunnelContext;
 import org.apache.seatunnel.api.serialization.DefaultSerializer;
 import org.apache.seatunnel.api.serialization.Serializer;
 import org.apache.seatunnel.api.sink.SeaTunnelSink;
@@ -67,7 +66,6 @@
 @AutoService(SeaTunnelSink.class)
 public class ClickhouseSink implements SeaTunnelSink<SeaTunnelRow, ClickhouseSinkState, CKCommitInfo, CKAggCommitInfo> {
 
-    private SeaTunnelContext seaTunnelContext;
     private ReaderOption option;
 
     @Override
@@ -194,8 +192,4 @@ public SeaTunnelDataType<SeaTunnelRow> getConsumedType() {
         return this.option.getSeaTunnelRowType();
     }
 
-    @Override
-    public void setSeaTunnelContext(SeaTunnelContext seaTunnelContext) {
-        this.seaTunnelContext = seaTunnelContext;
-    }
 }

seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/file/ClickhouseFileSink.java

@@ -30,7 +30,6 @@
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.Config.USERNAME;
 
 import org.apache.seatunnel.api.common.PrepareFailException;
-import org.apache.seatunnel.api.common.SeaTunnelContext;
 import org.apache.seatunnel.api.sink.SeaTunnelSink;
 import org.apache.seatunnel.api.sink.SinkWriter;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
@@ -65,7 +64,6 @@
 @AutoService(SeaTunnelSink.class)
 public class ClickhouseFileSink implements SeaTunnelSink<SeaTunnelRow, ClickhouseSinkState, CKCommitInfo, CKAggCommitInfo> {
 
-    private SeaTunnelContext seaTunnelContext;
     private FileReaderOption readerOption;
 
     @Override
@@ -140,9 +138,4 @@ public SeaTunnelDataType<SeaTunnelRow> getConsumedType() {
     public SinkWriter<SeaTunnelRow, CKCommitInfo, ClickhouseSinkState> createWriter(SinkWriter.Context context) throws IOException {
         return new ClickhouseFileSinkWriter(readerOption, context);
     }
-
-    @Override
-    public void setSeaTunnelContext(SeaTunnelContext seaTunnelContext) {
-        this.seaTunnelContext = seaTunnelContext;
-    }
 }

seatunnel-connectors-v2/connector-elasticsearch/src/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClient.java

@@ -17,6 +17,10 @@
 
 package org.apache.seatunnel.connectors.seatunnel.elasticsearch.client;
 
+import org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.BulkResponse;
+import org.apache.seatunnel.connectors.seatunnel.elasticsearch.exception.BulkElasticsearchException;
+import org.apache.seatunnel.connectors.seatunnel.elasticsearch.exception.GetElasticsearchVersionException;
+
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.commons.lang3.StringUtils;
@@ -27,9 +31,6 @@
 import org.apache.http.client.CredentialsProvider;
 import org.apache.http.impl.client.BasicCredentialsProvider;
 import org.apache.http.util.EntityUtils;
-import org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.BulkResponse;
-import org.apache.seatunnel.connectors.seatunnel.elasticsearch.exception.BulkElasticsearchException;
-import org.apache.seatunnel.connectors.seatunnel.elasticsearch.exception.GetElasticsearchVersionException;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.RestClient;
@@ -40,13 +41,14 @@
 
 public class EsRestClient {
 
-    private static EsRestClient esRestClient;
-    private static RestClient restClient;
+    private static EsRestClient ES_REST_CLIENT;
+    private static RestClient REST_CLIENT;
 
     private EsRestClient() {
 
     }
 
+    @SuppressWarnings("checkstyle:MagicNumber")
     private static RestClientBuilder getRestClientBuilder(List<String> hosts, String username, String password) {
         HttpHost[] httpHosts = new HttpHost[hosts.size()];
         for (int i = 0; i < hosts.size(); i++) {
@@ -68,19 +70,19 @@ private static RestClientBuilder getRestClientBuilder(List<String> hosts, String
     }
 
     public static EsRestClient getInstance(List<String> hosts, String username, String password) {
-        if (restClient == null) {
+        if (REST_CLIENT == null) {
             RestClientBuilder restClientBuilder = getRestClientBuilder(hosts, username, password);
-            restClient = restClientBuilder.build();
-            esRestClient = new EsRestClient();
+            REST_CLIENT = restClientBuilder.build();
+            ES_REST_CLIENT = new EsRestClient();
         }
-        return esRestClient;
+        return ES_REST_CLIENT;
     }
 
     public BulkResponse bulk(String requestBody) {
         Request request = new Request("POST", "_bulk");
         request.setJsonEntity(requestBody);
         try {
-            Response response = restClient.performRequest(request);
+            Response response = REST_CLIENT.performRequest(request);
             if (response == null) {
                 throw new BulkElasticsearchException("bulk es Response is null");
             }
@@ -105,7 +107,7 @@ public BulkResponse bulk(String requestBody) {
     public static String getClusterVersion() {
         Request request = new Request("GET", "/");
         try {
-            Response response = restClient.performRequest(request);
+            Response response = REST_CLIENT.performRequest(request);
             String result = EntityUtils.toString(response.getEntity());
             ObjectMapper objectMapper = new ObjectMapper();
             JsonNode jsonNode = objectMapper.readTree(result);
@@ -117,7 +119,7 @@ public static String getClusterVersion() {
     }
 
     public void close() throws IOException {
-        restClient.close();
+        REST_CLIENT.close();
     }
 
 }

seatunnel-connectors-v2/connector-elasticsearch/src/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/SinkConfig.java

@@ -35,11 +35,11 @@ public class SinkConfig {
 
     public static final String MAX_RETRY_SIZE = "max_retry_size";
 
-    public static void setValue(org.apache.seatunnel.shade.com.typesafe.config.Config pluginConfig){
-        if(pluginConfig.hasPath(MAX_BATCH_SIZE)){
+    public static void setValue(org.apache.seatunnel.shade.com.typesafe.config.Config pluginConfig) {
+        if (pluginConfig.hasPath(MAX_BATCH_SIZE)) {
             BulkConfig.MAX_BATCH_SIZE = pluginConfig.getInt(MAX_BATCH_SIZE);
         }
-        if(pluginConfig.hasPath(MAX_RETRY_SIZE)){
+        if (pluginConfig.hasPath(MAX_RETRY_SIZE)) {
             BulkConfig.MAX_RETRY_SIZE = pluginConfig.getInt(MAX_RETRY_SIZE);
         }
     }

seatunnel-connectors-v2/connector-elasticsearch/src/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/constant/BulkConfig.java

@@ -27,11 +27,13 @@ public class BulkConfig {
      * once bulk es include max document size
      * {@link SinkConfig#MAX_BATCH_SIZE}
      */
+    @SuppressWarnings("checkstyle:MagicNumber")
     public static int MAX_BATCH_SIZE = 10;
 
     /**
      * the max retry size of bulk es
      * {@link SinkConfig#MAX_RETRY_SIZE}
      */
+    @SuppressWarnings("checkstyle:MagicNumber")
     public static int MAX_RETRY_SIZE = 3;
 }