[HUDI-4536] ClusteringOperator causes the NullPointerException when writing with BulkInsertWriterHelper in clustering

Author: SteNicholas

hudi-common/src/main/java/org/apache/hudi/common/fs/HoodieWrapperFileSystem.java

@@ -1015,6 +1015,10 @@ public void createImmutableFileInPath(Path fullPath, Option<byte[]> content)
     boolean needTempFile = needCreateTempFile();
 
     try {
+      if (exists(fullPath)) {
+        return;
+      }
+
       if (!content.isPresent()) {
         fsout = fileSystem.create(fullPath, false);
       }

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java

@@ -233,6 +233,7 @@ private void doClustering(String instantTime, ClusteringPlanEvent event) throws
 
     List<WriteStatus> writeStatuses = this.writerHelper.getWriteStatuses(this.taskID);
     collector.collect(new ClusteringCommitEvent(instantTime, writeStatuses, this.taskID));
+    this.writerHelper = null;
   }
 
   private void initWriterHelper(String clusteringInstantTime) {