[Fix][Kafka-Sink] fix kafka sink factory option rule (apache#6657)

lostinwind · Apr 15, 2024 · 9b4fe62 · 9b4fe62
1 parent bdece86
commit 9b4fe62
Showing 1 changed file with 2 additions and 10 deletions.
diff --git a/.../src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkFactory.java b/.../src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkFactory.java
@@ -39,17 +39,9 @@ public String factoryIdentifier() {
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder()
-                .required(Config.FORMAT, Config.BOOTSTRAP_SERVERS)
-                .conditional(
-                        Config.FORMAT,
-                        Arrays.asList(
-                                MessageFormat.JSON,
-                                MessageFormat.CANAL_JSON,
-                                MessageFormat.TEXT,
-                                MessageFormat.OGG_JSON,
-                                MessageFormat.AVRO),
-                        Config.TOPIC)
+                .required(Config.TOPIC, Config.BOOTSTRAP_SERVERS)
                 .optional(
+                        Config.FORMAT,
                         Config.KAFKA_CONFIG,
                         Config.ASSIGN_PARTITIONS,
                         Config.TRANSACTION_PREFIX,