Improve deduplicateBatch logic (#168)

* Improve deduplicateBatch logic

* Improve deduplicateBatch logic

* Improve deduplicateBatch logic

.github/workflows/build.yml

@@ -17,6 +17,9 @@ on:
       - '.idea/**'
       - '.run/**'
 
+env:
+  SPARK_LOCAL_IP: 127.0.0.1
+
 jobs:
   build:
 

debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/tableoperator/IcebergTableOperator.java

@@ -58,24 +58,20 @@ public class IcebergTableOperator {
 
   private List<IcebergChangeEvent> deduplicateBatch(List<IcebergChangeEvent> events) {
 
-    ConcurrentHashMap<JsonNode, IcebergChangeEvent> icebergRecordsmap = new ConcurrentHashMap<>();
-
-    for (IcebergChangeEvent e : events) {
-
-      // deduplicate using key(PK) @TODO improve using map.merge
-      if (icebergRecordsmap.containsKey(e.key())) {
-
-        // replace it if it's new
-        if (this.compareByTsThenOp(icebergRecordsmap.get(e.key()).value(), e.value()) <= 0) {
-          icebergRecordsmap.put(e.key(), e);
-        }
-
-      } else {
-        icebergRecordsmap.put(e.key(), e);
-      }
-
-    }
-    return new ArrayList<>(icebergRecordsmap.values());
+    ConcurrentHashMap<JsonNode, IcebergChangeEvent> deduplicatedEvents = new ConcurrentHashMap<>();
+
+    events.forEach(e ->
+        // deduplicate using key(PK)
+        deduplicatedEvents.merge(e.key(), e, (oldValue, newValue) -> {
+          if (this.compareByTsThenOp(oldValue.value(), newValue.value()) <= 0) {
+            return newValue;
+          } else {
+            return oldValue;
+          }
+        })
+    );
+
+    return new ArrayList<>(deduplicatedEvents.values());
   }
 
   /**