fix

Author: JingsongLi

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java

@@ -69,6 +69,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
+import static org.apache.paimon.CoreOptions.CHANGELOG_PRODUCER;
 import static org.apache.paimon.CoreOptions.MergeEngine.FIRST_ROW;
 import static org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_ASYNC;
 import static org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_ASYNC_THREAD_NUMBER;
@@ -149,6 +150,10 @@ public ChangelogMode getChangelogMode() {
             return ChangelogMode.all();
         }
 
+        if (options.get(CHANGELOG_PRODUCER) != CoreOptions.ChangelogProducer.NONE) {
+            return ChangelogMode.all();
+        }
+
         return ChangelogMode.upsert();
     }
 

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogModeTest.java

@@ -18,7 +18,6 @@
 
 package org.apache.paimon.flink;
 
-import org.apache.paimon.CoreOptions;
 import org.apache.paimon.flink.sink.FlinkTableSink;
 import org.apache.paimon.flink.source.DataTableSource;
 import org.apache.paimon.fs.Path;
@@ -40,6 +39,9 @@
 
 import java.util.Collections;
 
+import static org.apache.paimon.CoreOptions.CHANGELOG_PRODUCER;
+import static org.apache.paimon.CoreOptions.ChangelogProducer.INPUT;
+import static org.apache.paimon.CoreOptions.ChangelogProducer.LOOKUP;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** Test for changelog mode with flink source and sink. */
@@ -90,7 +92,14 @@ public void testDefault() throws Exception {
     @Test
     public void testInputChangelogProducer() throws Exception {
         Options options = new Options();
-        options.set(CoreOptions.CHANGELOG_PRODUCER, CoreOptions.ChangelogProducer.INPUT);
+        options.set(CHANGELOG_PRODUCER, INPUT);
         test(options, ChangelogMode.all(), ChangelogMode.all());
     }
+
+    @Test
+    public void testLookupChangelogProducer() throws Exception {
+        Options options = new Options();
+        options.set(CHANGELOG_PRODUCER, LOOKUP);
+        test(options, ChangelogMode.all(), ChangelogMode.upsert());
+    }
 }