[improve][SelectDB] Add a jobId to the selectDB label to distinguish between tasks (#4864)

Co-authored-by: zhouyao <[email protected]>

Author: Carl-Zhou-CN

seatunnel-connectors-v2/connector-selectdb-cloud/src/main/java/org/apache/seatunnel/connectors/selectdb/sink/SelectDBSink.java

@@ -19,6 +19,7 @@
 
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
+import org.apache.seatunnel.api.common.JobContext;
 import org.apache.seatunnel.api.common.PrepareFailException;
 import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
 import org.apache.seatunnel.api.serialization.Serializer;
@@ -59,6 +60,7 @@ public class SelectDBSink
                 SeaTunnelRow, SelectDBSinkState, SelectDBCommitInfo, SelectDBCommitInfo> {
     private Config pluginConfig;
     private SeaTunnelRowType seaTunnelRowType;
+    private String jobId;
 
     @Override
     public String getPluginName() {
@@ -85,6 +87,11 @@ public void prepare(Config pluginConfig) throws PrepareFailException {
         }
     }
 
+    @Override
+    public void setJobContext(JobContext jobContext) {
+        this.jobId = jobContext.getJobId();
+    }
+
     @Override
     public void setTypeInfo(SeaTunnelRowType seaTunnelRowType) {
         this.seaTunnelRowType = seaTunnelRowType;
@@ -98,20 +105,20 @@ public SeaTunnelDataType<SeaTunnelRow> getConsumedType() {
     @Override
     public SinkWriter<SeaTunnelRow, SelectDBCommitInfo, SelectDBSinkState> createWriter(
             SinkWriter.Context context) throws IOException {
-        SelectDBSinkWriter dorisWriter =
+        SelectDBSinkWriter selectDBSinkWriter =
                 new SelectDBSinkWriter(
-                        context, Collections.emptyList(), seaTunnelRowType, pluginConfig);
-        dorisWriter.initializeLoad(Collections.emptyList());
-        return dorisWriter;
+                        context, Collections.emptyList(), seaTunnelRowType, pluginConfig, jobId);
+        selectDBSinkWriter.initializeLoad(Collections.emptyList());
+        return selectDBSinkWriter;
     }
 
     @Override
     public SinkWriter<SeaTunnelRow, SelectDBCommitInfo, SelectDBSinkState> restoreWriter(
             SinkWriter.Context context, List<SelectDBSinkState> states) throws IOException {
-        SelectDBSinkWriter dorisWriter =
-                new SelectDBSinkWriter(context, states, seaTunnelRowType, pluginConfig);
-        dorisWriter.initializeLoad(states);
-        return dorisWriter;
+        SelectDBSinkWriter selectDBSinkWriter =
+                new SelectDBSinkWriter(context, states, seaTunnelRowType, pluginConfig, jobId);
+        selectDBSinkWriter.initializeLoad(states);
+        return selectDBSinkWriter;
     }
 
     @Override

seatunnel-connectors-v2/connector-selectdb-cloud/src/main/java/org/apache/seatunnel/connectors/selectdb/sink/writer/SelectDBSinkStateSerializer.java

@@ -29,10 +29,11 @@
 public class SelectDBSinkStateSerializer implements Serializer<SelectDBSinkState> {
 
     @Override
-    public byte[] serialize(SelectDBSinkState obj) throws IOException {
+    public byte[] serialize(SelectDBSinkState selectDBSinkState) throws IOException {
         try (final ByteArrayOutputStream baos = new ByteArrayOutputStream();
                 final DataOutputStream out = new DataOutputStream(baos)) {
-            out.writeUTF(obj.getLabelPrefix());
+            out.writeUTF(selectDBSinkState.getLabelPrefix());
+            out.writeLong(selectDBSinkState.getCheckpointId());
             out.flush();
             return baos.toByteArray();
         }

seatunnel-connectors-v2/connector-selectdb-cloud/src/main/java/org/apache/seatunnel/connectors/selectdb/sink/writer/SelectDBSinkWriter.java

@@ -54,15 +54,17 @@ public SelectDBSinkWriter(
             SinkWriter.Context context,
             List<SelectDBSinkState> state,
             SeaTunnelRowType seaTunnelRowType,
-            Config pluginConfig) {
+            Config pluginConfig,
+            String jobId) {
         this.selectdbConfig = SelectDBConfig.loadConfig(pluginConfig);
         this.lastCheckpointId = state.size() != 0 ? state.get(0).getCheckpointId() : 0;
         log.info("restore checkpointId {}", lastCheckpointId);
         // filename prefix is uuid
         log.info("labelPrefix " + selectdbConfig.getLabelPrefix());
         this.selectdbSinkState =
                 new SelectDBSinkState(selectdbConfig.getLabelPrefix(), lastCheckpointId);
-        this.labelPrefix = selectdbConfig.getLabelPrefix() + "_" + context.getIndexOfSubtask();
+        this.labelPrefix =
+                selectdbConfig.getLabelPrefix() + "_" + jobId + "_" + context.getIndexOfSubtask();
         this.lineDelimiter =
                 selectdbConfig
                         .getStageLoadProps()