[flink] not getting commitUser from state

Author: LsomeYeah

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcAppendTableWriteOperator.java

@@ -53,6 +53,13 @@ protected StoreSinkWriteState createState(
         return new NoopStoreSinkWriteState(subtaskId);
     }
 
+    @Override
+    protected String getCommitUser(StateInitializationContext context) throws Exception {
+        // No conflicts will occur in append only unaware bucket writer, so
+        // commitUser does not matter.
+        return commitUser == null ? initialCommitUser : commitUser;
+    }
+
     @Override
     public void processElement(StreamRecord<CdcRecord> element) throws Exception {
         // only accepts INSERT record

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkWriteSink.java

@@ -93,6 +93,14 @@ protected StoreSinkWriteState createState(
                         // is needed.
                         return new NoopStoreSinkWriteState(subtaskId);
                     }
+
+                    @Override
+                    protected String getCommitUser(StateInitializationContext context)
+                            throws Exception {
+                        // No conflicts will occur in append only unaware bucket writer, so
+                        // commitUser does not matter.
+                        return commitUser == null ? initialCommitUser : commitUser;
+                    }
                 };
             }
         };