Use RecordWrapper

memiiso · Oct 5, 2024 · 58fa00c · 58fa00c
1 parent c994488
commit 58fa00c
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/...erg-sink/src/main/java/io/debezium/server/iceberg/tableoperator/IcebergTableOperator.java b/...erg-sink/src/main/java/io/debezium/server/iceberg/tableoperator/IcebergTableOperator.java
@@ -178,7 +178,7 @@ private void addToTablePerSchema(Table icebergTable, List<RecordConverter> event
     BaseTaskWriter<Record> writer = writerFactory.create(icebergTable);
     try (writer) {
       for (RecordConverter e : events) {
-        final RecordWrapper record = e.convert(tableSchema, cdcOpField);
+        final RecordWrapper record = (upsert && !tableSchema.identifierFieldIds().isEmpty()) ? e.convert(tableSchema, cdcOpField) : e.convert(tableSchema);
         writer.write(record);
       }