fix

Author: JingsongLi

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

@@ -100,6 +100,14 @@ public void testInputChangelogProducer() throws Exception {
     public void testLookupChangelogProducer() throws Exception {
         Options options = new Options();
         options.set(CHANGELOG_PRODUCER, LOOKUP);
-        test(options, ChangelogMode.all(), ChangelogMode.upsert());
+
+        // Compatible to Flink 2.0 (keyOnlyDeletes)
+        ChangelogMode.Builder upsertBuilder = ChangelogMode.newBuilder();
+        for (RowKind kind : ChangelogMode.all().getContainedKinds()) {
+            if (kind != RowKind.UPDATE_BEFORE) {
+                upsertBuilder.addContainedKind(kind);
+            }
+        }
+        test(options, ChangelogMode.all(), upsertBuilder.build());
     }
 }