Re: [PR] [cdc] Minor improve CdcUnawareBucketSink and fix unstable test [incubator-paimon]

["JingsongLi (via GitHub)" <gi...@apache.org>]

JingsongLi commented on code in PR #2666:
URL: https://github.com/apache/incubator-paimon/pull/2666#discussion_r1446864894


##########
paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketWriteOperator.java:
##########
@@ -21,77 +21,24 @@
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.flink.sink.PrepareCommitOperator;
 import org.apache.paimon.flink.sink.StoreSinkWrite;
-import org.apache.paimon.flink.sink.TableWriteOperator;
-import org.apache.paimon.options.ConfigOption;
-import org.apache.paimon.options.ConfigOptions;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.types.RowKind;
 
-import org.apache.flink.runtime.state.StateInitializationContext;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-
-import java.io.IOException;
-import java.time.Duration;
-import java.util.Optional;
-
-import static org.apache.paimon.flink.sink.cdc.CdcRecordUtils.toGenericRow;
-
-/** A {@link PrepareCommitOperator} to write {@link CdcRecord} with bucket. */
-public class CdcUnawareBucketWriteOperator extends TableWriteOperator<CdcRecord> {
-
-    private static final long serialVersionUID = 1L;
-
-    public static final ConfigOption<Duration> RETRY_SLEEP_TIME =
-            ConfigOptions.key("cdc.retry-sleep-time")
-                    .durationType()
-                    .defaultValue(Duration.ofMillis(500));
-
-    protected final long retrySleepMillis;
+/** A {@link PrepareCommitOperator} to write {@link CdcRecord} to unaware-bucket mode table. */
+public class CdcUnawareBucketWriteOperator extends CdcRecordStoreWriteOperator {
 
     public CdcUnawareBucketWriteOperator(
             FileStoreTable table,
             StoreSinkWrite.Provider storeSinkWriteProvider,
             String initialCommitUser) {
         super(table, storeSinkWriteProvider, initialCommitUser);
-        this.retrySleepMillis =
-                table.coreOptions().toConfiguration().get(RETRY_SLEEP_TIME).toMillis();
-    }
-
-    @Override
-    public void initializeState(StateInitializationContext context) throws Exception {
-        table = table.copyWithLatestSchema();
-        super.initializeState(context);
     }
 
     @Override
-    protected boolean containLogSystem() {
-        return false;
-    }
-
-    @Override
-    public void processElement(StreamRecord<CdcRecord> element) throws Exception {

Review Comment:
   if is not insert, just return



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org