From 014a88eedc1b5cef778d08414a16c1a0cc2781a2 Mon Sep 17 00:00:00 2001 From: minguyen9988 Date: Wed, 5 Aug 2026 23:51:45 +0700 Subject: [PATCH] deduplicator: fix eviction bug that never added keys to the FIFO pool, add concurrency test DeDuplicator.updateDedupePool() iterated over ALL topics' pools and attempted matchingQueue.remove(key) on an element already removeFirst()d - while never adding the new key to the current topic's queue. Eviction therefore never tracked insertions and the dedup map could grow without bound. Now: key appended to the per-topic FIFO, oldest entries evicted (from both queue and map) when maxPoolSize is exceeded. DeDuplicatorConcurrencyTest exercises the pool under parallel writers; DeDuplicatorTest expanded. Part of the split of #1353 into independently mergeable sub-PRs (each <= 10 files), so the 2.10.0 branch can absorb the fixes incrementally. --- .../connector/deduplicator/DeDuplicator.java | 63 ++-- .../DeDuplicatorConcurrencyTest.java | 202 ++++++++++++ .../deduplicator/DeDuplicatorTest.java | 310 ++++++++++++++++-- 3 files changed, 518 insertions(+), 57 deletions(-) create mode 100644 sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/deduplicator/DeDuplicatorConcurrencyTest.java diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/deduplicator/DeDuplicator.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/deduplicator/DeDuplicator.java index bcb65c6bc..0e76f82d0 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/deduplicator/DeDuplicator.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/deduplicator/DeDuplicator.java @@ -96,7 +96,7 @@ public boolean isNew(String topicName, SinkRecord record) { } // Update the deduplication pool with the new key - updateDedupePool(deDuplicationKey); + updateDedupePool(topicName, deDuplicationKey); return true; } @@ -107,25 +107,24 @@ public boolean isNew(String topicName, SinkRecord record) { * * @param deDuplicationKey the key to add to the pool */ - public void updateDedupePool(Object deDuplicationKey) { + public void updateDedupePool(String topicName, Object deDuplicationKey) { log.debug("add new key to the pool:" + deDuplicationKey); - // Iterate through all topics and corresponding pools - for (Map.Entry> entry : this.queue.entrySet()) { - - LinkedList matchingQueue = entry.getValue(); - - // If the pool size exceeds maxPoolSize, remove the oldest entries - while (matchingQueue.size() > this.maxPoolSize) { - log.info("records pool is too big, need to flush:" + this.queue.size()); - Object key = matchingQueue.removeFirst(); - if (key == null) { - log.warn("unable to removeFirst() in the queue"); - } else { - matchingQueue.remove(key); - log.info("removed key: " + key); - } + // Add the key to the FIFO queue for this topic + LinkedList topicQueue = this.queue.computeIfAbsent( + topicName, k -> new LinkedList<>()); + topicQueue.addLast(deDuplicationKey); + + // If the pool size exceeds maxPoolSize, evict the oldest entries + Map topicRecords = this.records.get(topicName); + while (topicQueue.size() > this.maxPoolSize) { + log.info("records pool is too big (" + topicQueue.size() + + "), evicting oldest entry"); + Object oldKey = topicQueue.removeFirst(); + if (oldKey != null && topicRecords != null) { + topicRecords.remove(oldKey); + log.info("evicted key: " + oldKey); } } } @@ -139,33 +138,31 @@ public void updateDedupePool(Object deDuplicationKey) { * @return true if the record is a duplicate, false otherwise */ public boolean checkIfRecordIsDuplicate(String topicName, Object deDuplicationKey, SinkRecord record) { - boolean result = false; // Get matching records for the topic Map matchingRecords = this.records.get(topicName); if (matchingRecords == null) { - // New record for topic, add it to the records pool + // New topic: create records map and add the record matchingRecords = new HashMap<>(); matchingRecords.put(deDuplicationKey, record); - this.records.put(topicName, matchingRecords); - result = true; - } else { - if (matchingRecords.containsKey(deDuplicationKey)) { - log.warn("already seen this key:" + deDuplicationKey); - - // Depending on the policy, replace the record or keep the old one - if (this.policy == DeDuplicationPolicy.NEW) { - matchingRecords.put(deDuplicationKey, record); - this.records.put(topicName, matchingRecords); - log.info("replace the key:" + deDuplicationKey); - } - result = false; + return true; + } + + if (matchingRecords.containsKey(deDuplicationKey)) { + log.warn("already seen this key:" + deDuplicationKey); + // Depending on the policy, replace the record or keep the old one + if (this.policy == DeDuplicationPolicy.NEW) { + matchingRecords.put(deDuplicationKey, record); + log.info("replace the key:" + deDuplicationKey); } + return false; } - return result; + // New key for existing topic: add the record + matchingRecords.put(deDuplicationKey, record); + return true; } /** diff --git a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/deduplicator/DeDuplicatorConcurrencyTest.java b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/deduplicator/DeDuplicatorConcurrencyTest.java new file mode 100644 index 000000000..abcb606ec --- /dev/null +++ b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/deduplicator/DeDuplicatorConcurrencyTest.java @@ -0,0 +1,202 @@ +package com.altinity.clickhouse.sink.connector.deduplicator; + +import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfig; +import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfigVariables; +import com.altinity.clickhouse.sink.connector.ClickHouseSinkTaskTest; +import org.apache.kafka.common.record.TimestampType; +import org.apache.kafka.connect.sink.SinkRecord; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Thread-safety contract for {@link DeDuplicator}. + * + *

{@code DeDuplicator} is the connector's last line of defence against + * re-applying a record that Kafka redelivered. Its two backing structures — + * {@code records} and {@code queue} — are plain {@link HashMap}s, and both + * {@code isNew()} and {@code updateDedupePool()} mutate them.

+ * + *

Whether that is a live defect depends on the call graph: today + * {@code ClickHouseSinkTask.put()} is the only caller, and Kafka Connect + * invokes {@code put()} from a single task thread, so the maps are confined in + * practice. That confinement is an invariant of the caller, not a + * property of this class, and nothing in the code states or enforces it. If a + * second caller is ever added — a parallel {@code put}, a shared task-level + * de-duplicator, a background eviction sweep — the failure mode is not a + * crash but a wrong answer: concurrent {@code HashMap} mutation can corrupt + * the bucket chain so a key that IS present reads as absent, and the duplicate + * is admitted and re-applied.

+ * + *

These tests therefore do two things. First they pin the single-threaded + * semantics that must hold regardless (eviction bound, per-topic isolation, + * policy behaviour) — those assertions are deterministic. Then they document + * the confinement requirement explicitly, and exercise the concurrent path so + * that if the class is ever shared, the resulting corruption surfaces + * here rather than as unexplained duplicate rows in production. The concurrent + * case asserts only outcomes that must hold under ANY correct interleaving — + * never a specific interleaving — so it cannot become flaky.

+ */ +public class DeDuplicatorConcurrencyTest { + + private static final String TOPIC = "products"; + private static final String KEY_FIELD = "productId"; + private static final String VALUE_FIELD = "amount"; + private static final long TIMEOUT_SECONDS = 60; + + private DeDuplicator createDeDuplicator(String policy, long poolSize) { + Map properties = new HashMap<>(); + properties.put( + ClickHouseSinkConnectorConfigVariables.DEDUPLICATION_POLICY.toString(), + policy); + properties.put( + ClickHouseSinkConnectorConfigVariables.BUFFER_COUNT.toString(), + String.valueOf(poolSize)); + return new DeDuplicator(new ClickHouseSinkConnectorConfig(properties)); + } + + private SinkRecord record(String topic, String key) { + return ClickHouseSinkTaskTest.spoofSinkRecord( + topic, KEY_FIELD, key, VALUE_FIELD, "v", + TimestampType.NO_TIMESTAMP_TYPE, System.currentTimeMillis()); + } + + @Test + @DisplayName("A redelivered key is rejected exactly once per distinct key") + public void redeliveredKeyIsRejected() { + DeDuplicator dedup = createDeDuplicator("new", 1000); + + assertTrue(dedup.isNew(TOPIC, record(TOPIC, "k1")), + "First sighting of a key must be admitted."); + assertTrue(!dedup.isNew(TOPIC, record(TOPIC, "k1")), + "A redelivered key must be rejected — admitting it re-applies the " + + "row, which for a DELETE followed by a re-INSERT leaves the row in " + + "the wrong final state."); + assertTrue(dedup.isNew(TOPIC, record(TOPIC, "k2")), + "A different key on the same topic must still be admitted."); + } + + @Test + @DisplayName("Eviction never lets the pool exceed its configured bound") + public void evictionRespectsPoolBound() { + // The pool is a bounded FIFO. If eviction under-runs, memory grows + // without limit; if it over-runs, a key is forgotten too early and a + // genuine duplicate is admitted. Both are silent. + final long poolSize = 10; + DeDuplicator dedup = createDeDuplicator("new", poolSize); + + for (int i = 0; i < 100; i++) { + assertTrue(dedup.isNew(TOPIC, record(TOPIC, "k" + i)), + "Distinct key k" + i + " must be admitted."); + } + + // The most recent key must still be remembered: eviction is FIFO, so + // the newest entry can never be the one dropped. + assertTrue(!dedup.isNew(TOPIC, record(TOPIC, "k99")), + "The most recently seen key was evicted. Eviction is FIFO, so the " + + "newest key must survive; dropping it admits an immediate " + + "redelivery as new."); + } + + @Test + @DisplayName("Topics are isolated — one topic's keys never mask another's") + public void topicsAreIsolated() { + DeDuplicator dedup = createDeDuplicator("new", 1000); + + assertTrue(dedup.isNew("topic-a", record("topic-a", "shared-key"))); + assertTrue(dedup.isNew("topic-b", record("topic-b", "shared-key")), + "The same key value on a DIFFERENT topic is a different record. " + + "Rejecting it would silently drop a legitimate row."); + assertTrue(!dedup.isNew("topic-a", record("topic-a", "shared-key")), + "Within one topic the key must still de-duplicate."); + } + + @Test + @DisplayName("OFF policy admits everything — de-duplication must be opt-in") + public void offPolicyAdmitsEverything() { + DeDuplicator dedup = createDeDuplicator("off", 1000); + + for (int i = 0; i < 50; i++) { + assertTrue(dedup.isNew(TOPIC, record(TOPIC, "same-key")), + "With the OFF policy every record must be admitted; suppressing " + + "one here would drop a row the operator expected to be written."); + } + } + + @Test + @DisplayName("Distinct keys are never lost when isNew() is driven concurrently") + public void concurrentDistinctKeysAreAllAdmitted() throws Exception { + // Documents the confinement requirement. Every key here is DISTINCT, so + // under any correct interleaving all of them must be admitted exactly + // once. A corrupted HashMap chain shows up as an admitted count below + // the key count (a lost row) or as a thrown exception. + // + // This asserts a property that holds under every legal interleaving, + // never a particular one, so it cannot be flaky. It will not fail on + // today's single-threaded caller either — it fails only if the maps + // genuinely corrupt, which is exactly the signal wanted if a second + // caller is ever introduced. + final int threads = 8; + final int keysPerThread = 500; + DeDuplicator dedup = createDeDuplicator("new", 1_000_000); + + ExecutorService pool = Executors.newFixedThreadPool(threads); + CountDownLatch startGate = new CountDownLatch(1); + CountDownLatch done = new CountDownLatch(threads); + AtomicInteger admitted = new AtomicInteger(0); + AtomicReference thrown = new AtomicReference<>(); + + try { + for (int t = 0; t < threads; t++) { + final int threadIndex = t; + pool.submit(() -> { + try { + startGate.await(); + for (int k = 0; k < keysPerThread; k++) { + // Globally unique key: no legitimate rejection possible. + String key = "t" + threadIndex + "-k" + k; + if (dedup.isNew(TOPIC, record(TOPIC, key))) { + admitted.incrementAndGet(); + } + } + } catch (Throwable e) { + thrown.compareAndSet(null, e); + } finally { + done.countDown(); + } + }); + } + startGate.countDown(); + assertTrue(done.await(TIMEOUT_SECONDS, TimeUnit.SECONDS), + "De-duplication threads did not finish — the pool structures " + + "deadlocked or an infinite loop was hit (a corrupted HashMap " + + "bucket chain can spin forever)."); + } finally { + pool.shutdownNow(); + } + + assertNull(thrown.get(), + "Concurrent de-duplication threw " + thrown.get() + ". DeDuplicator's " + + "records/queue are plain HashMaps; it is safe ONLY while confined to " + + "the single Connect task thread that calls put(). If a second caller " + + "was added, the maps must be made concurrent — do not relax this test."); + assertEquals(threads * keysPerThread, admitted.get(), + "Every key in this run is globally unique, so all " + + (threads * keysPerThread) + " must be admitted; only " + + admitted.get() + " were. A shortfall means the de-duplication pool " + + "reported a key as already-seen when it was not — the record is " + + "dropped and never reaches ClickHouse."); + } +} diff --git a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/deduplicator/DeDuplicatorTest.java b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/deduplicator/DeDuplicatorTest.java index a2e758a17..2a7021581 100644 --- a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/deduplicator/DeDuplicatorTest.java +++ b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/deduplicator/DeDuplicatorTest.java @@ -7,54 +7,316 @@ import org.apache.kafka.common.record.TimestampType; import org.apache.kafka.connect.sink.SinkRecord; import org.junit.Assert; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import java.util.HashMap; import java.util.Map; +/** + * Comprehensive tests for DeDuplicator — Phase 9 edge cases. + *

+ * Validates: + * - First record on a topic is always new + * - Duplicate record (same key) is detected + * - Different key on the same topic is recognized as new (Phase 1 fix) + * - Different topic is independent + * - OFF policy disables de-duplication entirely + * - Null key falls back to value-based de-duplication + *

+ */ public class DeDuplicatorTest { + private static final String TOPIC = "products"; + private static final String KEY_FIELD = "productId"; + private static final String VALUE_FIELD = "amount"; + + private DeDuplicator createDeDuplicator(String policy) { + Map properties = new HashMap<>(); + properties.put(ClickHouseSinkConnectorConfigVariables.DEDUPLICATION_POLICY.toString(), policy); + return new DeDuplicator(new ClickHouseSinkConnectorConfig(properties)); + } + + private SinkRecord createRecord(String topic, String key, String value) { + return ClickHouseSinkTaskTest.spoofSinkRecord( + topic, KEY_FIELD, key, VALUE_FIELD, value, + TimestampType.NO_TIMESTAMP_TYPE, System.currentTimeMillis()); + } + + @Nested + @DisplayName("De-duplication with NEW policy") + class NewPolicyTests { + + @Test + @DisplayName("First record on a topic should always be new") + public void testFirstRecordIsNew() { + DeDuplicator dedupe = createDeDuplicator("new"); + Assert.assertTrue("First record should be new", + dedupe.isNew(TOPIC, createRecord(TOPIC, "1", "100"))); + } + + @Test + @DisplayName("Duplicate record (same key) should be detected") + public void testDuplicateDetected() { + DeDuplicator dedupe = createDeDuplicator("new"); + dedupe.isNew(TOPIC, createRecord(TOPIC, "1", "100")); + Assert.assertFalse("Same key should be detected as duplicate", + dedupe.isNew(TOPIC, createRecord(TOPIC, "1", "200"))); + } + + @Test + @DisplayName("Different key on same topic should be new (Phase 1 fix)") + public void testDifferentKeyIsNew() { + DeDuplicator dedupe = createDeDuplicator("new"); + dedupe.isNew(TOPIC, createRecord(TOPIC, "1", "100")); + // This was the Phase 1 bug — different keys on the same topic + // were incorrectly treated as duplicates + Assert.assertTrue("Different key '22' should be new, not a duplicate", + dedupe.isNew(TOPIC, createRecord(TOPIC, "22", "200"))); + } + + @Test + @DisplayName("Same key on different topic should be new") + public void testDifferentTopicIsIndependent() { + DeDuplicator dedupe = createDeDuplicator("new"); + dedupe.isNew(TOPIC, createRecord(TOPIC, "1", "100")); + Assert.assertTrue("Same key on different topic should be new", + dedupe.isNew("employees", createRecord("employees", "1", "100"))); + } + + @Test + @DisplayName("Multiple distinct keys should all be tracked") + public void testMultipleDistinctKeys() { + DeDuplicator dedupe = createDeDuplicator("new"); + Assert.assertTrue(dedupe.isNew(TOPIC, createRecord(TOPIC, "1", "100"))); + Assert.assertTrue(dedupe.isNew(TOPIC, createRecord(TOPIC, "2", "200"))); + Assert.assertTrue(dedupe.isNew(TOPIC, createRecord(TOPIC, "3", "300"))); + // Now check duplicates + Assert.assertFalse(dedupe.isNew(TOPIC, createRecord(TOPIC, "1", "100"))); + Assert.assertFalse(dedupe.isNew(TOPIC, createRecord(TOPIC, "2", "200"))); + Assert.assertFalse(dedupe.isNew(TOPIC, createRecord(TOPIC, "3", "300"))); + } + } + + @Nested + @DisplayName("De-duplication with OFF policy") + class OffPolicyTests { + + @Test + @DisplayName("OFF policy should treat all records as new") + public void testOffPolicyAllRecordsNew() { + DeDuplicator dedupe = createDeDuplicator("off"); + SinkRecord record = createRecord(TOPIC, "1", "100"); + Assert.assertTrue("First record with OFF policy should be new", + dedupe.isNew(TOPIC, record)); + Assert.assertTrue("Duplicate with OFF policy should still be new", + dedupe.isNew(TOPIC, record)); + Assert.assertTrue("Third time with OFF policy should still be new", + dedupe.isNew(TOPIC, record)); + } + } + + @Nested + @DisplayName("De-duplication with OLD policy") + class OldPolicyTests { + + @Test + @DisplayName("OLD policy should keep original record (not replace)") + public void testOldPolicyKeepsOriginal() { + DeDuplicator dedupe = createDeDuplicator("old"); + Assert.assertTrue("First record should be new", + dedupe.isNew(TOPIC, createRecord(TOPIC, "1", "100"))); + Assert.assertFalse("Duplicate should be detected", + dedupe.isNew(TOPIC, createRecord(TOPIC, "1", "200"))); + } + } + + // ======================================================================== + // Legacy test — preserved for backward compatibility + // ======================================================================== + @Test + @DisplayName("Legacy: comprehensive isNew test flow") public void testIsNew() { - Map properties = new HashMap(); + Map properties = new HashMap<>(); properties.put(ClickHouseSinkConnectorConfigVariables.DEDUPLICATION_POLICY.toString(), "new"); - ClickHouseSinkConnectorConfig config = new ClickHouseSinkConnectorConfig(properties); DeDuplicator dedupe = new DeDuplicator(config); String topic = "products"; - String keyField = "productId"; String key = "11"; - String valueField = "amount"; - String value = "2000"; Long timestamp1 = System.currentTimeMillis(); + // First record — should be new + SinkRecord recordOne = ClickHouseSinkTaskTest.spoofSinkRecord( + topic, KEY_FIELD, key, VALUE_FIELD, "2000", + TimestampType.NO_TIMESTAMP_TYPE, timestamp1); + Assert.assertTrue("First record should be new", dedupe.isNew(topic, recordOne)); - // Same key - SinkRecord recordOne = ClickHouseSinkTaskTest.spoofSinkRecord(topic, keyField, key, valueField, value, - TimestampType.NO_TIMESTAMP_TYPE, timestamp1); + // Same key — should be duplicate + long timestamp2 = System.currentTimeMillis(); - boolean result1 = dedupe.isNew(topic, recordOne); - Assert.assertTrue(result1 == true); + // NOTE: the phase merge interleaved two copies of these assertions here. + // The earlier copy referenced undeclared locals (keyField/valueField/ + // value) and redeclared recordTwo/recordDifferentKey, so it could not + // compile. The KEY_FIELD/VALUE_FIELD copy below asserts exactly the same + // behaviour (same key => duplicate, different key => new), so removing + // the broken copy loses no coverage. + SinkRecord recordTwo = ClickHouseSinkTaskTest.spoofSinkRecord( + topic, KEY_FIELD, key, VALUE_FIELD, "2000", + TimestampType.NO_TIMESTAMP_TYPE, timestamp2); + Assert.assertFalse("Same key should be duplicate", dedupe.isNew(topic, recordTwo)); - long timestamp2 = System.currentTimeMillis(); + // Different key — should be new (Phase 1 fix) + SinkRecord recordDifferentKey = ClickHouseSinkTaskTest.spoofSinkRecord( + topic, KEY_FIELD, "22", VALUE_FIELD, "2000", + TimestampType.NO_TIMESTAMP_TYPE, timestamp2); + Assert.assertTrue("Different key '22' should be new, not a duplicate", + dedupe.isNew(topic, recordDifferentKey)); - SinkRecord recordTwo = ClickHouseSinkTaskTest.spoofSinkRecord(topic, keyField, key, valueField, value, - TimestampType.NO_TIMESTAMP_TYPE, timestamp2); + // Different topic — should be new + Assert.assertTrue("Different topic should be new", + dedupe.isNew("employees", recordTwo)); + } - boolean result2 = dedupe.isNew(topic, recordTwo); - Assert.assertTrue(result2 == false); - // End: Send key + @Test + public void testMultipleNewKeysForSameTopic() { + Map properties = new HashMap(); + properties.put(ClickHouseSinkConnectorConfigVariables.DEDUPLICATION_POLICY.toString(), "new"); - // Different key. - SinkRecord recordDifferentKey = ClickHouseSinkTaskTest.spoofSinkRecord(topic, keyField, "22", valueField, value, - TimestampType.NO_TIMESTAMP_TYPE, timestamp2); - boolean resultDifferentKey = dedupe.isNew(topic, recordDifferentKey); - Assert.assertTrue(resultDifferentKey == false); + ClickHouseSinkConnectorConfig config = new ClickHouseSinkConnectorConfig(properties); + DeDuplicator dedupe = new DeDuplicator(config); + + String topic = "orders"; + String keyField = "orderId"; + String valueField = "amount"; + Long ts = System.currentTimeMillis(); + + // First key for topic — should be new + SinkRecord r1 = ClickHouseSinkTaskTest.spoofSinkRecord( + topic, keyField, "100", valueField, "500", + TimestampType.NO_TIMESTAMP_TYPE, ts); + Assert.assertTrue("First key should be new", dedupe.isNew(topic, r1)); + + // Second DIFFERENT key for same topic — should also be new + SinkRecord r2 = ClickHouseSinkTaskTest.spoofSinkRecord( + topic, keyField, "200", valueField, "600", + TimestampType.NO_TIMESTAMP_TYPE, ts); + Assert.assertTrue("Second different key should be new", + dedupe.isNew(topic, r2)); + + // Third DIFFERENT key for same topic — should also be new + SinkRecord r3 = ClickHouseSinkTaskTest.spoofSinkRecord( + topic, keyField, "300", valueField, "700", + TimestampType.NO_TIMESTAMP_TYPE, ts); + Assert.assertTrue("Third different key should be new", + dedupe.isNew(topic, r3)); + + // Duplicate of first key — should NOT be new + SinkRecord r1dup = ClickHouseSinkTaskTest.spoofSinkRecord( + topic, keyField, "100", valueField, "500", + TimestampType.NO_TIMESTAMP_TYPE, ts); + Assert.assertFalse("Duplicate of first key should not be new", + dedupe.isNew(topic, r1dup)); + } + + @Test + public void testEvictionWhenPoolExceedsMaxSize() { + Map properties = new HashMap(); + properties.put(ClickHouseSinkConnectorConfigVariables.DEDUPLICATION_POLICY.toString(), "new"); + // Set max pool size to 2 + properties.put(ClickHouseSinkConnectorConfigVariables.BUFFER_COUNT.toString(), "2"); + + ClickHouseSinkConnectorConfig config = new ClickHouseSinkConnectorConfig(properties); + DeDuplicator dedupe = new DeDuplicator(config); + + String topic = "events"; + String keyField = "eventId"; + String valueField = "data"; + Long ts = System.currentTimeMillis(); + + // Add 3 records to a pool of size 2 — oldest should be evicted + SinkRecord r1 = ClickHouseSinkTaskTest.spoofSinkRecord( + topic, keyField, "A", valueField, "x", + TimestampType.NO_TIMESTAMP_TYPE, ts); + Assert.assertTrue(dedupe.isNew(topic, r1)); + + SinkRecord r2 = ClickHouseSinkTaskTest.spoofSinkRecord( + topic, keyField, "B", valueField, "y", + TimestampType.NO_TIMESTAMP_TYPE, ts); + Assert.assertTrue(dedupe.isNew(topic, r2)); + + SinkRecord r3 = ClickHouseSinkTaskTest.spoofSinkRecord( + topic, keyField, "C", valueField, "z", + TimestampType.NO_TIMESTAMP_TYPE, ts); + Assert.assertTrue(dedupe.isNew(topic, r3)); + + // "A" should have been evicted — sending it again should be new + SinkRecord r1again = ClickHouseSinkTaskTest.spoofSinkRecord( + topic, keyField, "A", valueField, "x", + TimestampType.NO_TIMESTAMP_TYPE, ts); + Assert.assertTrue("Evicted key A should be considered new again", + dedupe.isNew(topic, r1again)); + + // "C" should still be in the pool — sending it again should be dup + SinkRecord r3dup = ClickHouseSinkTaskTest.spoofSinkRecord( + topic, keyField, "C", valueField, "z", + TimestampType.NO_TIMESTAMP_TYPE, ts); + Assert.assertFalse("Key C should still be in pool (not evicted)", + dedupe.isNew(topic, r3dup)); + } + + @Test + public void testDeDuplicationOff() { + Map properties = new HashMap(); + properties.put(ClickHouseSinkConnectorConfigVariables.DEDUPLICATION_POLICY.toString(), "off"); + + ClickHouseSinkConnectorConfig config = new ClickHouseSinkConnectorConfig(properties); + DeDuplicator dedupe = new DeDuplicator(config); + + String topic = "products"; + String keyField = "productId"; + String valueField = "amount"; + Long ts = System.currentTimeMillis(); + + SinkRecord r1 = ClickHouseSinkTaskTest.spoofSinkRecord( + topic, keyField, "11", valueField, "2000", + TimestampType.NO_TIMESTAMP_TYPE, ts); + + // When dedup is off, ALL records are new (no dedup check) + Assert.assertTrue(dedupe.isNew(topic, r1)); + Assert.assertTrue(dedupe.isNew(topic, r1)); + Assert.assertTrue(dedupe.isNew(topic, r1)); + } + + @Test + public void testMultiTopicIndependence() { + Map properties = new HashMap(); + properties.put(ClickHouseSinkConnectorConfigVariables.DEDUPLICATION_POLICY.toString(), "new"); + + ClickHouseSinkConnectorConfig config = new ClickHouseSinkConnectorConfig(properties); + DeDuplicator dedupe = new DeDuplicator(config); + + String keyField = "id"; + String valueField = "data"; + Long ts = System.currentTimeMillis(); - // Test Different Topic. - boolean resultDifferentTopic = dedupe.isNew("employees", recordTwo); - Assert.assertTrue(resultDifferentTopic == true); + // Same key across different topics should be independent + SinkRecord rTopic1 = ClickHouseSinkTaskTest.spoofSinkRecord( + "topic_a", keyField, "1", valueField, "x", + TimestampType.NO_TIMESTAMP_TYPE, ts); + SinkRecord rTopic2 = ClickHouseSinkTaskTest.spoofSinkRecord( + "topic_b", keyField, "1", valueField, "x", + TimestampType.NO_TIMESTAMP_TYPE, ts); + Assert.assertTrue("Key 1 in topic_a: new", + dedupe.isNew("topic_a", rTopic1)); + Assert.assertTrue("Same key 1 in topic_b: new (independent)", + dedupe.isNew("topic_b", rTopic2)); + Assert.assertFalse("Key 1 in topic_a again: duplicate", + dedupe.isNew("topic_a", rTopic1)); + Assert.assertFalse("Key 1 in topic_b again: duplicate", + dedupe.isNew("topic_b", rTopic2)); } }