diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/CacheInvalidationManager.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/CacheInvalidationManager.java index b3f593209..4b9403ace 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/CacheInvalidationManager.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/CacheInvalidationManager.java @@ -5,39 +5,58 @@ 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. + * + *

In addition to explicit DDL-triggered invalidation, this manager enforces + * a time-to-live (TTL) 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.

+ * + *

This class is thread-safe.

*/ 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 tableVersions = new ConcurrentHashMap<>(); + private final Map 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 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. @@ -45,6 +64,61 @@ public class CacheInvalidationManager { private CacheInvalidationManager() { } + /** + * Sets the cache TTL. A value <= 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(); } } diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/DbWriter.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/DbWriter.java index 417dee1f0..2dece72d9 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/DbWriter.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/DbWriter.java @@ -45,14 +45,14 @@ public class DbWriter extends BaseDbWriter { */ private Map 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]; diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchExecutor.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchExecutor.java index 2efd8de8b..560404f89 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchExecutor.java +++ b/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. + * + *

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.

+ * + *

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}.

*/ - boolean isPaused = false; + volatile boolean isPaused = false; /** * Constructs a ClickHouseBatchExecutor with the given core pool size diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchRunnable.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchRunnable.java index 32f3ae56a..cf8405dee 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchRunnable.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchRunnable.java @@ -64,6 +64,13 @@ public class ClickHouseBatchRunnable implements Runnable { /** * Connection used to create the Debezium storage database. + * + *

Opened LAZILY via {@link #getSystemConnection()}. It used to be opened + * in the constructor, which meant simply constructing this object required a + * reachable ClickHouse — even for pure-parsing calls like + * {@link #getTableFromTopic(String)} that touch no database at all. That + * became a hard failure once BaseDbWriter.createConnection was changed to + * throw rather than return null.

*/ private Connection systemConnection; @@ -83,6 +90,12 @@ public class ClickHouseBatchRunnable implements Runnable { */ private Map topicToDbWriterMap; + /** + * DDL generation each cached DbWriter in {@link #topicToDbWriterMap} was built at, + * keyed by topic name. Used to detect schema changes applied since caching. + */ + private Map topicToDbWriterGeneration; + /** * Database credentials. */ @@ -93,6 +106,17 @@ public class ClickHouseBatchRunnable implements Runnable { */ private List currentBatch = null; + /** + * True when {@link #currentBatch} has already been flushed to ClickHouse + * successfully and is only waiting for older in-flight batches to clear + * before its offsets can be committed. While set, the retry loop must NOT + * re-execute the inserts: re-flushing an already-flushed batch writes a + * duplicate part per retry, and with no pacing sleep the loop re-inserted + * one record 1,000 times in CI — SELECTs without FINAL then saw duplicate + * rows until the merge caught up. + */ + private boolean currentBatchFlushed = false; + /** * Shared watermark (owned by ClickHouseSinkTask): highest Kafka offset per * TopicPartition durably inserted into ClickHouse. Updated after each @@ -190,9 +214,12 @@ private ClickHouseBatchRunnable( } //this.queryToRecordsMap = new HashMap<>(); this.topicToDbWriterMap = new HashMap<>(); + this.topicToDbWriterGeneration = new HashMap<>(); //this.topicToRecordsMap = new HashMap<>(); this.dbCredentials = parseDBConfiguration(); - this.systemConnection = createConnection(BaseDbWriter.SYSTEM_DB); + // systemConnection is opened lazily — see getSystemConnection(). Opening + // it here made construction require a reachable ClickHouse even for + // operations that never touch the database. try { this.databaseOverrideMap = Utils.parseSourceToDestinationDatabaseMap( this.config.getString( @@ -355,27 +382,77 @@ public void run() { logErrorToClickHouse(e, taskId, errorTableName); } - // Classify the error to decide whether to retry or stop - ClickHouseErrorClassifier.ErrorCategory category = ClickHouseErrorClassifier.classify(e); + // Two INDEPENDENT fatal detectors, both retained. Detector 2 is + // 2.10.0's error classifier; detector 1 is develop's addition and is + // a strict superset — 2.10.0 contributes nothing that is dropped here. + // + // 1. A poisoned OffsetStorageWriter is unrecoverable in-process: + // once its flush semaphore is leaked, every future beginFlush() + // throws for the life of the JVM. Swallowing it turns the fault + // into SILENT data divergence -- ClickHouse writes keep + // succeeding while the binlog offset is frozen, so the connector + // looks healthy, replays the same batch forever, and re-delivers + // from a stale offset on restart. This exception is a Kafka + // ConnectException carrying NO ClickHouse error code, so the + // classifier below cannot see it -- it must be checked first. + if (isOffsetWriterPoisoned(e)) { + log.error("FATAL: the Debezium OffsetStorageWriter is stuck in the " + + "'already flushing' state. Offsets can no longer be committed, " + + "so replication would continue writing rows while the binlog " + + "position stays frozen. Stopping task Task({}) to prevent " + + "silent data divergence -- the connector must be restarted.", + taskId); + throw new RuntimeException( + "OffsetStorageWriter is permanently stuck flushing; " + + "stopping task to prevent silent data divergence", e); + } + + // 2. Deterministic ClickHouse errors (auth, schema, type mismatch) + // will never succeed on retry, so retrying forever stalls binlog + // advancement for ALL tables. Classify by error code and stop. + ClickHouseErrorClassifier.ErrorCategory category = + ClickHouseErrorClassifier.classify(e); int errorCode = ClickHouseErrorClassifier.extractErrorCode(e); if (category == ClickHouseErrorClassifier.ErrorCategory.FATAL) { - log.error("FATAL ClickHouse error (Code: {}) -- this batch will never succeed. " + - "Discarding batch and stopping task to prevent silent data loss. " + - "Manual intervention required.", errorCode); - // Clear the stuck batch so it is not retried forever + log.error("FATAL ClickHouse error (Code: {}) -- this batch will never succeed. " + + "Discarding batch and stopping task to prevent silent data loss. " + + "Manual intervention required.", errorCode); + // Clear the stuck batch so it is not retried forever. currentBatch = null; - // Rethrow to stop the scheduled executor -- silent swallowing causes - // binlog advancement to stall and blocks replication for ALL tables - throw new RuntimeException("Fatal ClickHouse error, stopping task", e); + // Wrapped rather than rethrown directly: run() implements + // Runnable and cannot declare a checked exception, and the + // upstream `throw e` does not compile here. + throw new RuntimeException( + "Fatal ClickHouse error (Code: " + errorCode + + "), stopping task", e); } else { - log.warn("Retriable ClickHouse error (Code: {}, Category: {}) -- " + - "batch will be retried on next scheduled run.", errorCode, category); + log.warn("Retriable ClickHouse error (Code: {}, Category: {}) -- " + + "batch will be retried on next scheduled run.", errorCode, category); } } } + /** + * Detects the unrecoverable "OffsetStorageWriter is already flushing" condition. + * + * @param e the exception thrown from the batch loop. + * @return true when offset commits can no longer succeed in this JVM. + */ + private boolean isOffsetWriterPoisoned(Throwable e) { + Throwable current = e; + while (current != null) { + String message = current.getMessage(); + if (message != null && message.contains("OffsetStorageWriter is already flushing")) { + return true; + } + current = current.getCause(); + } + return false; + } + + /** * Run loop for hash-based routing mode. * Only processes batches assigned to this thread. @@ -451,6 +528,24 @@ private void runLegacyMode(Long taskId, String sourceTimeZone, String serverTime * @throws Exception if processing fails */ private void processBatch(String sourceTimeZone, String serverTimeZone) throws Exception { + // Retry of an ALREADY-FLUSHED batch: the data is in ClickHouse and the + // batch is only waiting for older in-flight batches to clear before + // its offsets may be committed. Do NOT re-run the inserts (each replay + // writes a duplicate part, visible to SELECTs without FINAL until the + // merge collapses it — CI observed one record re-inserted 1,000 times) + // and do NOT re-add the batch to the in-flight map (that would undo + // its move to completedBatches and re-block every newer batch). Just + // re-check committability, with a short pause to avoid a hot spin. + if (currentBatchFlushed) { + if (DebeziumOffsetManagement.checkIfBatchCanBeCommitted(currentBatch)) { + currentBatch = null; + currentBatchFlushed = false; + } else { + Thread.sleep(100); + } + return; + } + // If replication history is enabled, add the records to the history table. addRecordsToHistoryTable(currentBatch, sourceTimeZone, serverTimeZone); @@ -464,7 +559,7 @@ private void processBatch(String sourceTimeZone, String serverTimeZone) throws E // Group records by topic name. // Create a new map of topic name to list of records. Map> topicToRecordsMap = - new ConcurrentHashMap<>(); + new HashMap<>(); currentBatch.forEach(record -> { String topicName = record.getTopic(); // If the topic name is not present, create a new list and @@ -509,14 +604,16 @@ private void processBatch(String sourceTimeZone, String serverTimeZone) throws E if (result) { + // The flush succeeded: remember that so a commit-blocked retry + // does not re-execute the inserts (duplicate parts) or re-add the + // batch to the in-flight map. + currentBatchFlushed = true; // Step 2: Check if the batch can be committed. if(DebeziumOffsetManagement.checkIfBatchCanBeCommitted(currentBatch)) { currentBatch = null; + currentBatchFlushed = false; } } - Thread.sleep(config.getLong( - ClickHouseSinkConnectorConfigVariables. - BUFFER_FLUSH_TIME.toString())); ///// ***** END PROCESSING BATCH ************************** } @@ -528,12 +625,27 @@ private void processBatch(String sourceTimeZone, String serverTimeZone) throws E * @throws SQLException */ private void addRecordsToHistoryTable(List records, String sourceTimeZone, String serverTimeZone) throws SQLException { + if(records == null || records.isEmpty()) { + return; + } if(config.getBoolean(ClickHouseSinkConnectorConfigVariables.REPLICATION_HISTORY_ENABLE.toString())) { String databaseName = config.getString(ClickHouseSinkConnectorConfigVariables.REPLICATION_HISTORY_DATABASE_NAME.toString()); String tableName = config.getString(ClickHouseSinkConnectorConfigVariables.REPLICATION_HISTORY_TABLE_NAME.toString()); + if (records == null || records.isEmpty()) { + log.warn("Skipping history table update — batch is empty"); + return; + } Connection databaseConn = getClickHouseConnection(databaseName); DbWriter writer = getDbWriterForTable(databaseName + "." + tableName, tableName, databaseName, records.get(0), databaseConn); + if (writer == null) { + // getDbWriterForTable returns null when the table is still + // frozen for a DDL reconciliation. Skip the history write for + // this batch rather than dereferencing null; it is retried. + log.error("*** DbWriter is null for {}.{} (table frozen for DDL); " + + "skipping history write for this batch", databaseName, tableName); + return; + } BinLogHistory binLogHistory = new BinLogHistory(); binLogHistory.addRecordsToHistoryTable(config, tableName, writer.getConnection(), "", records, sourceTimeZone, serverTimeZone); @@ -571,53 +683,187 @@ public DbWriter getDbWriterForTable(String topicName, String tableName, String databaseName, ClickHouseStruct record, Connection connection) { - // Compare the cached writer's build version against the shared, monotonic - // table version. A mismatch means a DDL invalidated this table after the - // writer was built, so it must be rebuilt with the fresh schema. + DbWriter writer = null; String fullyQualifiedTableName = databaseName + "." + tableName; - long currentVersion = CacheInvalidationManager.getInstance() - .getVersion(fullyQualifiedTableName); - DbWriter writer = this.topicToDbWriterMap.get(topicName); - boolean invalidated = false; - if (writer != null) { - if (writer.getCacheInvalidationVersion() == currentVersion) { - return writer; + // Block while a DDL schema change is being applied + reconciled for + // this table. This prevents inserting against a stale column cache + // (which would silently drop newly added source columns). The DDL + // thread freezes the table before ALTER and only unfreezes after the + // destination schema, the source schema, and the cache all agree. + // The return value MUST be honoured: false means the freeze did not + // clear within the timeout, i.e. a DDL reconciliation is stuck. + // Proceeding anyway would insert against the very stale column cache + // this freeze exists to prevent, silently dropping newly added source + // columns. Return null so the caller defers the batch and retries. + boolean unfrozen = TableReplicationFreezeManager.getInstance() + .awaitUnfrozen(fullyQualifiedTableName, DDLSchemaChangeWaiter.DEFAULT_TIMEOUT_MS); + if (!unfrozen) { + log.error("Table {} still frozen after {}ms; deferring this batch rather " + + "than inserting against a possibly stale schema cache.", + fullyQualifiedTableName, DDLSchemaChangeWaiter.DEFAULT_TIMEOUT_MS); + return null; + } + + long generation = CacheInvalidationManager.getInstance() + .currentGeneration(fullyQualifiedTableName); + + if (this.topicToDbWriterMap.containsKey(topicName)) { + // Rebuild when EITHER signal fires. Both sides of this merge are + // kept deliberately: + // - generation change: a DDL was applied since this writer was + // cached. Comparing generations (rather than consuming a + // one-shot flag) is what lets every worker thread's private + // cache invalidate independently; the old remove-on-read Set + // let the first reader consume the signal and left the other + // threads serving a stale column list. + // - TTL expiry: defensive self-heal in case a schema change was + // missed entirely and never bumped the generation. + Long cachedGeneration = this.topicToDbWriterGeneration.get(topicName); + boolean ddlInvalidated = + cachedGeneration == null || cachedGeneration != generation; + boolean ttlExpired = CacheInvalidationManager.getInstance() + .isCacheExpired(fullyQualifiedTableName); + if (!ddlInvalidated && !ttlExpired) { + return this.topicToDbWriterMap.get(topicName); } - log.info("Invalidating cached DbWriter for {} after DDL (version {} -> {})", - topicName, writer.getCacheInvalidationVersion(), currentVersion); + log.info("Rebuilding cached DbWriter for {} ({}; generation {} -> {})", + topicName, + ddlInvalidated ? "DDL invalidation" : "cache TTL expiry", + cachedGeneration, generation); this.topicToDbWriterMap.remove(topicName); - invalidated = true; } writer = new DbWriter(this.dbCredentials.getHostName(), this.dbCredentials.getPort(), databaseName, tableName, this.dbCredentials.getUserName(), this.dbCredentials.getPassword(), this.config, record, connection); - writer.setCacheInvalidationVersion(currentVersion); this.topicToDbWriterMap.put(topicName, writer); - // Log the resolved schema whenever this table has seen a DDL (version > 0). - // This covers both rebuilding a stale writer and building a fresh writer at - // the current version after a burst of DDLs, so the post-DDL schema is always - // observable regardless of which thread ends up owning the writer. - if (invalidated || currentVersion > 0) { - logRefreshedColumns(topicName, writer, invalidated); + // Record the generation this writer was built at (so the next call can + // detect a later DDL) AND stamp the TTL clock, then verify the freshly + // built column cache actually covers every source column. + this.topicToDbWriterGeneration.put(topicName, generation); + CacheInvalidationManager.getInstance().markCacheBuilt(fullyQualifiedTableName); + if (!verifySourceSchemaIntegrity(fullyQualifiedTableName, record, writer)) { + // The writer's column map does not cover every source column, so + // inserting with it would silently DROP those columns. Evict it and + // return null: the caller defers this batch and retries, by which + // time the rebuild picks up the now-visible columns. Returning the + // writer anyway was the stale-writer escape hatch that made the + // integrity check advisory instead of protective. + this.topicToDbWriterMap.remove(topicName); + this.topicToDbWriterGeneration.remove(topicName); + return null; } return writer; } /** - * Logs the refreshed column name and type map of a DbWriter that was rebuilt - * after a DDL cache invalidation. + * Verifies that every column present in the source change event also exists + * in the freshly rebuilt destination column cache. If a source column is + * missing, the INSERT path would silently drop it (data loss), so we log + * loudly and re-mark the table for invalidation. Re-marking forces the next + * batch to rebuild again rather than insert lossy data, which lets a + * still-propagating schema change catch up. This is a safety net layered on + * top of the per-table replication freeze. * - * @param topicName the topic whose writer was rebuilt - * @param writer the freshly constructed DbWriter + * @return {@code true} when the writer may safely be used; {@code false} + * when its column map is missing source columns, in which case the + * caller MUST NOT insert with it. */ - private void logRefreshedColumns(String topicName, DbWriter writer, boolean rebuilt) { - Map cols = writer.getColumnNameToDataTypeMap(); - if (cols != null) { - log.info("{} DbWriter schema for {} at cache version {} ({} columns): {}", - rebuilt ? "Rebuilt" : "Built", topicName, - writer.getCacheInvalidationVersion(), cols.size(), cols); + private boolean verifySourceSchemaIntegrity(String fullyQualifiedTableName, + ClickHouseStruct record, + DbWriter writer) { + try { + // The replication-history table is EXEMPT: it has its own fixed + // audit schema (gtid/ddl/before/after/...) and source-row columns + // are serialized into its payload columns, not mapped one-to-one. + // Comparing a source event against it reports every source column + // "missing" — the gate then blocks each batch for the full + // visibility-wait timeout polling system.columns for columns that + // will never appear, skips the history write, and starves the + // shared connection pool for the whole process. + // See SourceSchemaIntegrityValidator.isReplicationHistoryTable. + if (SourceSchemaIntegrityValidator.isReplicationHistoryTable( + this.config, fullyQualifiedTableName)) { + return true; + } + java.util.List sourceColumns = + SourceSchemaColumns.fromRecord(record); + if (sourceColumns.isEmpty() || writer == null + || writer.getColumnNameToDataTypeMap() == null) { + return true; + } + SourceSchemaIntegrityValidator.Result result = + SourceSchemaIntegrityValidator.check(sourceColumns, + writer.getColumnNameToDataTypeMap().keySet()); + if (!result.isConsistent()) { + String[] parts = fullyQualifiedTableName.split("\\.", 2); + // Source columns that map to destination ALIAS/MATERIALIZED + // columns are NOT missing: the insertable cache excludes them + // by design — ClickHouse computes their values and rejects + // inserts into them, so the source value is intentionally not + // written. MySQL generated columns land here on EVERY batch; + // without this filter the gate invalidated and rebuilt the + // writer forever, livelocking replication for the table + // (observed: 8,208 invalidate/rebuild cycles in one CI run). + java.util.List genuinelyMissing = + result.getMissingInDestination(); + if (parts.length == 2 && writer.getConnection() != null) { + try { + java.util.Set generated = new DBMetadata(this.config) + .getAliasAndMaterializedColumnsForTableAndDatabase( + parts[1], parts[0], writer.getConnection()); + genuinelyMissing = SourceSchemaIntegrityValidator + .excludeGeneratedColumns(genuinelyMissing, generated); + } catch (Exception ex) { + log.warn("Could not fetch ALIAS/MATERIALIZED columns for {}: {}", + fullyQualifiedTableName, ex.getMessage()); + } + } + if (genuinelyMissing.isEmpty()) { + // Every "missing" column is computed by the destination — + // the writer is correct as built. Invalidating here is the + // livelock; the writer must be used as-is. + return true; + } + // Generalized visibility gate: rather than only logging, WAIT for + // the destination to actually gain the missing columns. This is + // what covers RENAME/MODIFY COLUMN, whose net effect is "these + // columns must exist" but which the ADD/DROP text parser in + // waitForSchemaVisibility cannot see. + java.util.Collection stillMissing = genuinelyMissing; + if (parts.length == 2 && writer.getConnection() != null) { + stillMissing = new DDLSchemaChangeWaiter() + .waitForExpectedColumns(writer.getConnection(), parts[0], + parts[1], genuinelyMissing); + } + if (stillMissing.isEmpty()) { + // Columns landed while we waited; force a rebuild so the next + // call picks up a writer that actually knows about them. + log.warn("Schema integrity for {}: source column(s) {} were missing " + + "but became visible while waiting; re-marking for " + + "invalidation so the writer is rebuilt.", + fullyQualifiedTableName, genuinelyMissing); + } else { + log.error("Schema integrity violation for {}: source column(s) {} " + + "still missing from the destination after waiting; " + + "inserting now would drop them. Re-marking for " + + "invalidation.", + fullyQualifiedTableName, stillMissing); + } + CacheInvalidationManager.getInstance().invalidateTable(fullyQualifiedTableName); + // Either way the CURRENT writer's column map predates those + // columns, so it must not be used for this batch. + return false; + } + return true; + } catch (Exception e) { + log.warn("Error during source schema integrity check for {}: {}", + fullyQualifiedTableName, e.getMessage()); + // Fail open on an unexpected checker error: the freeze and + // generation gates upstream are the primary protections, and + // blocking every batch on a bug in this safety net would be worse. + return true; } } @@ -644,7 +890,23 @@ public ZoneId getServerTimeZone(ClickHouseSinkConnectorConfig config) { if (userProvidedTimeZoneId != null) { return userProvidedTimeZoneId; } - return new DBMetadata(config).getServerTimeZone(this.systemConnection); + return new DBMetadata(config).getServerTimeZone(getSystemConnection()); + } + + /** + * Returns the shared system-database connection, opening it on first use. + * + *

Lazy so that constructing this runnable does not require a reachable + * ClickHouse. Callers that genuinely need the database still fail loudly, + * because createConnection throws when it cannot connect.

+ * + * @return the system-database connection. + */ + private synchronized Connection getSystemConnection() { + if (this.systemConnection == null) { + this.systemConnection = createConnection(BaseDbWriter.SYSTEM_DB); + } + return this.systemConnection; } /** @@ -693,40 +955,60 @@ private boolean processBatchRecords(List records, String topic DbWriter writer = getDbWriterForTable(topicName, tableName, databaseName, firstRecord, databaseConn); - PreparedStatementExecutor preparedStatementExecutor = new - PreparedStatementExecutor(writer.getReplacingMergeTreeDeleteColumn(), - writer.isReplacingMergeTreeWithIsDeletedColumn(), writer.getSignColumn(), - writer.getVersionColumn(), writer.getDatabaseName(), - getServerTimeZone(this.config)); - if (writer == null || writer.wasTableMetaDataRetrieved() == false) { + // Validate writer before using it — null writer causes NPE in + // PreparedStatementExecutor creation and error logging + if (writer == null) { + log.error("*** DbWriter is null for {}.{}, retrying", databaseName, tableName); + writer = getDbWriterForTable(topicName, tableName, databaseName, + firstRecord, databaseConn); + } + if (writer == null) { + log.error("*** DbWriter still null for {}.{}, retrying on next attempt", + databaseName, tableName); + return false; + } + if (writer.wasTableMetaDataRetrieved() == false) { log.error(String.format("*** TABLE METADATA not retrieved for " + "Database(%s), table(%s) retrying", writer.getDatabaseName(), writer.getTableName())); - if (writer == null) { - writer = getDbWriterForTable(topicName, tableName, databaseName, - firstRecord, databaseConn); - } - if (writer.wasTableMetaDataRetrieved() == false) - writer.updateColumnNameToDataTypeMap(); - if (writer == null || - writer.wasTableMetaDataRetrieved() == false) { + writer.updateColumnNameToDataTypeMap(); + if (writer.wasTableMetaDataRetrieved() == false) { log.error(String.format("*** TABLE METADATA not retrieved for " + "Database(%s), table(%s), retrying on next attempt", - writer.getDatabaseName(), writer.getTableName())); + databaseName, tableName)); return false; } } + PreparedStatementExecutor preparedStatementExecutor = new + PreparedStatementExecutor(writer.getReplacingMergeTreeDeleteColumn(), + writer.isReplacingMergeTreeWithIsDeletedColumn(), writer.getSignColumn(), + writer.getVersionColumn(), writer.getDatabaseName(), + getServerTimeZone(this.config)); // Step 1: The Batch Insert with preparedStatement in JDBC works by // forming the Query and then adding records to the Batch. // This step creates a Map of Query -> Records (List of ClickHouseStruct). Map>, List> queryToRecordsMap = new HashMap<>(); Map partitionToOffsetMap = new HashMap<>(); + // Pass the writer's connector-managed column names (version/sign/ + // delete columns as actually named in the table engine) so the query + // formatter keeps them in the insert list while omitting genuine + // destination-only data columns the source event does not carry. + java.util.List managedColumns = new ArrayList<>(); + if (writer.getVersionColumn() != null) { + managedColumns.add(writer.getVersionColumn()); + } + if (writer.getSignColumn() != null) { + managedColumns.add(writer.getSignColumn()); + } + if (writer.getReplacingMergeTreeDeleteColumn() != null) { + managedColumns.add(writer.getReplacingMergeTreeDeleteColumn()); + } result = new GroupInsertQueryWithBatchRecords() .groupQueryWithRecords(records, queryToRecordsMap, partitionToOffsetMap, this.config, tableName, writer.getDatabaseName(), writer.getConnection(), - writer.getColumnNameToDataTypeMap()); + writer.getColumnNameToDataTypeMap(), managedColumns); BlockMetaData bmd = new BlockMetaData(); long maxBufferSize = this.config.getLong( ClickHouseSinkConnectorConfigVariables. @@ -737,17 +1019,7 @@ private boolean processBatchRecords(List records, String topic // and the records are flushed to ClickHouse. result = flushRecordsToClickHouse(topicName, writer, queryToRecordsMap, bmd, maxBufferSize, preparedStatementExecutor); - if (result) { - // Records are now DURABLY in ClickHouse: advance the shared watermark - // (max offset per TopicPartition) so ClickHouseSinkTask.preCommit() - // only commits offsets that were actually persisted. Without this the - // offset advances on consume, and a crash/restart silently loses the - // records that were consumed but never inserted. - partitionToOffsetMap.forEach((tp, offset) -> - this.durablyInsertedOffsets.merge(tp, offset, Math::max)); - // Remove the entry. - queryToRecordsMap.remove(topicName); - } + // queryToRecordsMap uses MutablePair keys, not String — remove was a no-op if (this.config.getBoolean( ClickHouseSinkConnectorConfigVariables. ENABLE_KAFKA_OFFSET.toString())) { @@ -790,12 +1062,10 @@ private boolean flushRecordsToClickHouse(String topicName, DbWriter writer, PreparedStatementExecutor preparedStatementExecutor) throws Exception { boolean result = false; - synchronized (queryToRecordsMap) { - result = preparedStatementExecutor.addToPreparedStatementBatch( - topicName, queryToRecordsMap, bmd, config, - writer.getConnection(), writer.getTableName(), - writer.getColumnNameToDataTypeMap(), writer.getEngine()); - } + result = preparedStatementExecutor.addToPreparedStatementBatch( + topicName, queryToRecordsMap, bmd, config, + writer.getConnection(), writer.getTableName(), + writer.getColumnNameToDataTypeMap(), writer.getEngine()); try { Metrics.updateMetrics(bmd); } catch (Exception e) { diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchWriter.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchWriter.java index 82bee9963..4100af96b 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchWriter.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchWriter.java @@ -7,8 +7,11 @@ import com.altinity.clickhouse.sink.connector.db.BaseDbWriter; import com.altinity.clickhouse.sink.connector.db.CacheInvalidationManager; import com.altinity.clickhouse.sink.connector.db.DBMetadata; +import com.altinity.clickhouse.sink.connector.db.DDLSchemaChangeWaiter; import com.altinity.clickhouse.sink.connector.db.DbKafkaOffsetWriter; import com.altinity.clickhouse.sink.connector.db.DbWriter; +import com.altinity.clickhouse.sink.connector.db.SourceSchemaIntegrityValidator; +import com.altinity.clickhouse.sink.connector.db.TableReplicationFreezeManager; import com.altinity.clickhouse.sink.connector.db.batch.GroupInsertQueryWithBatchRecords; import com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor; import com.altinity.clickhouse.sink.connector.model.BlockMetaData; @@ -70,6 +73,12 @@ public class ClickHouseBatchWriter { */ private Map topicToDbWriterMap; + /** + * DDL generation each cached DbWriter in {@link #topicToDbWriterMap} was built at, + * keyed by topic name. Used to detect schema changes applied since caching. + */ + private Map topicToDbWriterGeneration; + /** * Database credentials. */ @@ -102,9 +111,12 @@ public ClickHouseBatchWriter( } //this.queryToRecordsMap = new HashMap<>(); this.topicToDbWriterMap = new HashMap<>(); + this.topicToDbWriterGeneration = new HashMap<>(); //this.topicToRecordsMap = new HashMap<>(); this.dbCredentials = parseDBConfiguration(); - this.systemConnection = createConnection(BaseDbWriter.SYSTEM_DB); + // systemConnection is opened lazily — see getSystemConnection(). Opening + // it here made construction require a reachable ClickHouse even for + // operations that never touch the database. try { this.databaseOverrideMap = Utils.parseSourceToDestinationDatabaseMap( this.config.getString( @@ -230,26 +242,74 @@ public void persistRecords(List records) { // acknowledge the records. if (result) { log.info("****** Acknowledging records ******"); - records.forEach(record -> { + // Route through DebeziumOffsetManagement so this path uses the SAME + // OFFSET_COMMIT_LOCK as the batch-runnable path. Calling + // markProcessed()/markBatchFinished() directly here bypassed the + // serialization entirely and could drive a concurrent beginFlush() + // into the non-thread-safe OffsetStorageWriter. + for (ClickHouseStruct record : records) { + if (record.getCommitter() == null || record.getSourceRecord() == null) { + continue; + } try { - record.getCommitter().markProcessed( - record.getSourceRecord()); + DebeziumOffsetManagement.acknowledgeRecord( + record.getCommitter(), + record.getSourceRecord(), + record.isLastRecordInBatch()); } catch (InterruptedException e) { - //throw new RuntimeException(e); - log.error("Error marking records as processed" + e); + // Preserve the interrupt and stop acknowledging: silently + // continuing would advance offsets for records whose commit + // never completed. + Thread.currentThread().interrupt(); + log.error("Interrupted while acknowledging records", e); + throw new RuntimeException(e); } - if (record.isLastRecordInBatch()) { - try { - record.getCommitter().markBatchFinished(); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } - } - }); + // NOTE: markBatchFinished() is deliberately NOT re-invoked + // here. acknowledgeRecord() above already performs it, under + // the shared OFFSET_COMMIT_LOCK, when isLastRecordInBatch() + // is true. The PR-28 variant called it again outside that + // lock, which is precisely the unserialized second flush + // path that produced "OffsetStorageWriter is already + // flushing" (fixed in PR #31). It must stay removed. + } } } catch (Exception e) { - log.error("Error persisting records to ClickHouse" + e); + // A poisoned OffsetStorageWriter must NOT be absorbed here. Once its + // flush semaphore is leaked, offsets can never be committed again in + // this JVM, so logging and returning normally would let ClickHouse + // writes continue against a frozen binlog position -- silent + // divergence. Propagate so the task stops, matching the handling in + // ClickHouseBatchRunnable. + if (isOffsetWriterPoisoned(e)) { + log.error("FATAL: the Debezium OffsetStorageWriter is stuck in the " + + "'already flushing' state. Offsets can no longer be " + + "committed, so replication would keep writing rows while " + + "the binlog position stays frozen. Propagating to stop " + + "processing and prevent silent data divergence."); + throw new RuntimeException( + "OffsetStorageWriter is permanently stuck flushing; " + + "stopping to prevent silent data divergence", e); + } + log.error("Error persisting records to ClickHouse", e); + } + } + + /** + * Detects the unrecoverable "OffsetStorageWriter is already flushing" condition. + * + * @param e the exception thrown while persisting or acknowledging records. + * @return true when offset commits can no longer succeed in this JVM. + */ + private boolean isOffsetWriterPoisoned(Throwable e) { + Throwable current = e; + while (current != null) { + String message = current.getMessage(); + if (message != null && message.contains("OffsetStorageWriter is already flushing")) { + return true; + } + current = current.getCause(); } + return false; } /** @@ -283,53 +343,163 @@ public DbWriter getDbWriterForTable(String topicName, String tableName, String databaseName, ClickHouseStruct record, Connection connection) { - // Compare the cached writer's build version against the shared, monotonic - // table version. A mismatch means a DDL invalidated this table after the - // writer was built, so it must be rebuilt with the fresh schema. + DbWriter writer = null; String fullyQualifiedTableName = databaseName + "." + tableName; - long currentVersion = CacheInvalidationManager.getInstance() - .getVersion(fullyQualifiedTableName); - DbWriter writer = this.topicToDbWriterMap.get(topicName); - boolean invalidated = false; - if (writer != null) { - if (writer.getCacheInvalidationVersion() == currentVersion) { - return writer; + // Block while a DDL schema change is being applied + reconciled for + // this table, so inserts never run against a stale column cache that + // would silently drop newly added source columns. + // The return value MUST be honoured — see the matching comment in + // ClickHouseBatchRunnable.getDbWriterForTable. false means a DDL + // reconciliation is stuck; inserting anyway would use the stale cache + // this freeze exists to prevent. Return null so the batch is deferred. + boolean unfrozen = TableReplicationFreezeManager.getInstance() + .awaitUnfrozen(fullyQualifiedTableName, DDLSchemaChangeWaiter.DEFAULT_TIMEOUT_MS); + if (!unfrozen) { + log.error("Table {} still frozen after {}ms; deferring this batch rather " + + "than inserting against a possibly stale schema cache.", + fullyQualifiedTableName, DDLSchemaChangeWaiter.DEFAULT_TIMEOUT_MS); + return null; + } + + long generation = CacheInvalidationManager.getInstance() + .currentGeneration(fullyQualifiedTableName); + + if (this.topicToDbWriterMap.containsKey(topicName)) { + // Rebuild when EITHER signal fires — see the matching comment in + // ClickHouseBatchRunnable.getDbWriterForTable. Generation change + // means a DDL landed since this writer was cached; TTL expiry is + // the self-heal for a schema change that was missed entirely. + Long cachedGeneration = this.topicToDbWriterGeneration.get(topicName); + boolean ddlInvalidated = + cachedGeneration == null || cachedGeneration != generation; + boolean ttlExpired = CacheInvalidationManager.getInstance() + .isCacheExpired(fullyQualifiedTableName); + if (!ddlInvalidated && !ttlExpired) { + return this.topicToDbWriterMap.get(topicName); } - log.info("Invalidating cached DbWriter for {} after DDL (version {} -> {})", - topicName, writer.getCacheInvalidationVersion(), currentVersion); + log.info("Rebuilding cached DbWriter for {} ({}; generation {} -> {})", + topicName, + ddlInvalidated ? "DDL invalidation" : "cache TTL expiry", + cachedGeneration, generation); this.topicToDbWriterMap.remove(topicName); - invalidated = true; } writer = new DbWriter(this.dbCredentials.getHostName(), this.dbCredentials.getPort(), databaseName, tableName, this.dbCredentials.getUserName(), this.dbCredentials.getPassword(), this.config, record, connection); - writer.setCacheInvalidationVersion(currentVersion); this.topicToDbWriterMap.put(topicName, writer); - // Log the resolved schema whenever this table has seen a DDL (version > 0). - // This covers both rebuilding a stale writer and building a fresh writer at - // the current version after a burst of DDLs, so the post-DDL schema is always - // observable regardless of which thread ends up owning the writer. - if (invalidated || currentVersion > 0) { - logRefreshedColumns(topicName, writer, invalidated); + // Record the generation this writer was built at AND stamp the TTL + // clock, then verify the rebuilt cache covers every source column. + this.topicToDbWriterGeneration.put(topicName, generation); + CacheInvalidationManager.getInstance().markCacheBuilt(fullyQualifiedTableName); + if (!verifySourceSchemaIntegrity(fullyQualifiedTableName, record, writer)) { + // Column map does not cover every source column — inserting with + // this writer would silently DROP them. Evict and return null so + // the caller defers the batch; see the matching comment in + // ClickHouseBatchRunnable.getDbWriterForTable. + this.topicToDbWriterMap.remove(topicName); + this.topicToDbWriterGeneration.remove(topicName); + return null; } return writer; } /** - * Logs the refreshed column name and type map of a DbWriter that was rebuilt - * after a DDL cache invalidation. - * - * @param topicName the topic whose writer was rebuilt - * @param writer the freshly constructed DbWriter + * Verifies that every source-event column exists in the rebuilt destination + * cache; if not, logs loudly and re-marks the table for invalidation so the + * next batch rebuilds again rather than inserting lossy rows. Safety net on + * top of the per-table replication freeze. See {@link SourceSchemaIntegrityValidator}. */ - private void logRefreshedColumns(String topicName, DbWriter writer, boolean rebuilt) { - Map cols = writer.getColumnNameToDataTypeMap(); - if (cols != null) { - log.info("{} DbWriter schema for {} at cache version {} ({} columns): {}", - rebuilt ? "Rebuilt" : "Built", topicName, - writer.getCacheInvalidationVersion(), cols.size(), cols); + private boolean verifySourceSchemaIntegrity(String fullyQualifiedTableName, + ClickHouseStruct record, + DbWriter writer) { + try { + // The replication-history table is EXEMPT: it has its own fixed + // audit schema (gtid/ddl/before/after/...) and source-row columns + // are serialized into its payload columns, not mapped one-to-one. + // Comparing a source event against it reports every source column + // "missing" — the gate then blocks each batch for the full + // visibility-wait timeout polling system.columns for columns that + // will never appear, skips the history write, and starves the + // shared connection pool for the whole process. + // See SourceSchemaIntegrityValidator.isReplicationHistoryTable. + if (SourceSchemaIntegrityValidator.isReplicationHistoryTable( + this.config, fullyQualifiedTableName)) { + return true; + } + java.util.List sourceColumns = + com.altinity.clickhouse.sink.connector.db.SourceSchemaColumns.fromRecord(record); + if (sourceColumns.isEmpty() || writer == null + || writer.getColumnNameToDataTypeMap() == null) { + return true; + } + SourceSchemaIntegrityValidator.Result result = + SourceSchemaIntegrityValidator.check(sourceColumns, + writer.getColumnNameToDataTypeMap().keySet()); + if (!result.isConsistent()) { + String[] parts = fullyQualifiedTableName.split("\\.", 2); + // Source columns that map to destination ALIAS/MATERIALIZED + // columns are NOT missing: the insertable cache excludes them + // by design — ClickHouse computes their values and rejects + // inserts into them. MySQL generated columns land here on + // EVERY batch; without this filter the gate invalidated and + // rebuilt the writer forever, livelocking replication for the + // table. See the matching comment in ClickHouseBatchRunnable. + java.util.List genuinelyMissing = + result.getMissingInDestination(); + if (parts.length == 2 && writer.getConnection() != null) { + try { + java.util.Set generated = new DBMetadata(this.config) + .getAliasAndMaterializedColumnsForTableAndDatabase( + parts[1], parts[0], writer.getConnection()); + genuinelyMissing = SourceSchemaIntegrityValidator + .excludeGeneratedColumns(genuinelyMissing, generated); + } catch (Exception ex) { + log.warn("Could not fetch ALIAS/MATERIALIZED columns for {}: {}", + fullyQualifiedTableName, ex.getMessage()); + } + } + if (genuinelyMissing.isEmpty()) { + // Every "missing" column is computed by the destination — + // the writer is correct as built. Invalidating here is the + // livelock; the writer must be used as-is. + return true; + } + // Generalized visibility gate: WAIT for the destination to gain + // the missing columns rather than only logging. This covers + // RENAME/MODIFY COLUMN, which the ADD/DROP text parser in + // waitForSchemaVisibility cannot see. + java.util.Collection stillMissing = genuinelyMissing; + if (parts.length == 2 && writer.getConnection() != null) { + stillMissing = new DDLSchemaChangeWaiter() + .waitForExpectedColumns(writer.getConnection(), parts[0], + parts[1], genuinelyMissing); + } + if (stillMissing.isEmpty()) { + log.warn("Schema integrity for {}: source column(s) {} were missing " + + "but became visible while waiting; re-marking for " + + "invalidation so the writer is rebuilt.", + fullyQualifiedTableName, genuinelyMissing); + } else { + log.error("Schema integrity violation for {}: source column(s) {} " + + "still missing from the destination after waiting; " + + "inserting now would drop them. Re-marking for " + + "invalidation.", + fullyQualifiedTableName, stillMissing); + } + CacheInvalidationManager.getInstance().invalidateTable(fullyQualifiedTableName); + // Either way the CURRENT writer's column map predates those + // columns, so it must not be used for this batch. + return false; + } + return true; + } catch (Exception e) { + log.warn("Error during source schema integrity check for {}: {}", + fullyQualifiedTableName, e.getMessage()); + // Fail open on an unexpected checker error: the freeze and + // generation gates upstream are the primary protections. + return true; } } @@ -357,7 +527,23 @@ public ZoneId getServerTimeZone( if (userProvidedTimeZoneId != null) { return userProvidedTimeZoneId; } - return new DBMetadata(config).getServerTimeZone(this.systemConnection); + return new DBMetadata(config).getServerTimeZone(getSystemConnection()); + } + + /** + * Returns the shared system-database connection, opening it on first use. + * + *

Lazy so that constructing this writer does not require a reachable + * ClickHouse. Callers that genuinely need the database still fail loudly, + * because createConnection throws when it cannot connect.

+ * + * @return the system-database connection. + */ + private synchronized Connection getSystemConnection() { + if (this.systemConnection == null) { + this.systemConnection = createConnection(BaseDbWriter.SYSTEM_DB); + } + return this.systemConnection; } /** @@ -403,34 +589,38 @@ private boolean processRecordsByTopic(String topicName, Connection databaseConn = getClickHouseConnection(databaseName); DbWriter writer = getDbWriterForTable(topicName, tableName, databaseName, firstRecord, databaseConn); - PreparedStatementExecutor preparedStatementExecutor = - new PreparedStatementExecutor(writer. - getReplacingMergeTreeDeleteColumn(), - writer.isReplacingMergeTreeWithIsDeletedColumn(), - writer.getSignColumn(), writer.getVersionColumn(), - writer.getDatabaseName(), - getServerTimeZone(this.config)); - if (writer == null || writer.wasTableMetaDataRetrieved() == false) { + if (writer == null) { log.error(String.format( - "*** TABLE METADATA not retrieved for " + - "Database(%s), table(%s) retrying", - writer.getDatabaseName(), writer.getTableName())); + "*** DbWriter is null for Database(%s), table(%s) -- retrying", + databaseName, tableName)); + writer = getDbWriterForTable(topicName, tableName, + databaseName, firstRecord, databaseConn); if (writer == null) { - writer = getDbWriterForTable(topicName, tableName, - databaseName, firstRecord, databaseConn); + log.error(String.format( + "*** DbWriter still null for Database(%s), table(%s) -- giving up", + databaseName, tableName)); + return false; } - if (writer.wasTableMetaDataRetrieved() == false) - writer.updateColumnNameToDataTypeMap(); - if (writer == null || - writer.wasTableMetaDataRetrieved() == false) { + } + if (!writer.wasTableMetaDataRetrieved()) { + log.warn(String.format( + "*** TABLE METADATA not retrieved for Database(%s), table(%s) -- retrying", + writer.getDatabaseName(), writer.getTableName())); + writer.updateColumnNameToDataTypeMap(); + if (!writer.wasTableMetaDataRetrieved()) { log.error(String.format( - "*** TABLE METADATA not retrieved for " + - "Database(%s), table(%s), retrying on next " + - "attempt", + "*** TABLE METADATA not retrieved for Database(%s), table(%s) -- giving up", writer.getDatabaseName(), writer.getTableName())); return false; } } + PreparedStatementExecutor preparedStatementExecutor = + new PreparedStatementExecutor(writer. + getReplacingMergeTreeDeleteColumn(), + writer.isReplacingMergeTreeWithIsDeletedColumn(), + writer.getSignColumn(), writer.getVersionColumn(), + writer.getDatabaseName(), + getServerTimeZone(this.config)); // Step 1: The Batch Insert with preparedStatement in JDBC works by // forming the Query and then adding records to the Batch. // This step creates a Map of Query -> Records (List of @@ -438,11 +628,25 @@ private boolean processRecordsByTopic(String topicName, Map>, List> queryToRecordsMap = new HashMap<>(); Map partitionToOffsetMap = new HashMap<>(); + // Pass the writer's connector-managed column names (version/sign/ + // delete columns as actually named in the table engine) so the query + // formatter keeps them in the insert list while omitting genuine + // destination-only data columns the source event does not carry. + java.util.List managedColumns = new ArrayList<>(); + if (writer.getVersionColumn() != null) { + managedColumns.add(writer.getVersionColumn()); + } + if (writer.getSignColumn() != null) { + managedColumns.add(writer.getSignColumn()); + } + if (writer.getReplacingMergeTreeDeleteColumn() != null) { + managedColumns.add(writer.getReplacingMergeTreeDeleteColumn()); + } result = new GroupInsertQueryWithBatchRecords() .groupQueryWithRecords(records, queryToRecordsMap, partitionToOffsetMap, this.config, tableName, writer.getDatabaseName(), writer.getConnection(), - writer.getColumnNameToDataTypeMap()); + writer.getColumnNameToDataTypeMap(), managedColumns); BlockMetaData bmd = new BlockMetaData(); long maxBufferSize = this.config.getLong( ClickHouseSinkConnectorConfigVariables. diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/DebeziumOffsetManagement.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/DebeziumOffsetManagement.java index 2e53f8277..27c475e25 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/DebeziumOffsetManagement.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/executor/DebeziumOffsetManagement.java @@ -40,12 +40,18 @@ public class DebeziumOffsetManagement { completedBatches = new ConcurrentHashMap<>(); /** - * Shared lock that serializes every offset commit driven by the connector - * worker threads, guaranteeing a single connector-side flush at a time so - * that concurrent worker threads never issue overlapping - * {@code markBatchFinished()} calls against the same OffsetStorageWriter. + * Shared lock serialising every connector-driven offset commit. + *

+ * Kafka's {@code OffsetStorageWriter} is not thread-safe and rejects overlapping + * flushes with {@code ConnectException: OffsetStorageWriter is already flushing}. + * Debezium's own {@code RecordCommitter} methods are {@code synchronized}, but a + * NEW committer instance is built per batch + * ({@code EmbeddedEngine.buildRecordCommitter}), so concurrent worker threads + * synchronise on different monitors and get no mutual exclusion. This single + * static lock is the actual barrier. + *

*/ - private static final Object OFFSET_COMMIT_LOCK = new Object(); + static final Object OFFSET_COMMIT_LOCK = new Object(); /** * Constructor to initialize DebeziumOffsetManagement with a provided @@ -100,6 +106,9 @@ public Map, List> getBatchTimestamps() { */ public static Pair calculateMinMaxTimestampFromBatch( List batch) { + if (batch == null || batch.isEmpty()) { + return Pair.of(0L, 0L); + } long min = Long.MAX_VALUE; long max = Long.MIN_VALUE; for (ClickHouseStruct clickHouseStruct : batch) { @@ -170,17 +179,24 @@ static synchronized public boolean checkIfBatchCanBeCommitted( acknowledgeRecords(batch); result = true; // Check if completed batches can also be acknowledged. - completedBatches.forEach((k, v) -> { - if (false == checkIfThereAreInflightRequests(v)) { + // Collect keys first to avoid ConcurrentModificationException + // when removing entries during iteration + java.util.List> toRemove = new java.util.ArrayList<>(); + for (java.util.Map.Entry, java.util.List> entry + : completedBatches.entrySet()) { + if (false == checkIfThereAreInflightRequests(entry.getValue())) { try { - acknowledgeRecords(v); + acknowledgeRecords(entry.getValue()); } catch (InterruptedException e) { - log.error("*** Error acknowlegeRecords ***", e); + log.error("*** Error acknowledgeRecords ***", e); throw new RuntimeException(e); } - completedBatches.remove(k); + toRemove.add(entry.getKey()); } - }); + } + for (Pair key : toRemove) { + completedBatches.remove(key); + } } return result; } @@ -196,24 +212,37 @@ static synchronized public boolean checkIfBatchCanBeCommitted( * @param batch The batch of ClickHouseStruct records to acknowledge. * @throws InterruptedException If the commit operation is interrupted. */ - static synchronized void acknowledgeRecords(List batch) + static synchronized void acknowledgeRecords(List batch) throws InterruptedException { // acknowledge records // Iterate through the records // and use the record committer to commit the offsets. - for(ClickHouseStruct record: batch) { - if (record.getCommitter() != null && record.getSourceRecord() != null) { + // + // Both sides of the merge are kept, because they fix DIFFERENT halves of + // the same race: + // - develop widens the critical section: markProcessed() and + // markBatchFinished() MUST be inside the SAME one. Debezium builds a + // NEW RecordCommitter per batch (EmbeddedEngine.buildRecordCommitter), + // so its own `synchronized` methods lock different monitors for + // different batches and provide no mutual exclusion across worker + // threads. 2.10.0 locked only the markBatchFinished() call, leaving + // markProcessed() outside the barrier. + // - 2.10.0 routes the finish through markBatchFinishedSafely(), which + // null-guards the committer and keeps every finish path funnelled + // through one helper. The helper re-acquires OFFSET_COMMIT_LOCK; that + // is safe because Java monitors are reentrant. + synchronized (OFFSET_COMMIT_LOCK) { + for (ClickHouseStruct record : batch) { + if (record.getCommitter() != null && record.getSourceRecord() != null) { - record.getCommitter().markProcessed(record.getSourceRecord()); -// log.debug("***** Record successfully marked as processed ****" + "Binlog file:" + -// record.getFile() + " Binlog position: " + record.getPos() + " GTID: " + record.getGtid() -// + "Sequence Number: " + record.getSequenceNumber() + "Debezium Timestamp: " + record.getDebezium_ts_ms()); + record.getCommitter().markProcessed(record.getSourceRecord()); - if(record.isLastRecordInBatch()) { - markBatchFinishedSafely(record.getCommitter()); - log.info("***** BATCH marked as processed to debezium ****" + "Binlog file:" + - record.getFile() + " Binlog position: " + record.getPos() + " GTID: " + record.getGtid() - + " Sequence Number: " + record.getSequenceNumber() + " Debezium Timestamp: " + record.getDebezium_ts_ms()); + if (record.isLastRecordInBatch()) { + markBatchFinishedSafely(record.getCommitter()); + log.info("***** BATCH marked as processed to debezium ****" + "Binlog file:" + + record.getFile() + " Binlog position: " + record.getPos() + " GTID: " + record.getGtid() + + " Sequence Number: " + record.getSequenceNumber() + " Debezium Timestamp: " + record.getDebezium_ts_ms()); + } } } } @@ -239,8 +268,48 @@ public static synchronized void acknowledgeRecords( boolean lastRecordInBatch) throws InterruptedException { if (sourceRecord != null) { + // Same critical section as the batch variant above — see the comment there. + synchronized (OFFSET_COMMIT_LOCK) { + recordCommitter.markProcessed(sourceRecord); + if (lastRecordInBatch) { + // Every finish path goes through the null-guarded helper, so + // no caller can reach markBatchFinished() outside the lock. + markBatchFinishedSafely(recordCommitter); + } + } + } + } + + /** + * Acknowledges a single record on the shared offset-commit lock. + *

+ * Exposed so that every offset-committing path in the connector funnels through + * the SAME lock. Any path that calls {@code markProcessed()} / + * {@code markBatchFinished()} directly bypasses the serialization and can drive + * concurrent {@code beginFlush()} calls into the non-thread-safe + * OffsetStorageWriter, which throws + * {@code ConnectException: OffsetStorageWriter is already flushing}. + *

+ * + * @param recordCommitter The record committer to be used. + * @param sourceRecord The source record to mark as processed. + * @param lastRecordInBatch True if this is the last record in the batch. + * @throws InterruptedException If the commit operation is interrupted. + */ + public static void acknowledgeRecord( + DebeziumEngine.RecordCommitter> + recordCommitter, + ChangeEvent sourceRecord, + boolean lastRecordInBatch) + throws InterruptedException { + if (recordCommitter == null || sourceRecord == null) { + return; + } + synchronized (OFFSET_COMMIT_LOCK) { recordCommitter.markProcessed(sourceRecord); - if (lastRecordInBatch == true) { + if (lastRecordInBatch) { + // Funnel every finish through the null-guarded helper (2.10.0) + // while staying inside develop's widened critical section. markBatchFinishedSafely(recordCommitter); } } diff --git a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/CacheInvalidationManagerTest.java b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/CacheInvalidationManagerTest.java new file mode 100644 index 000000000..4739f20fd --- /dev/null +++ b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/CacheInvalidationManagerTest.java @@ -0,0 +1,103 @@ +package com.altinity.clickhouse.sink.connector.db; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * Regression tests for {@link CacheInvalidationManager}. + * + *

The connector runs a pool of worker threads, each holding its own + * topicToDbWriterMap cache. The previous remove-on-read implementation let the + * FIRST reader consume the invalidation signal, so every other thread kept + * serving a DbWriter built against the pre-DDL column list. Those writers build + * the INSERT column list from a snapshot taken at construction, so the newly + * added column was silently dropped from every subsequent INSERT and ClickHouse + * diverged from MySQL permanently.

+ */ +public class CacheInvalidationManagerTest { + + private static final String TABLE = "testdb.orders"; + + @BeforeEach + public void reset() { + CacheInvalidationManager.getInstance().clearAll(); + } + + @Test + public void generationStartsAtZeroForUnknownTable() { + Assertions.assertEquals(0L, + CacheInvalidationManager.getInstance().currentGeneration(TABLE)); + } + + @Test + public void invalidationIsVisibleToEveryConsumerNotJustTheFirst() { + CacheInvalidationManager manager = CacheInvalidationManager.getInstance(); + + // Two worker threads each cached a DbWriter at generation 0. + long threadOneCachedAt = manager.currentGeneration(TABLE); + long threadTwoCachedAt = manager.currentGeneration(TABLE); + + // ALTER TABLE ... ADD COLUMN arrives. + manager.invalidateTable(TABLE); + + long generationAfterDdl = manager.currentGeneration(TABLE); + + // BOTH threads must observe that their cached writer is stale. Under the + // old remove-on-read behaviour only the first caller saw the signal. + Assertions.assertNotEquals(threadOneCachedAt, generationAfterDdl, + "first worker thread must rebuild its DbWriter after DDL"); + Assertions.assertNotEquals(threadTwoCachedAt, generationAfterDdl, + "second worker thread must ALSO rebuild its DbWriter after DDL"); + } + + @Test + public void generationIsStableWhenNoDdlOccurs() { + CacheInvalidationManager manager = CacheInvalidationManager.getInstance(); + manager.invalidateTable(TABLE); + + long cachedAt = manager.currentGeneration(TABLE); + + // No further DDL: repeated reads must not invalidate the cache, otherwise + // every insert would rebuild the writer and re-query the schema. + Assertions.assertEquals(cachedAt, manager.currentGeneration(TABLE)); + Assertions.assertEquals(cachedAt, manager.currentGeneration(TABLE)); + } + + @Test + public void successiveDdlsProduceDistinctGenerations() { + CacheInvalidationManager manager = CacheInvalidationManager.getInstance(); + + manager.invalidateTable(TABLE); + long afterFirst = manager.currentGeneration(TABLE); + manager.invalidateTable(TABLE); + long afterSecond = manager.currentGeneration(TABLE); + + Assertions.assertNotEquals(afterFirst, afterSecond, + "a second DDL must invalidate caches rebuilt after the first"); + } + + @Test + public void tablesAreTrackedIndependently() { + CacheInvalidationManager manager = CacheInvalidationManager.getInstance(); + + long otherCachedAt = manager.currentGeneration("testdb.customers"); + manager.invalidateTable(TABLE); + + Assertions.assertEquals(otherCachedAt, + manager.currentGeneration("testdb.customers"), + "DDL on one table must not invalidate caches for another"); + } + + @Test + public void nullAndEmptyNamesAreIgnored() { + CacheInvalidationManager manager = CacheInvalidationManager.getInstance(); + + manager.invalidateTable(null); + manager.invalidateTable(""); + + Assertions.assertEquals(0L, manager.currentGeneration(null)); + Assertions.assertEquals(0L, manager.currentGeneration("")); + Assertions.assertEquals(0, manager.pendingInvalidations()); + } +} diff --git a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/CacheInvalidationManagerTtlTest.java b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/CacheInvalidationManagerTtlTest.java new file mode 100644 index 000000000..58b883578 --- /dev/null +++ b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/CacheInvalidationManagerTtlTest.java @@ -0,0 +1,106 @@ +package com.altinity.clickhouse.sink.connector.db; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests for the per-table schema cache TTL added to + * {@link CacheInvalidationManager}. The TTL ensures the metadata cache is never + * kept forever, so any missed schema change self-heals within one TTL window. + */ +class CacheInvalidationManagerTtlTest { + + private CacheInvalidationManager mgr; + private long originalTtl; + + @BeforeEach + void setUp() { + mgr = CacheInvalidationManager.getInstance(); + originalTtl = mgr.getCacheTtlMs(); + mgr.clearAll(); + } + + @AfterEach + void tearDown() { + mgr.clearAll(); + mgr.setCacheTtlMs(originalTtl); + } + + @Test + @DisplayName("Default TTL is one hour") + void defaultTtlIsOneHour() { + assertEquals(60L * 60L * 1000L, CacheInvalidationManager.DEFAULT_CACHE_TTL_MS); + } + + @Test + @DisplayName("Never-built table is treated as expired (forces first build)") + void unbuiltTableIsExpired() { + mgr.setCacheTtlMs(60_000); + assertTrue(mgr.isCacheExpired("db.fresh")); + } + + @Test + @DisplayName("Freshly built table is not expired before TTL elapses") + void freshlyBuiltNotExpired() { + mgr.setCacheTtlMs(60_000); + mgr.markCacheBuilt("db.t"); + assertFalse(mgr.isCacheExpired("db.t")); + } + + @Test + @DisplayName("Cache expires once the TTL elapses") + void expiresAfterTtl() throws Exception { + mgr.setCacheTtlMs(150); + mgr.markCacheBuilt("db.t"); + assertFalse(mgr.isCacheExpired("db.t")); + Thread.sleep(250); + assertTrue(mgr.isCacheExpired("db.t"), "Cache should be expired after the TTL window"); + } + + @Test + @DisplayName("Rebuild resets the TTL clock") + void rebuildResetsClock() throws Exception { + mgr.setCacheTtlMs(300); + mgr.markCacheBuilt("db.t"); + Thread.sleep(200); + // Rebuild before expiry -> clock resets. + mgr.markCacheBuilt("db.t"); + Thread.sleep(200); + assertFalse(mgr.isCacheExpired("db.t"), "Total 400ms but reset at 200ms -> not expired"); + } + + @Test + @DisplayName("TTL <= 0 disables expiry (DDL-triggered invalidation still works)") + void ttlDisabled() { + mgr.setCacheTtlMs(0); + // Never-built normally counts as expired, but expiry is disabled. + assertFalse(mgr.isCacheExpired("db.t")); + // DDL invalidation path is unaffected. shouldInvalidate() was replaced + // by a monotonic generation counter (the old remove-on-read Set let the + // FIRST worker thread consume the signal, leaving every other thread + // serving a stale writer), so the equivalent assertion is that the + // generation advances. + long before = mgr.currentGeneration("db.t"); + mgr.invalidateTable("db.t"); + assertNotEquals(before, mgr.currentGeneration("db.t")); + } + + @Test + @DisplayName("DDL invalidation and TTL are independent signals") + void invalidationIndependentOfTtl() { + mgr.setCacheTtlMs(60_000); + mgr.markCacheBuilt("db.t"); + assertFalse(mgr.isCacheExpired("db.t")); + long before = mgr.currentGeneration("db.t"); + mgr.invalidateTable("db.t"); + assertNotEquals(before, mgr.currentGeneration("db.t"), + "DDL invalidation fires even when TTL is not expired"); + } +} diff --git a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/executor/DebeziumOffsetManagementTest.java b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/executor/DebeziumOffsetManagementTest.java index 568307c1f..bdf83b2f0 100644 --- a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/executor/DebeziumOffsetManagementTest.java +++ b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/executor/DebeziumOffsetManagementTest.java @@ -13,6 +13,7 @@ import org.apache.kafka.connect.source.SourceRecord; import org.junit.Assert; import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import java.util.ArrayList; @@ -256,4 +257,81 @@ public static Struct getKafkaStruct() { return kafkaConnectStruct; } + + @Test + @DisplayName("calculateMinMaxTimestampFromBatch returns (0,0) for empty batch") + public void testCalculateMinMaxTimestampFromBatchEmpty() { + // Before the fix, an empty batch returned (Long.MAX_VALUE, Long.MIN_VALUE) + // which is an inverted range that could cause downstream issues + List emptyBatch = new ArrayList<>(); + Pair result = DebeziumOffsetManagement.calculateMinMaxTimestampFromBatch(emptyBatch); + Assert.assertEquals("Min should be 0 for empty batch", Long.valueOf(0L), result.getLeft()); + Assert.assertEquals("Max should be 0 for empty batch", Long.valueOf(0L), result.getRight()); + } + + + @Test + @DisplayName("calculateMinMaxTimestampFromBatch returns (0,0) for null batch") + public void testCalculateMinMaxTimestampFromBatchNull() { + Pair result = DebeziumOffsetManagement.calculateMinMaxTimestampFromBatch(null); + Assert.assertEquals("Min should be 0 for null batch", Long.valueOf(0L), result.getLeft()); + Assert.assertEquals("Max should be 0 for null batch", Long.valueOf(0L), result.getRight()); + } + + + @Test + @DisplayName("calculateMinMaxTimestampFromBatch handles single-element batch") + public void testCalculateMinMaxTimestampFromBatchSingleElement() { + List batch = new ArrayList<>(); + ClickHouseStruct ch = new ClickHouseStruct(10, "SERVER5432.test.t", getKafkaStruct(), + 2, 42L, null, getKafkaStruct(), null, ClickHouseConverter.CDC_OPERATION.CREATE); + ch.setDebezium_ts_ms(500L); + batch.add(ch); + + Pair result = DebeziumOffsetManagement.calculateMinMaxTimestampFromBatch(batch); + Assert.assertEquals("Min should equal the single element", Long.valueOf(500L), result.getLeft()); + Assert.assertEquals("Max should equal the single element", Long.valueOf(500L), result.getRight()); + } + + + @Test + @DisplayName("checkIfBatchCanBeCommitted handles concurrent batch tracking without ConcurrentModificationException") + public void testCheckIfBatchCanBeCommittedConcurrentSafety() { + // Clear any previous state + DebeziumOffsetManagement.inFlightBatches.clear(); + + // Add several batches + for (int i = 0; i < 10; i++) { + List batch = new ArrayList<>(); + ClickHouseStruct ch = new ClickHouseStruct(i, "SERVER5432.test.t", getKafkaStruct(), + 2, (long) i, null, getKafkaStruct(), null, ClickHouseConverter.CDC_OPERATION.CREATE); + ch.setDebezium_ts_ms((long) (i * 100)); + batch.add(ch); + DebeziumOffsetManagement.addToBatchTimestamps(batch); + } + + // This should not throw ConcurrentModificationException + // The fix uses collect-then-remove pattern instead of forEach+remove + try { + List testBatch = new ArrayList<>(); + ClickHouseStruct ch = new ClickHouseStruct(99, "SERVER5432.test.t", getKafkaStruct(), + 2, 99L, null, getKafkaStruct(), null, ClickHouseConverter.CDC_OPERATION.CREATE); + ch.setDebezium_ts_ms(50L); + testBatch.add(ch); + DebeziumOffsetManagement.checkIfBatchCanBeCommitted(testBatch); + // Success — no exception thrown + } catch (java.util.ConcurrentModificationException e) { + Assert.fail("checkIfBatchCanBeCommitted should not throw ConcurrentModificationException"); + } catch (InterruptedException e) { + // checkIfBatchCanBeCommitted commits offsets and is declared to + // throw InterruptedException. Restore the flag and fail rather than + // swallowing it, so an interrupted run never looks like a pass. + Thread.currentThread().interrupt(); + Assert.fail("Interrupted while checking batch commit: " + e); + } + + // Clean up + DebeziumOffsetManagement.inFlightBatches.clear(); + } + }