[BUG-CONC-5] Check-Then-Act Race in getDbWriterForTable - Non-Atomic Map Operations

[minguyen9988]

Bug Description

Severity: HIGH
Component: ClickHouseBatchRunnable
Version: 2.9.1

Problem

The getDbWriterForTable() method uses a check-then-act pattern on topicToDbWriterMap (a plain HashMap) that is not atomic. In a multi-threaded environment, the containsKey() check (line 515) and subsequent get()/remove() (lines 520, 522) are separate operations that can be interleaved by other threads.

Affected Code

File: sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchRunnable.java

// Lines 510-533: Non-atomic check-then-act pattern
public DbWriter getDbWriterForTable(String topicName, ...) {
    DbWriter writer = null;
    if (this.topicToDbWriterMap.containsKey(topicName)) {  // Line 515: CHECK
        // ... cache invalidation check ...
        this.topicToDbWriterMap.remove(topicName);          // Line 520: ACT (not atomic with CHECK)
        writer = this.topicToDbWriterMap.get(topicName);    // Line 522: ACT (not atomic with CHECK)
        return writer;
    }
    writer = new DbWriter(...);                             // Line 526: Multiple threads may create
    this.topicToDbWriterMap.put(topicName, writer);          // Line 531: Last write wins
    return writer;
}

Impact

• Duplicate DbWriter creation: Multiple threads creating DbWriter for the same topic wastes resources and connections
• Lost updates: One thread's DbWriter may be overwritten by another thread
• Stale references: A thread may use a DbWriter that was removed by another thread for cache invalidation
• Related to BUG: HashMap Race Condition in ClickHouseBatchWriter (not fixed like ClickHouseBatchRunnable) #1251 (HashMap race condition) - both issues share the root cause of using non-thread-safe HashMap

Proposed Fix

Use ConcurrentHashMap.computeIfAbsent() for atomic check-then-create:

// Thread-safe atomic creation
private final Map<String, DbWriter> topicToDbWriterMap = new ConcurrentHashMap<>();

public DbWriter getDbWriterForTable(String topicName, ...) {
    // Handle cache invalidation
    String fqtn = databaseName + "." + tableName;
    if (CacheInvalidationManager.getInstance().shouldInvalidate(fqtn)) {
        topicToDbWriterMap.remove(topicName);
    }
    
    return topicToDbWriterMap.computeIfAbsent(topicName, key ->
        new DbWriter(dbCredentials.getHostName(), ...));
}