diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
index 11bf054e96034..1c86f139f791c 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
@@ -351,7 +351,14 @@ private void validateOptions(Map<String, String> options) {
                         .collect(Collectors.toSet());
 
         Set<String> additionalValidOptions =
-                new HashSet<>(Arrays.asList("is_streaming", "connector", "log.system"));
+                new HashSet<>(
+                        Arrays.asList(
+                                "is_streaming",
+                                "connector",
+                                "log.system",
+                                "kafka.bootstrap.servers",
+                                "kafka.topic",
+                                "kafka.transaction.timeout.ms"));
 
         options.keySet()
                 .forEach(