[cdc] Kafka database sync new table shouldn't require primary keys

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java

@@ -148,6 +148,7 @@ protected EventParser.Factory<RichCdcMultiplexRecord> buildEventParserFactory()
                         allowUpperCase,
                         partitionKeys,
                         primaryKeys,
+                        requirePrimaryKeys(),
                         partitionKeyMultiple,
                         metadataConverters);
         Pattern includingPattern = Pattern.compile(includingTables);
@@ -170,6 +171,8 @@ protected EventParser.Factory<RichCdcMultiplexRecord> buildEventParserFactory()
                         createdTables);
     }
 
+    protected abstract boolean requirePrimaryKeys();
+
     @Override
     protected void buildSink(
             DataStream<RichCdcMultiplexRecord> input,

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java

@@ -40,4 +40,9 @@ public KafkaSyncDatabaseAction(
     protected CdcTimestampExtractor createCdcTimestampExtractor() {
         return new MessageQueueCdcTimestampExtractor();
     }
+
+    @Override
+    protected boolean requirePrimaryKeys() {
+        return false;
+    }
 }

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java

@@ -73,4 +73,9 @@ protected MongoDBSource<CdcSourceRecord> buildSource() {
                         includingTables,
                         Collections.emptyList()));
     }
+
+    @Override
+    protected boolean requirePrimaryKeys() {
+        return true;
+    }
 }

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java

@@ -259,4 +259,9 @@ public List<Identifier> monitoredTables() {
     public List<Identifier> excludedTables() {
         return excludedTables;
     }
+
+    @Override
+    protected boolean requirePrimaryKeys() {
+        return true;
+    }
 }

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseAction.java

@@ -40,4 +40,9 @@ public PulsarSyncDatabaseAction(
     protected CdcTimestampExtractor createCdcTimestampExtractor() {
         return new MessageQueueCdcTimestampExtractor();
     }
+
+    @Override
+    protected boolean requirePrimaryKeys() {
+        return false;
+    }
 }

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

@@ -24,7 +24,6 @@
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -38,21 +37,24 @@ public class NewTableSchemaBuilder implements Serializable {
     private final boolean caseSensitive;
     private final List<String> partitionKeys;
     private final List<String> primaryKeys;
+    private final boolean requirePrimaryKeys;
     private final CdcMetadataConverter[] metadataConverters;
-    protected Map<String, List<String>> partitionKeyMultiple = new HashMap<>();
+    private final Map<String, List<String>> partitionKeyMultiple;
 
     public NewTableSchemaBuilder(
             Map<String, String> tableConfig,
             boolean caseSensitive,
             List<String> partitionKeys,
             List<String> primaryKeys,
+            boolean requirePrimaryKeys,
             Map<String, List<String>> partitionKeyMultiple,
             CdcMetadataConverter[] metadataConverters) {
         this.tableConfig = tableConfig;
         this.caseSensitive = caseSensitive;
         this.metadataConverters = metadataConverters;
         this.partitionKeys = partitionKeys;
         this.primaryKeys = primaryKeys;
+        this.requirePrimaryKeys = requirePrimaryKeys;
         this.partitionKeyMultiple = partitionKeyMultiple;
     }
 
@@ -84,6 +86,6 @@ public Optional<Schema> build(RichCdcMultiplexRecord record) {
                         metadataConverters,
                         caseSensitive,
                         false,
-                        true));
+                        requirePrimaryKeys));
     }
 }