[SeaTunnel API] [Sink] remove useless context field (#2124)

Author: Hisoka-X

docs/en/connector/source/Http.md

@@ -37,7 +37,7 @@ HTTP request header, json format.
 
 ### request_params[string]
 
-HTTP request parameters, json format.
+HTTP request parameters, json format. Use string with escapes to save json
 
 ### sync_path[string]
 

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

@@ -17,35 +17,19 @@
 
 package org.apache.seatunnel.api.sink;
 
-import java.util.Map;
-
 /**
  * The default {@link SinkWriter.Context} implement class.
  */
 public class DefaultSinkWriterContext implements SinkWriter.Context {
-
-    private final Map<String, String> configuration;
     private final int subtask;
-    private final int parallelism;
 
-    public DefaultSinkWriterContext(Map<String, String> configuration, int subtask, int parallelism) {
-        this.configuration = configuration;
+    public DefaultSinkWriterContext(int subtask) {
         this.subtask = subtask;
-        this.parallelism = parallelism;
-    }
-
-    @Override
-    public Map<String, String> getConfiguration() {
-        return configuration;
     }
 
     @Override
     public int getIndexOfSubtask() {
         return subtask;
     }
 
-    @Override
-    public int getNumberOfParallelSubtasks() {
-        return parallelism;
-    }
 }

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

@@ -21,7 +21,6 @@
 import java.io.Serializable;
 import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 
 /**
@@ -76,20 +75,10 @@ default List<StateT> snapshotState(long checkpointId) throws IOException {
 
     interface Context extends Serializable{
 
-        /**
-         * Gets the configuration with which Job was started.
-         */
-        Map<String, String> getConfiguration();
-
         /**
          * @return The index of this subtask.
          */
         int getIndexOfSubtask();
 
-        /**
-         * @return The number of parallel Sink tasks.
-         */
-        int getNumberOfParallelSubtasks();
-
     }
 }

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/SemanticXidGenerator.java

@@ -80,8 +80,7 @@ public boolean belongsToSubtask(Xid xid, SeaTunnelContext context, SinkWriter.Co
             return false;
         }
         int subtaskIndex = readNumber(xid.getGlobalTransactionId(), JOB_ID_BYTES, Integer.BYTES);
-        if (subtaskIndex != sinkContext.getIndexOfSubtask()
-            && subtaskIndex <= sinkContext.getNumberOfParallelSubtasks() - 1) {
+        if (subtaskIndex != sinkContext.getIndexOfSubtask()) {
             return false;
         }
         byte[] jobIdBytes = new byte[JOB_ID_BYTES];

seatunnel-core/seatunnel-flink-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java

@@ -36,7 +36,6 @@
 
 import java.net.URL;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -76,7 +75,7 @@ public List<DataStream<Row>> execute(List<DataStream<Row>> upstreamDataStreams)
             SeaTunnelSink<SeaTunnelRow, Serializable, Serializable, Serializable> seaTunnelSink = plugins.get(i);
             DataStream<Row> stream = fromSourceTable(sinkConfig).orElse(input);
             seaTunnelSink.setTypeInfo((SeaTunnelRowType) TypeConverterUtils.convert(stream.getType()));
-            stream.sinkTo(new FlinkSink<>(seaTunnelSink, Collections.emptyMap()));
+            stream.sinkTo(new FlinkSink<>(seaTunnelSink));
         }
         // the sink is the last stream
         return null;

seatunnel-core/seatunnel-spark-starter/src/main/java/org/apache/seatunnel/core/starter/spark/execution/SinkExecuteProcessor.java

@@ -20,7 +20,6 @@
 import org.apache.seatunnel.api.common.SeaTunnelContext;
 import org.apache.seatunnel.api.sink.SeaTunnelSink;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
-import org.apache.seatunnel.common.config.Common;
 import org.apache.seatunnel.core.starter.exception.TaskExecuteException;
 import org.apache.seatunnel.plugin.discovery.PluginIdentifier;
 import org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelSinkPluginDiscovery;
@@ -36,7 +35,6 @@
 
 import java.net.URL;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -74,8 +72,7 @@ public List<Dataset<Row>> execute(List<Dataset<Row>> upstreamDataStreams) throws
             Dataset<Row> dataset = fromSourceTable(sinkConfig, sparkEnvironment).orElse(input);
             // TODO modify checkpoint location
             seaTunnelSink.setTypeInfo((SeaTunnelRowType) TypeConverterUtils.convert(dataset.schema()));
-            SparkSinkInjector.inject(dataset.write(), seaTunnelSink, new HashMap<>(Common.COLLECTION_SIZE)).option(
-                "checkpointLocation", "/tmp").save();
+            SparkSinkInjector.inject(dataset.write(), seaTunnelSink).option("checkpointLocation", "/tmp").save();
         }
         // the sink is the last stream
         return null;

seatunnel-translation/seatunnel-translation-flink/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSink.java

@@ -32,28 +32,21 @@
 
 import java.io.IOException;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
-@SuppressWarnings("unchecked")
 public class FlinkSink<InputT, CommT, WriterStateT, GlobalCommT> implements Sink<InputT, CommitWrapper<CommT>,
         FlinkWriterState<WriterStateT>, GlobalCommT> {
 
     private final SeaTunnelSink<SeaTunnelRow, WriterStateT, CommT, GlobalCommT> sink;
-    private final Map<String, String> configuration;
 
-    public FlinkSink(SeaTunnelSink<SeaTunnelRow, WriterStateT, CommT, GlobalCommT> sink,
-              Map<String, String> configuration) {
+    public FlinkSink(SeaTunnelSink<SeaTunnelRow, WriterStateT, CommT, GlobalCommT> sink) {
         this.sink = sink;
-        this.configuration = configuration;
     }
 
     @Override
     public SinkWriter<InputT, CommitWrapper<CommT>, FlinkWriterState<WriterStateT>> createWriter(org.apache.flink.api.connector.sink.Sink.InitContext context, List<FlinkWriterState<WriterStateT>> states) throws IOException {
-        // TODO add subtask and parallelism.
-        org.apache.seatunnel.api.sink.SinkWriter.Context stContext =
-                new DefaultSinkWriterContext(configuration, 0, 0);
+        org.apache.seatunnel.api.sink.SinkWriter.Context stContext = new DefaultSinkWriterContext(context.getSubtaskId());
 
         if (states == null || states.isEmpty()) {
             return new FlinkSinkWriter<>(sink.createWriter(stContext), 1, sink.getConsumedType());

seatunnel-translation/seatunnel-translation-spark/src/main/java/org/apache/seatunnel/translation/spark/sink/SparkDataSourceWriter.java

@@ -33,27 +33,24 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
 public class SparkDataSourceWriter<StateT, CommitInfoT, AggregatedCommitInfoT> implements DataSourceWriter {
 
     protected final SeaTunnelSink<SeaTunnelRow, StateT, CommitInfoT, AggregatedCommitInfoT> sink;
-    protected final Map<String, String> configuration;
     @Nullable
     protected final SinkAggregatedCommitter<CommitInfoT, AggregatedCommitInfoT> sinkAggregatedCommitter;
 
-    SparkDataSourceWriter(SeaTunnelSink<SeaTunnelRow, StateT, CommitInfoT, AggregatedCommitInfoT> sink,
-                          Map<String, String> configuration) throws IOException {
+    SparkDataSourceWriter(SeaTunnelSink<SeaTunnelRow, StateT, CommitInfoT, AggregatedCommitInfoT> sink)
+            throws IOException {
         this.sink = sink;
-        this.configuration = configuration;
         this.sinkAggregatedCommitter = sink.createAggregatedCommitter().orElse(null);
     }
 
     @Override
     public DataWriterFactory<InternalRow> createWriterFactory() {
-        return new SparkDataWriterFactory<>(sink, configuration);
+        return new SparkDataWriterFactory<>(sink);
     }
 
     @Override

seatunnel-translation/seatunnel-translation-spark/src/main/java/org/apache/seatunnel/translation/spark/sink/SparkDataWriterFactory.java

@@ -28,23 +28,17 @@
 import org.apache.spark.sql.sources.v2.writer.DataWriterFactory;
 
 import java.io.IOException;
-import java.util.Map;
 
 public class SparkDataWriterFactory<CommitInfoT, StateT> implements DataWriterFactory<InternalRow> {
 
     private final SeaTunnelSink<SeaTunnelRow, StateT, CommitInfoT, ?> sink;
-    private final Map<String, String> configuration;
-    SparkDataWriterFactory(SeaTunnelSink<SeaTunnelRow, StateT, CommitInfoT, ?> sink,  Map<String, String> configuration) {
+    SparkDataWriterFactory(SeaTunnelSink<SeaTunnelRow, StateT, CommitInfoT, ?> sink) {
         this.sink = sink;
-        this.configuration = configuration;
     }
 
     @Override
     public DataWriter<InternalRow> createDataWriter(int partitionId, long taskId, long epochId) {
-        // TODO use partitionID, taskId information.
-        // TODO add subtask and parallelism.
-        org.apache.seatunnel.api.sink.SinkWriter.Context context =
-            new DefaultSinkWriterContext(configuration, (int) taskId, 0);
+        org.apache.seatunnel.api.sink.SinkWriter.Context context = new DefaultSinkWriterContext((int) taskId);
         SinkWriter<SeaTunnelRow, CommitInfoT, StateT> writer;
         SinkCommitter<CommitInfoT> committer;
         try {

seatunnel-translation/seatunnel-translation-spark/src/main/java/org/apache/seatunnel/translation/spark/sink/SparkSink.java

@@ -32,23 +32,18 @@
 import org.apache.spark.sql.types.StructType;
 
 import java.io.IOException;
-import java.util.Map;
 import java.util.Optional;
 
 public class SparkSink<StateT, CommitInfoT, AggregatedCommitInfoT> implements WriteSupport,
     StreamWriteSupport, DataSourceV2 {
 
     private volatile SeaTunnelSink<SeaTunnelRow, StateT, CommitInfoT, AggregatedCommitInfoT> sink;
-    private Map<String, String> configuration;
 
     private void init(DataSourceOptions options) {
         if (sink == null) {
             this.sink = SerializationUtils.stringToObject(
                     options.get("sink").orElseThrow(() -> new IllegalArgumentException("can not find sink " +
                             "class string in DataSourceOptions")));
-            this.configuration = SerializationUtils.stringToObject(
-                    options.get("configuration").orElseThrow(() -> new IllegalArgumentException("can not " +
-                            "find configuration class string in DataSourceOptions")));
         }
     }
 
@@ -57,7 +52,7 @@ public StreamWriter createStreamWriter(String queryId, StructType schema, Output
         init(options);
 
         try {
-            return new SparkStreamWriter<>(sink, configuration);
+            return new SparkStreamWriter<>(sink);
         } catch (IOException e) {
             throw new RuntimeException("find error when createStreamWriter", e);
         }
@@ -68,7 +63,7 @@ public Optional<DataSourceWriter> createWriter(String writeUUID, StructType sche
         init(options);
 
         try {
-            return Optional.of(new SparkDataSourceWriter<>(sink, configuration));
+            return Optional.of(new SparkDataSourceWriter<>(sink));
         } catch (IOException e) {
             throw new RuntimeException("find error when createStreamWriter", e);
         }