[Bug]add 3node worker done test and fix some bug (#3115)

* Add master node switch test and fix bug

Add master node switch test and fix bug

* fix ci error

* add todo

* fix ci error

* fix ci error

* add testBatchJobRestoreIn3NodeWorkerDown and testStreamJobRestoreIn3NodeWorkerDown

* add testBatchJobRestoreIn3NodeMasterDown and testStreamJobRestoreIn3NodeMasterDown

* Add withTryCatch to the whenComplete which may throw exception

* Add withTryCatch to the whenComplete which may throw exception

* fix ci error

* improve test case

* notify checkpoint state

* add worker node and master node done test

* revert some file

* fix ci error

* improve test

* add checkpoint.interval to test job config

* fix some error

* add try catch to test case

* fix ci error

* fix checkpoint error

* [engine][tests] add cluster test case

* [engine] Close checkpoint at the end of the pipeline

* [engine] Fix hazelcast master node switching exception

* [hotfix][connector][file] fix file sink error

* fix checkpoint error

# Conflicts:
#	seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManager.java

* fix PipelineState to PipelineStatus bug

* [engine][checkpoint] add barrier flow operation (#3158)

* fix operator retry error

* fix operator retry error

* fix checkpoint error

* fix checkpoint error

* fix checkpoint error

* fix TaskExecutionService task is not exist when cancel task

Co-authored-by: Zongwen Li <zongwen@apache.org>

Author: EricJoy2048

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/BaseFileSinkWriter.java

@@ -46,12 +46,15 @@ public BaseFileSinkWriter(WriteStrategy writeStrategy, HadoopConf hadoopConf, Si
         if (!fileSinkStates.isEmpty()) {
             List<String> transactionIds = writeStrategy.getTransactionIdFromStates(fileSinkStates);
             transactionIds.forEach(writeStrategy::abortPrepare);
-            writeStrategy.beginTransaction(fileSinkStates.get(0).getCheckpointId());
+            writeStrategy.beginTransaction(fileSinkStates.get(0).getCheckpointId() + 1);
+        } else {
+            writeStrategy.beginTransaction(1L);
         }
     }
 
     public BaseFileSinkWriter(WriteStrategy writeStrategy, HadoopConf hadoopConf, SinkWriter.Context context, String jobId) {
         this(writeStrategy, hadoopConf, context, jobId, Collections.emptyList());
+        writeStrategy.beginTransaction(1L);
     }
 
     @Override

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/commit/FileSinkAggregatedCommitter.java

@@ -83,6 +83,7 @@ public FileAggregatedCommitInfo combine(List<FileCommitInfo> commitInfos) {
      */
     @Override
     public void abort(List<FileAggregatedCommitInfo> aggregatedCommitInfos) throws Exception {
+        log.info("rollback aggregate commit");
         if (aggregatedCommitInfos == null || aggregatedCommitInfos.size() == 0) {
             return;
         }

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/util/FileSystemUtils.java

@@ -76,12 +76,19 @@ public static void deleteFile(@NonNull String file) throws IOException {
      * @param rmWhenExist if this is true, we will delete the target file when it already exists
      * @throws IOException throw IOException
      */
-    public static void renameFile(@NonNull String oldName, @NonNull String newName, boolean rmWhenExist) throws IOException {
+    public static void renameFile(@NonNull String oldName, @NonNull String newName, boolean rmWhenExist)
+        throws IOException {
         FileSystem fileSystem = getFileSystem(newName);
         log.info("begin rename file oldName :[" + oldName + "] to newName :[" + newName + "]");
 
         Path oldPath = new Path(oldName);
         Path newPath = new Path(newName);
+
+        if (!fileExist(oldPath.toString())) {
+            log.warn("rename file :[" + oldPath + "] to [" + newPath + "] already finished in the last commit, skip");
+            return;
+        }
+
         if (rmWhenExist) {
             if (fileExist(newName) && fileExist(oldName)) {
                 fileSystem.delete(newPath, true);
@@ -119,6 +126,9 @@ public static boolean fileExist(@NonNull String filePath) throws IOException {
     public static List<Path> dirList(@NonNull String filePath) throws FileNotFoundException, IOException {
         FileSystem fileSystem = getFileSystem(filePath);
         List<Path> pathList = new ArrayList<>();
+        if (!fileExist(filePath)) {
+            return pathList;
+        }
         Path fileName = new Path(filePath);
         FileStatus[] status = fileSystem.listStatus(fileName);
         if (status != null && status.length > 0) {

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/AbstractWriteStrategy.java

@@ -70,7 +70,9 @@ public abstract class AbstractWriteStrategy implements WriteStrategy {
     protected Map<String, String> beingWrittenFile;
     private Map<String, List<String>> partitionDirAndValuesMap;
     protected SeaTunnelRowType seaTunnelRowType;
-    protected Long checkpointId = 1L;
+
+    // Checkpoint id from engine is start with 1
+    protected Long checkpointId = 0L;
 
     public AbstractWriteStrategy(TextFileSinkConfig textFileSinkConfig) {
         this.textFileSinkConfig = textFileSinkConfig;
@@ -88,7 +90,6 @@ public void init(HadoopConf conf, String jobId, int subTaskIndex) {
         this.jobId = jobId;
         this.subTaskIndex = subTaskIndex;
         FileSystemUtils.CONF = getConfiguration(hadoopConf);
-        this.beginTransaction(this.checkpointId);
     }
 
     /**
@@ -233,6 +234,7 @@ public void abortPrepare(String transactionId) {
      * @param checkpointId checkpoint id
      */
     public void beginTransaction(Long checkpointId) {
+        this.checkpointId = checkpointId;
         this.transactionId = "T" + Constant.TRANSACTION_ID_SPLIT + jobId + Constant.TRANSACTION_ID_SPLIT + subTaskIndex + Constant.TRANSACTION_ID_SPLIT + checkpointId;
         this.transactionDirectory = getTransactionDir(this.transactionId);
         this.needMoveFiles = new HashMap<>();
@@ -265,8 +267,7 @@ public List<String> getTransactionIdFromStates(List<FileSinkState> fileStates) {
     @Override
     public List<FileSinkState> snapshotState(long checkpointId) {
         ArrayList<FileSinkState> fileState = Lists.newArrayList(new FileSinkState(this.transactionId, this.checkpointId));
-        this.checkpointId = checkpointId;
-        this.beginTransaction(checkpointId);
+        this.beginTransaction(checkpointId + 1);
         return fileState;
     }
 
@@ -303,4 +304,8 @@ public String getTargetLocation(@NonNull String seaTunnelFilePath) {
                 Matcher.quoteReplacement(textFileSinkConfig.getPath()));
         return tmpPath.replaceAll(Constant.NON_PARTITION + Matcher.quoteReplacement(File.separator), "");
     }
+
+    public long getCheckpointId() {
+        return this.checkpointId;
+    }
 }

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/WriteStrategy.java

@@ -72,4 +72,6 @@ public interface WriteStrategy extends Transaction, Serializable {
      * when a transaction is triggered, release resources
      */
     void finishAndCloseFile();
+
+    long getCheckpointId();
 }