[HUDI-1778] Add setter to CompactionPlanEvent and CompactionCommitEvent to have better SE/DE performance for Flink (#2789)

hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitEvent.java

@@ -32,15 +32,18 @@ public class CompactionCommitEvent implements Serializable {
   /**
    * The compaction commit instant time.
    */
-  private final String instant;
+  private String instant;
   /**
    * The write statuses.
    */
-  private final List<WriteStatus> writeStatuses;
+  private List<WriteStatus> writeStatuses;
   /**
    * The compaction task identifier.
    */
-  private final int taskID;
+  private int taskID;
+
+  public CompactionCommitEvent() {
+  }
 
   public CompactionCommitEvent(String instant, List<WriteStatus> writeStatuses, int taskID) {
     this.instant = instant;
@@ -48,6 +51,18 @@ public class CompactionCommitEvent implements Serializable {
     this.taskID = taskID;
   }
 
+  public void setInstant(String instant) {
+    this.instant = instant;
+  }
+
+  public void setWriteStatuses(List<WriteStatus> writeStatuses) {
+    this.writeStatuses = writeStatuses;
+  }
+
+  public void setTaskID(int taskID) {
+    this.taskID = taskID;
+  }
+
   public String getInstant() {
     return instant;
   }

hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanEvent.java

@@ -28,15 +28,26 @@ import java.io.Serializable;
 public class CompactionPlanEvent implements Serializable {
   private static final long serialVersionUID = 1L;
 
-  private final String compactionInstantTime;
+  private String compactionInstantTime;
 
-  private final CompactionOperation operation;
+  private CompactionOperation operation;
+
+  public CompactionPlanEvent() {
+  }
 
   public CompactionPlanEvent(String instantTime, CompactionOperation operation) {
     this.compactionInstantTime = instantTime;
     this.operation = operation;
   }
 
+  public void setCompactionInstantTime(String compactionInstantTime) {
+    this.compactionInstantTime = compactionInstantTime;
+  }
+
+  public void setOperation(CompactionOperation operation) {
+    this.operation = operation;
+  }
+
   public String getCompactionInstantTime() {
     return compactionInstantTime;
   }