[split 20/22] executor: offset re-insert fix, volatile pause barrier, per-thread cache invalidation generations, schema-cache TTL

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/CacheInvalidationManager.java

@@ -5,46 +5,120 @@
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * Singleton class that manages cache invalidation signals between DDL execution
  * and batch processing threads.
- * 
+ *
  * When a DDL statement is executed (e.g., ALTER TABLE ADD/DROP COLUMN), this manager
  * is notified so that cached DbWriter instances can be invalidated and refreshed
  * with the new schema.
- * 
- * Invalidation is tracked with a per-table monotonically increasing version
- * counter rather than a remove-on-read signal. Each cached DbWriter records the
- * version it was built at; a consumer rebuilds its writer whenever that version
- * is stale relative to the current table version. Because the version map is a
- * shared singleton but every batch-processing thread keeps its own writer cache,
- * this lets every thread independently detect and rebuild a stale writer without
- * any shared writer state or races.
- * 
- * This class is thread-safe and uses a lock-free concurrent map.
+ *
+ * <p>In addition to explicit DDL-triggered invalidation, this manager enforces
+ * a <b>time-to-live (TTL)</b> on the per-table schema cache. Keeping a cache
+ * entry forever is unsafe: if a schema change is ever missed (a DDL that is
+ * filtered, a connector that was down during the ALTER, or a metadata-cache bug
+ * like the Altinity 2.8.0 issue), the connector would keep inserting against a
+ * stale schema indefinitely. With a TTL, any such drift self-heals within at
+ * most one TTL window because the cache is rebuilt from {@code system.columns}.
+ * The TTL defaults to one hour.</p>
+ *
+ * <p>This class is thread-safe.</p>
  */
 public class CacheInvalidationManager {
 
     private static final Logger log = LogManager.getLogger(CacheInvalidationManager.class);
 
+    /** Default cache TTL: rebuild every table's schema cache at least hourly. */
+    public static final long DEFAULT_CACHE_TTL_MS = 60L * 60L * 1000L;
+
     /**
      * Singleton instance.
      */
     private static final CacheInvalidationManager INSTANCE = new CacheInvalidationManager();
 
     /**
-     * Thread-safe map of table names (in format "database.table") to their current
-     * invalidation version. Absent tables are treated as version 0.
+     * Monotonic DDL generation per table, keyed by "database.table".
+     *
+     * A plain Set with a remove-on-read check was not usable here: the connector runs
+     * a pool of worker threads, each holding its OWN topicToDbWriterMap cache. The
+     * first thread to observe the invalidation consumed it, so every other thread
+     * kept serving a stale DbWriter (and therefore a stale column list) forever.
+     * A generation counter lets every cache holder invalidate independently.
      */
-    private final Map<String, Long> tableVersions = new ConcurrentHashMap<>();
+    private final Map<String, AtomicLong> tableGenerations = new ConcurrentHashMap<>();
+
+    /**
+     * Per-table timestamp (epoch ms) of the last cache build, for TTL expiry.
+     * Keyed by fully qualified table name ("database.table").
+     */
+    private final Map<String, Long> lastBuildEpochMs = new ConcurrentHashMap<>();
+
+    /** Cache TTL in milliseconds; entries older than this are forced to rebuild. */
+    private volatile long cacheTtlMs = DEFAULT_CACHE_TTL_MS;
 
     /**
      * Private constructor to enforce singleton pattern.
      */
     private CacheInvalidationManager() {
     }
 
+    /**
+     * Sets the cache TTL. A value &lt;= 0 disables TTL-based expiry (DDL-triggered
+     * invalidation still applies).
+     *
+     * @param ttlMs TTL in milliseconds.
+     */
+    public void setCacheTtlMs(long ttlMs) {
+        this.cacheTtlMs = ttlMs;
+        log.info("Schema cache TTL set to {}ms", ttlMs);
+    }
+
+    /** Returns the current cache TTL in milliseconds. */
+    public long getCacheTtlMs() {
+        return cacheTtlMs;
+    }
+
+    /**
+     * Records that a table's schema cache was just (re)built. Resets the TTL
+     * clock for that table. Call this whenever a fresh {@code DbWriter} /
+     * column map is constructed for the table.
+     *
+     * @param tableName fully qualified table name ("database.table").
+     */
+    public void markCacheBuilt(String tableName) {
+        if (tableName != null && !tableName.isEmpty()) {
+            lastBuildEpochMs.put(tableName, System.currentTimeMillis());
+        }
+    }
+
+    /**
+     * Returns true if the table's cache has exceeded its TTL and must be
+     * rebuilt. Tables with no recorded build time are treated as expired so the
+     * first access establishes a fresh, timestamped entry.
+     *
+     * @param tableName fully qualified table name ("database.table").
+     * @return true if the cache is stale per the TTL policy.
+     */
+    public boolean isCacheExpired(String tableName) {
+        if (tableName == null || tableName.isEmpty()) {
+            return false;
+        }
+        if (cacheTtlMs <= 0) {
+            return false;
+        }
+        Long built = lastBuildEpochMs.get(tableName);
+        if (built == null) {
+            return true;
+        }
+        boolean expired = (System.currentTimeMillis() - built) >= cacheTtlMs;
+        if (expired) {
+            log.info("Schema cache for {} exceeded TTL ({}ms); forcing rebuild", tableName, cacheTtlMs);
+        }
+        return expired;
+    }
+
     /**
      * Returns the singleton instance.
      * 
@@ -55,50 +129,56 @@ public static CacheInvalidationManager getInstance() {
     }
 
     /**
-     * Marks a table for cache invalidation by bumping its version. This should be
-     * called after a DDL statement is successfully executed. The version is
-     * monotonically increasing, so any cached DbWriter built at an older version
-     * will be rebuilt on its next access.
+     * Marks a table for cache invalidation. This should be called after a DDL
+     * statement is successfully executed.
      * 
      * @param tableName The fully qualified table name in format "database.table".
      */
     public void invalidateTable(String tableName) {
         if (tableName != null && !tableName.isEmpty()) {
-            long version = tableVersions.merge(tableName, 1L, Long::sum);
-            log.info("Marked table {} for cache invalidation after DDL (version {})",
-                    tableName, version);
+            long generation = tableGenerations
+                    .computeIfAbsent(tableName, k -> new AtomicLong())
+                    .incrementAndGet();
+            log.info("Marked table {} for cache invalidation after DDL (generation {})",
+                    tableName, generation);
         }
     }
 
     /**
-     * Returns the current invalidation version for a table. Tables that have never
-     * been invalidated are treated as version 0.
-     * 
+     * Returns the current DDL generation for a table. A cache holder records the
+     * generation it built its entry at and rebuilds whenever the value changes.
+     *
      * @param tableName The fully qualified table name in format "database.table".
-     * @return The current invalidation version, or 0 if the table has never been
-     *         invalidated.
+     * @return The current generation, or 0 if the table has never had a DDL applied.
      */
-    public long getVersion(String tableName) {
+    public long currentGeneration(String tableName) {
         if (tableName == null || tableName.isEmpty()) {
             return 0L;
         }
-        return tableVersions.getOrDefault(tableName, 0L);
+        AtomicLong generation = tableGenerations.get(tableName);
+        return generation == null ? 0L : generation.get();
     }
 
     /**
-     * Clears all invalidation versions. Useful for testing.
+     * Clears all pending invalidations. Useful for testing.
      */
     public void clearAll() {
-        tableVersions.clear();
+        // tablesToInvalidate (a remove-on-read Set) no longer exists: it was
+        // replaced by the tableGenerations counter, because the Set let the
+        // FIRST worker thread consume an invalidation while every other thread
+        // kept serving a stale DbWriter. Clear the generation map plus this
+        // PR's TTL bookkeeping.
+        tableGenerations.clear();
+        lastBuildEpochMs.clear();
     }
 
     /**
-     * Returns the number of tables that have a tracked invalidation version.
+     * Returns the number of tables that have had at least one DDL applied.
      * Useful for testing.
-     * 
-     * @return The number of tables with a tracked invalidation version.
+     *
+     * @return The number of tracked tables.
      */
     public int pendingInvalidations() {
-        return tableVersions.size();
+        return tableGenerations.size();
     }
 }

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/DbWriter.java

@@ -45,14 +45,14 @@ public class DbWriter extends BaseDbWriter {
      */
     private Map<String, String> columnNameToDataTypeMap = new LinkedHashMap<>();
 
-    /**
-     * The cache invalidation version this writer was built at. Compared against
-     * {@link com.altinity.clickhouse.sink.connector.db.CacheInvalidationManager}
-     * to detect when the writer is stale and must be rebuilt after a DDL.
-     */
-    @Getter
-    @Setter
-    private long cacheInvalidationVersion = 0;
+    // NOTE: 2.10.0 tracked staleness with a `cacheInvalidationVersion` field on
+    // the writer itself. That mechanism is deliberately NOT carried over: it is
+    // fully superseded by the per-topic generation map the executors keep
+    // (topicToDbWriterGeneration) plus CacheInvalidationManager's TTL, which
+    // additionally self-heals a DDL that was missed entirely. Keeping the field
+    // as well would leave a second, unmaintained staleness signal that reads as
+    // authoritative — exactly the kind of stale-schema footgun this whole area
+    // exists to remove.
 
     /**
      * The engine type of the target table in ClickHouse (e.g., MergeTree,
@@ -144,7 +144,10 @@ public DbWriter(
             initializeTableEngine(hostName, record);
             configureEngineSpecificColumns();
         } catch (Exception e) {
-            log.error("***** DBWriter error initializing ****", e);
+            log.error("***** FATAL: DBWriter initialization failed for "
+                    + database + "." + tableName + " ****", e);
+            throw new RuntimeException("DbWriter initialization failed for "
+                    + database + "." + tableName, e);
         }
     }
 
@@ -169,6 +172,13 @@ private void autoCreateTable(long taskId, String hostName,
                         .toArray(new Field[0]);
             }
 
+            if (fields == null || fields.length == 0) {
+                log.error("Cannot auto-create table {}.{}: CDC record has "
+                        + "neither after nor before struct with fields",
+                        database, tableName);
+                return;
+            }
+
             String rmtDeleteColumn = this.config.getString(
                     ClickHouseSinkConnectorConfigVariables
                             .REPLACING_MERGE_TREE_DELETE_COLUMN
@@ -364,9 +374,9 @@ public String getOffsetStorageDatabaseName() {
         }
         String[] offsetStorageDatabaseNameArray = offsetSchemaHistoryTable.split(
                 "\\.");
-        if (offsetStorageDatabaseNameArray.length <= 2) {
-            log.warn("Skipping creating offset schema history table as the "
-                    + "query was not provided in configuration");
+        if (offsetStorageDatabaseNameArray.length < 2) {
+            log.warn("Invalid offset storage table name format: expected "
+                    + "database.table, got: " + offsetSchemaHistoryTable);
             return null;
         }
         String offsetStorageDatabaseName = offsetStorageDatabaseNameArray[0];

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchExecutor.java

@@ -21,8 +21,22 @@ public class ClickHouseBatchExecutor extends
 
     /**
      * Flag indicating whether the executor is paused.
+     *
+     * <p>Must be {@code volatile}. This flag is the DDL-versus-DML barrier:
+     * {@code DebeziumChangeEventCapture} calls {@link #pause()} from the
+     * Debezium change-event thread before applying a DDL and {@link #resume()}
+     * after it, while the batch-pool threads read the flag in the
+     * {@link #beforeExecute} spin loop. Writer and readers are different
+     * threads with no lock and no other happens-before edge between them, so
+     * without {@code volatile} the JMM gives no visibility guarantee (JLS
+     * 17.4), and the tight {@code while (isPaused)} loop - whose body touches
+     * no other shared state - may hoist the read out of the loop entirely.</p>
+     *
+     * <p>The consequence is not a stall but silent corruption: a batch thread
+     * that never observes the pause applies DML against a schema that is
+     * mid-DDL. Pinned by {@code ExecutorPauseVisibilityTest}.</p>
      */
-    boolean isPaused = false;
+    volatile boolean isPaused = false;
 
     /**
      * Constructs a ClickHouseBatchExecutor with the given core pool size