Repository navigation
Conversation
… DDL/DML race fix) (#1435) * add postgres * checksum for postgres * remove * clean up * clean up * update the dump script * build the python into a package * reinstate the lic * change back snapshot mode to initial * remove the unecessary files * sanitize the repo * Fix comments * fix the comments * centralize mapping * db name quoting issue * update the view to be compatible with postgres * fix the CI * fix the CI * fix kafka test * fix test * fix git work * address PR review comments from Kanthi (round 2) - Consolidate verbose per-type test methods into parameterized tests using @MethodSource in ClickHouseDataTypeMapperDDLTest and PostgresSchemaReconcilerTest (review comments on test verbosity) - Remove MariaDB test constant and MariaDB-specific tests from DDLParserFactoryTest; remove duplicate MySQL test - Remove plans/postgres/ directory (internal development docs not suitable for the repository) - Add junit-jupiter-params dependency to sink-connector-lightweight * use ClickHouseDbConstants in PostgresSchemaReconciler (PR review round 2) * fix CI test failures * fix the loader * change the dump * fix the nullable type ingestion * upgrade the version to match both side * fix the failing test * totel fix to make it work * auto create the heartbeat table option in dumper to prepare for the sink-connector * dumper check the user condition * ignore the internal directory * ignore the internal directory * fix the tests * new naming stuff * implement column type override * fix the test * update the feature * fix(test): PostgresSchemaDriftIT — isolate tables and replication slots per test Root cause: both test methods shared a single table (schema_drift_test) and the default replication slot name. When the first test finished and the second started, the leftover slot caused 'Failed to start replication stream' after 6 retries, which meant no CDC events flowed and the drift detector never ran. Changes: - Split into schema_drift_single / schema_drift_multi tables so ALTER TABLE in one test does not pollute the other. - Assign unique slot.name per test (drift_slot_single / drift_slot_multi). - Increase slot.max.retries from 6→12 and slot.retry.delay.ms from 5000→10000 for slow CI runners. - Drop replication slot before and after each test to prevent leftovers. - Add null-guard in PostgresSchemaChangeDetector.fetchClickHouseSchema() so a null writer logs a warning instead of throwing NPE. * fix(test): INSERT before waiting for drift column — drift detection is event-driven The schema drift detector (checkAndReconcile) only fires when a DML SourceRecord arrives in processEveryChangeRecord(). The previous test waited for the column to appear in ClickHouse BEFORE inserting a row, creating a deadlock: no DML event → no drift detection → column never appears → test times out. Fix: INSERT immediately after ALTER TABLE so the CDC event triggers drift detection, then poll for both the column and the row data. * fix(test): replace fixed Thread.sleep with polling-based snapshot wait Test 1 (testColumnAddedToPostgresIsPropagatedToClickHouse) failed in CI at the initial snapshot verification step because the Debezium engine encountered a Producer failure (PSQLException: I/O error) and could not complete the snapshot within the fixed 60s Thread.sleep window. Root cause: In CI, Test 2 runs first and passes. By the time Test 1 starts, the shared PostgreSQL container may have stale connections or resource pressure, causing the snapshot to take longer than 60s. Fix: Replace the rigid Thread.sleep(60_000) with a polling loop that checks every 5 seconds for up to 120 seconds whether the initial row has appeared in ClickHouse. This is: - More robust: adapts to varying snapshot completion times - Faster on success: returns immediately when the row appears - Better diagnostics: logs each poll attempt for debugging - Consistent: applied to both Test 1 and Test 2 Test 2 (multi-column) already passed in CI run 22839813547, confirming the drift detection INSERT-before-wait fix works correctly. * fix(test): use unique offset storage tables per test to prevent cross-contamination Root cause: Both test methods in PostgresSchemaDriftIT share the same ClickHouse offset storage table (altinity_sink_connector.replica_source_info) and schema history table. When Test 2 runs first and completes, its committed offsets (snapshot=FALSE, LSN=0/1A25218) remain in the shared table. When Test 1 starts, the Debezium engine reads Test 2's offsets, sees snapshot=FALSE, and skips the initial snapshot entirely — going straight to streaming from an LSN that is invalid for Test 1's replication slot. This causes: 1. 'Last recorded offset is no longer available on the server' 2. PostgreSQL postmaster crash: 'FATAL: terminating connection due to unexpected postmaster exit' 3. No snapshot = no initial row replicated = assertion failure Fix: Give each test its own unique offset storage table name and schema history table name, keyed by the slot name (which is already unique per test): - Test 1: replica_source_info_drift_slot_single - Test 2: replica_source_info_drift_slot_multi This ensures each test starts with a clean offset state and always performs its own initial snapshot. * revert the images * fix(test): remove obsolete ClickHouse 22.3 tests incompatible with Testcontainers 1.21.4 ReplicatedRMTClickHouse22TIT and ReplicatedRMTDDLClickHouse22TIT both fail with IllegalStateException because clickhouse-server:22.3 is no longer recognized as a compatible substitute by the Testcontainers ClickHouse module (v1.21.4). ClickHouse 22.3 reached end-of-life. The same ReplicatedReplacingMergeTree auto-creation functionality is already covered by: - ReplicatedRMTIT (CH 24.8) - passes - ReplicatedRMTDDLIT (CH 23.8) - passes Removing these redundant tests eliminates the last 2 errors in the CI pipeline (Tests run: 288, Failures: 0, Errors: 0 expected after this). * feat: add clickhouse.table.schema.prefix config for schema-prefixed table naming Add config option clickhouse.table.schema.prefix (boolean, default false) that prefixes ClickHouse table names with __schema__ derived from the Debezium topic. When enabled, a topic like 'prefix.public.MyTable' creates table '__public__MyTable'. Modified files: - ClickHouseSinkConnectorConfigVariables: new enum constant - ClickHouseSinkConnectorConfig: config registration (Type.BOOLEAN) - Utils: overloaded getTableNameFromTopic(String, boolean) - DebeziumChangeEventCapture: schemaPrefixEnabled field, getTableName(), extractTableNameFromTopic(), DDL cache, schema drift detection - ClickHouseBatchRunnable: getTableFromTopic() reads config Validated on fpif-metrics-dev-ch01: 19 tables with __public__ prefix, 272,379 rows checksummed against PostgreSQL source - 100% match. * feat: add template-based schema naming for database suffix and table prefix Add two new config options for flexible schema-aware naming: - clickhouse.database.schema.suffix (boolean, default false): when enabled, appends the resolved common schema template to the ClickHouse database name - clickhouse.common.schema.template (string, default empty): shared template with {{ schema }} placeholder used by both table prefix and database suffix Example: with template '__{{ schema }}__' and PG schema 'public': - Database suffix enabled: 'app' -> 'app__public__' - Table prefix enabled: 'orders' -> '__public__orders' The common template is shared between clickhouse.table.schema.prefix (boolean) and clickhouse.database.schema.suffix (boolean), providing a single template definition for consistent naming across both features. New utility methods in Utils: - extractSchemaFromTopic(): extracts schema from Debezium topic - resolveSchemaTemplate(): replaces {{ schema }} placeholder - applyDatabaseSchemaSuffix(): applies suffix to database name - getTableNameFromTopic(String, boolean, String): 3-arg template-aware version * delete plan * move plan out * feat: add clickhouse.common.database.prefix config for multi-connector isolation Add clickhouse.common.database.prefix (string, default empty) that prepends a static prefix to the ClickHouse database name, enabling multiple sink-connectors to write to the same ClickHouse instance without conflicts when source databases share the same name. Only alphanumeric characters and underscores allowed ([a-zA-Z0-9_]*), validated at startup with IllegalArgumentException on failure. Application order: prefix + database + suffix Example: prefix='litellm_dev_' + db='app' + suffix='__public__' Result: 'litellm_dev_app__public__' New utility methods: - Utils.isValidDatabasePrefix(): validates allowed characters - Utils.applyDatabasePrefix(): prepends prefix to database name - Utils.applyDatabaseNaming(): combined prefix + suffix application * fix: apply database prefix/suffix in DML batch write path + schema-aware naming tests Production fix: - ClickHouseBatchRunnable.processRecordsByTopic(): Apply database prefix (clickhouse.common.database.prefix) and database schema suffix (clickhouse.database.schema.suffix + clickhouse.common.schema.template) to the raw database name from ClickHouseStruct.getDatabase(). Previously only the DDL/schema-drift path in DebeziumChangeEventCapture applied these transformations; the DML write path used the raw Debezium source.db value, causing data to be written to the wrong database. Unit tests (37 tests, all passing): - UtilsSchemaAwareTest.java: Tests extractSchemaFromTopic(), resolveSchemaTemplate(), getTableNameFromTopic(), isValidDatabasePrefix(), applyDatabasePrefix(), applyDatabaseSchemaSuffix(), applyDatabaseNaming() with various inputs including null/empty edge cases. E2E integration test (1 test, passing): - PostgresSchemaAwareNamingIT.java: Enables all 4 naming configs (table schema prefix, common schema template, database schema suffix, database prefix) and verifies PostgreSQL public.tm replicates to ClickHouse dev_public__public__.__public__tm with correct row count. - init_clickhouse_schema_naming.sql: Pre-creates the expected database. * improve postgres dumper * pausing logic on replica * fix vairous bugs * fix the view * Fix the bugs * fix the duplicate entry * Fix the failing tests * Fix testSourceDirectory and update DateTime64 test expectations - Fix testSourceDirectory in sink-connector/pom.xml: changed from src/test to src/test/java to prevent java.lang.SecurityException (prohibited package name java.*) - Update ClickHouseDataTypeMapperDDLTest: DateTime64 expectations now include UTC timezone - Update ClickHouseTableOperationsBaseTest: DateTime64 expectations now include UTC timezone * fix bugs * Refactor UtilsSchemaAwareTest to parameterized tests; add null guard for heartbeat events - Consolidate 30 individual @Test methods into 7 @ParameterizedTest methods with @MethodSource in UtilsSchemaAwareTest.java (addresses review comment #22) - Add null check for chStruct in DebeziumChangeEventCapture.java:1184 to skip Debezium internal heartbeat events that produce null ClickHouseStruct * refactor: address review feedback — extract utilities, use constants, delegate to DBMetadata * refactor: address review feedback round 2 — restore tests, extract PostgresConnectorConfig, move extractDebeziumSchema to ClickHouseConverter, reduce test timeouts - Restore ReplicatedRMTClickHouse22TIT and ReplicatedRMTDDLClickHouse22TIT tests from origin/2.9.1, upgrading ClickHouse Docker image from 22.3 to 24.8 - Move extractDebeziumSchema() and safeGet() from PostgresSchemaChangeDetector to ClickHouseConverter as public static methods; PostgresSchemaChangeDetector now delegates to ClickHouseConverter.extractDebeziumSchema() - Extract PostgresConnectorConfig class from DebeziumChangeEventCapture to encapsulate postgres-specific fields (schemaPrefixEnabled, databaseSchemaSuffix, commonSchemaTemplate, commonDatabasePrefix, postgresSchemaChangeDetector) and their initialization logic; removes TODO comment - Reduce AlterTableModifyColumnIT retry count from 40 to 10 (max wait from 200s to 50s) * Move sink-connector/python to separate branch (feature/ch-sink-tools-python) Restore sink-connector/python to match 2.9.1 so PR #1245 focuses on Postgres/Java connector work. Python toolkit changes live on feature/ch-sink-tools-python for a dedicated PR. Made-with: Cursor * fix: serialize PG text[] arrays as JSON strings when CH column is String type When PostgreSQL text[] columns are auto-created as Nullable(String) in ClickHouse, the ARRAY branch in PreparedStatementFieldMapper calls setArray() which triggers ClassCastException. Now checks the CH column type first — if it is not Array(), serializes the array value as a JSON string using setString(). Fixes ClassCastException on LiteLLM_VerificationToken table for columns: models, allowed_cache_controls, allowed_routes, policies, access_group_ids * fix: handle PostgreSQL db field name in getDatabaseName() to avoid DataException PostgreSQL connectors use "db" in the source struct, not "databaseName" (MySQL). Check schema.field() existence before calling Struct.get() to avoid DataException on every record during schema drift detection. * fix: repair two rebase merge errors that broke the build The rebase onto 2.10.0 merged the DDL cache-invalidation block in DebeziumChangeEventCapture twice. The final conflict resolution replaced the leading database-resolution statements but left the earlier merged loop behind, producing four compile errors: schemaPrefixEnabled -> moved to pgConfig.isSchemaPrefixEnabled() commonSchemaTemplate -> moved to pgConfig.getCommonSchemaTemplate() tableName -> redeclared in the same method invalidationDatabaseName -> declaration had been dropped Rewrite the block to the intended merge: the release branch's override-map-aware database resolution and multi-table invalidation, plus this branch's topic-derived fallback for schema-change events that name no table and its schema-drift detector invalidation, both reached through pgConfig. Also align the BinLogHistoryTest expectations with the producer. An earlier commit on this branch changes the _time column from DateTime64(0) to DateTime64(9); the rebase kept the release branch's DateTime64(0) expectation against it, so both assertions failed. The backticked database name and the db_time column in those expectations are correct and are left as they are. Verified with a real JDK 17 build: mvn -pl sink-connector-lightweight -am compile now reports BUILD SUCCESS, and BinLogHistoryTest passes 2/2. * test: the Postgres ITs assert a timestamp type the connector never emits Standing up the real stack for this branch -- debezium/postgres:15-alpine -> sink connector -> clickhouse-server:24.8 -- PostgresInitialDockerIT fails at the `created` column assertion. The connector emits: CREATE TABLE `public`.`tm`(... `set_date` Nullable(DateTime64(6, 'UTC')), `created` Nullable(DateTime64(6, 'UTC')), `updated` Nullable(DateTime64(6, 'UTC')), ...) while the test asserts Nullable(DateTime64(6)). The test has never matched this branch. The production code is right and is not touched here. Debezium timestamps are UTC by definition, and commit 78f2618d tagged the zone in ClickHouseDataTypeMapper.mapDebeziumSchemaToDDL -- the record-schema auto-create path, which is the path PostgreSQL uses. Dropping the zone would make an instant look like a local wall-clock time to every reader. Only the expectations were left behind. Scope is deliberately the record-schema path only. The MySQL DDL path resolves its type through DataTypeConverter, not this mapper, so CreateTableDataTypesIT and friends still legitimately expect a bare DateTime64 and are untouched. Five assertions, one per Postgres IT, all on the same `created` column: ClickHouseDebeziumEmbeddedPostgresDecoderBufsDockerIT, ClickHouseDebeziumEmbeddedPostgresPgoutputDockerIT, PostgresInitialDockerIT, PostgresInitialDockerWKeeperMapStorageIT, PostgresPgoutputMultipleSchemaIT. No production file changes. Verification: - PostgresInitialDockerIT on real containers via a rootless podman socket: 0/1 -> 1/1. The run reaches Snapshot completed and auto-creates the tables, so the assertion is measured against DDL the connector actually produced, not against a fixture. - Not merge fallout, proven by A/B: the same IT fails identically at the same line on the untouched pre-merge head 4fb183f5, in a separate worktree against the same podman socket. Commit 78f2618d is on this branch and in neither base. The unit-suite counterpart of this same defect -- three PostgresSchemaReconcilerTest timestamp cases expecting the bare form -- is left for the branch owner, as noted on the PR: it is the same question of which side is authoritative, and the answer there may reasonably be to align the reconciler instead. * db_compare: MySQL/ClickHouse checksum column-set asymmetry (approach under revision) (#1429) * Added release notes for 2.10.1 release * fix: /metrics served with no Content-Type, breaking Prometheus 3.x (#1096) (#1413) The /metrics handler wrote its body without ever setting a Content-Type header. Prometheus 3.x treats that as a non-compliant target and fails the scrape outright: non-compliant scrape target sending blank Content-Type and no fallback_scrape_protocol specified for target so the connector silently stopped being monitored after a Prometheus upgrade. Set the Prometheus text exposition content type, "text/plain; version=0.0.4; charset=utf-8", before sendResponseHeaders, which is the call that commits the headers. Also encode the body once instead of calling response.getBytes() twice. That is a real correctness bug, not tidying: the two calls each used the platform default charset, so the length declared in the headers and the bytes actually written could differ for non-ASCII metric labels under a non-UTF-8 default charset, leaving a scraper short of the promised Content-Length. Encoding once to UTF-8 and using that array length makes the declared length true and matches the charset now advertised. Regression test MetricsContentTypeTest starts the real metrics server on a free port and scrapes it over HTTP, since the header is only observable on the wire. Verified failing first: Content-Type was null ("expected: not <null>"); it is the expected value after the fix. The Content-Length assertion passed before the fix too -- the current scrape output is ASCII -- and is there to pin the invariant, not to reproduce a live failure. * fix: INT8 (Byte) hard-cast to Integer fails every TINYINT batch (#385) (#1414) convert() groups INT8 and INT32 into a single isFieldTypeInt branch whose terminal case was: ps.setInt(index, (Integer) value); An INT8 schema delivers a java.lang.Byte, never an Integer, so this threw ClassCastException: class java.lang.Byte cannot be cast to class java.lang.Integer as reported -- and because it aborts the batch, the connector could not replicate any table holding a MySQL TINYINT. The isWiderIntegerTarget hatch immediately above does not cover it: it only fires for Int32/UInt32/Int64/UInt64 ClickHouse targets, and an INT8 source column maps to ClickHouse Int8 (Int32_SCHEMA -> Int8 in dataTypesMap), or UInt8 when the column is unsigned. Read the value as a Number and bind intValue(), so the boxed type no longer has to be guessed. Byte, Short and Integer all widen correctly and keep their sign. For the INT32 case that works today this is a no-op: Integer.intValue() returns the same int the cast produced, so the same setInt call with the same value is issued. Regression test ClickHouseDataTypeMapperInt8Test pins TINYINT to Int8 and UInt8 targets and negative/boundary bytes (-1, MIN_VALUE, MAX_VALUE), with two controls: INT32 still binds identically via setInt, and the isWiderIntegerTarget path still routes to setObject. Verified failing first: the 3 INT8 tests errored with the ClassCastException above while both controls passed; all 5 pass after the fix. * fix: ARRAY binding hard-casts to ArrayList, breaking every other List (#749) (#1415) The ARRAY branch of ClickHouseDataTypeMapper.convert cast the incoming value to java.util.ArrayList: ps.setArray(index, ps.getConnection().createArrayOf( dt.name(), ((ArrayList) value).toArray())); But the Kafka Connect contract for an array field is java.util.List, not ArrayList. Debezium and the Connect converters may hand over any List implementation, and an empty array in particular arrives as Collections.emptyList(). The cast then throws ClassCastException: class java.util.Collections$EmptyList cannot be cast to class java.util.ArrayList which fails the entire batch -- exactly the frame in the reporter stack trace (ClickHouseDataTypeMapper.convert, from PreparedStatementExecutor.insertPreparedStatement). Bind through java.util.Collection instead. Every value that can reach this branch is a Connect array, i.e. a List, so Collection covers all of them; Collection.toArray() returns the same Object[] the ArrayList cast produced, so the case that worked before is unchanged. Regression test ClickHouseDataTypeMapperArrayTest pins the reporter Collections.emptyList() case plus List.of, Arrays.asList and LinkedList, and keeps an ArrayList control asserting identical bound elements. Verified failing first: 4 of 5 errored with the ClassCastException above (the ArrayList control passed); all 5 pass after the fix. * fix: initial snapshot never finishes -- NPE on every unparseable record (#1416) * fix: initial snapshot never finishes -- NPE on every unparseable record (#1379) Root cause ---------- processEveryChangeRecord dereferenced the parser's return value before testing it for null: chStruct = debeziumRecordParserService.parse(record, recordCommitter, lastRecordInBatch); chStruct.setSequenceNumber(sequenceNumber); try { if (chStruct != null) { Returning null is contract, not failure. SourceRecordParserService.parse returns null for a null SourceRecord or null value (line 57), for a value that is not a Struct (line 62), and for a value the converter cannot map (line 85). The `if (chStruct != null)` on the very next line is the method's own admission that null is expected. Every such record therefore raised java.lang.NullPointerException: Cannot invoke "...ClickHouseStruct.setSequenceNumber(long)" because "chStruct" is null which the method's outer `catch (Exception e)` swallowed. The record was dropped silently and the snapshot loop logged the same stack trace once per record -- the two symptoms reported in #1379: an initial snapshot that never completes, and NPE spam. Fix --- Move setSequenceNumber inside the existing null check and make the null case observable with a WARN naming the record, instead of a swallowed NPE. No other change; the surrounding logic is untouched. Proof ----- NullParsedRecordSkipTest drives processEveryChangeRecord with a parser stubbed to return null (the documented path) and captures what the class logs via a log4j2 appender. Before the fix, the test fails and quotes back the reporter's exception verbatim: [ERROR] Tests run: 1, Failures: 1, Errors: 0, Skipped: 0 java.lang.AssertionError: an unconvertible record must not raise a NullPointerException; issue #1379 logged one per record for the whole snapshot, got: java.lang.NullPointerException: Cannot invoke "com.altinity.clickhouse.sink.connector.model.ClickHouseStruct.setSequenceNumber(long)" because "chStruct" is null After the fix: [INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0 No regression. Module suite (-pl sink-connector-lightweight -am test, -Dmaven.test.failure.ignore=true so the reactor reaches both modules; the 50 errors are *IT tests that need a Docker environment unavailable here, identical in both runs): baseline (stashed): Tests run: 288, Failures: 0, Errors: 50, Skipped: 11 with this change: Tests run: 289, Failures: 0, Errors: 50, Skipped: 11 A per-test-class diff of the two runs shows exactly one line of difference: the added NullParsedRecordSkipTest. Fixes #1379 * test: end-to-end regression coverage for #1379 * fix: MySQL DOUBLE auto-created as Float32, silently truncating precision (#1417) * fix: MySQL DOUBLE auto-created as Float32, silently truncating precision ClickHouseDataTypeMapper.dataTypesMap mapped BOTH FLOAT32 and FLOAT64 to ClickHouseDataType.Float32. That map is what types the columns of auto-created tables, so every replicated MySQL DOUBLE / Kafka Connect FLOAT64 landed in a 32-bit ClickHouse column: ~15 significant decimal digits truncated to ~7. Nothing was logged and row counts still matched, so checksum jobs reported the table clean. The example in the issue, 123456789.123456789, is stored as 123456792. Blast radius of the map, every consumer traced: - ClickHouseTableOperationsBase.getColumnNameToCHDataTypeMapping:127 picks the column type for CREATE TABLE (via ClickHouseAutoCreateTable.createNewTable:68) and for the ADD COLUMN clause of an auto-ALTER (via ClickHouseAlterTable.alterTable:133). This is the defect site, and it only affects tables created from now on -- a wider column for new tables, no rewrite of existing ones. - DataTypeConverter:196 (lightweight DDL path) already returns Float64 for Types.DOUBLE at :189, before this lookup is reached, so that path was already correct and is unchanged. - ClickHouseDataTypeMapper.convert:526 uses it only for the ARRAY element type, where widening a Float64 array element is the same correction. Existing tables are unaffected: the write path does not consult this map. PreparedStatementFieldMapper:274 resolves the target type per column from live table metadata (DBMetadata.getColumnsDataTypesForTable, querying system.columns), so a column already declared Float32 keeps binding exactly as before. The bind path is deliberately left alone. convert():325 narrows with setFloat only for a genuine Float; a Double goes through ClickHouseDoubleValue.of(v).asBigDecimal(), which is lossless -- verified against clickhouse-data 0.9.8: 0.1234567890123456 binds unchanged, where (float) of the same value gives 0.12345679. The narrowing was purely in the declared column type. FLOAT32 stays Float32. Its narrowing already happened inside Debezium, and widening the column would only promise a precision the pipeline cannot deliver -- the same reasoning DataTypeConverter documents for Types.REAL. The defect was invisible to the existing auto-create tests because ClickHouseAutoCreateTableBase.getExpectedColumnToDataTypesMap hand-builds the column map with the CORRECT Float64 entries rather than deriving it from the mapper, so the wrong lookup was never exercised. The new tests drive the real mapper and the real auto-create consumer. Verified failing before the fix: float64MapsToFloat64:57 expected Float64 but was Float32 autoCreatedColumnForDoubleIsFloat64:86 expected Float64 but was Float32 and passing after (4/4). Suite for -pl sink-connector -am, baseline vs post: before Tests run: 226, Failures: 0, Errors: 20 after Tests run: 230, Failures: 0, Errors: 20 The 20 errors are the pre-existing docker-unavailable ones and are an identical set in both runs (diff of the error list is empty); the whole delta is the 4 new tests. Fixes #1262 * test: end-to-end regression coverage for #1262 * fix: support PostgreSQL infinity values in timestamptz (#1418) * fix: support PostgreSQL infinity values in timestamptz (#1231) PostgreSQL timestamptz accepts the special values `infinity` and `-infinity`. Debezium delivers them verbatim as those literal strings (PostgresValueConverter#convertTimestampWithZone returns the constants POSITIVE_INFINITY / NEGATIVE_INFINITY, confirmed in the bytecode of debezium-connector-postgres 3.1.3.Final). ZonedTimestampConverter.convert() initialises `result` to "" and tries each of its 17 patterns in turn. None of them match `infinity`, so the loop falls through to the error log and returns the empty string -- the row is delivered to ClickHouse with an empty value rather than failing, which is exactly the symptom reported in #1231: ERROR - Error parsing zonedtimestamp -infinity ClickHouse DateTime64 cannot represent an actual infinity, so saturate to the bounds of the target type instead. These are the same bounds the method already clamps out-of-range timestamps to a few lines below (BinaryStreamUtils.DATETIME64_MAX / DATETIME64_MIN, i.e. 2299-12-31 23:59:59 and 1900-01-01 00:00:00), so no new sentinel is introduced and the PostgreSQL ordering semantics are preserved: `infinity` sorts after and `-infinity` before every other representable timestamp. The change is deliberately scoped to the two known literals. The wider silent-empty-string behaviour for genuinely unparseable input is left as-is: turning it into a throw would change the failure mode of every existing deployment from degraded to hard-stopped, which does not belong in a targeted bug fix. Test: testZonedTimestampConverterInfinity fails before the change with `expected:<[2299-12-31 23:59:59.000000]> but was:<[]>` and passes after. * test: end-to-end regression coverage for #1231 * fix: batch with a missing ClickHouse table is silently dropped (#1285) (#1419) With auto.create.tables disabled and a table absent from ClickHouse, the single-threaded writer path drops the batch and every topic ordered after the failing one, with nothing but a log line to show for it. processRecordsByTopic returns false once the metadata retries are exhausted for a table it cannot read. persistRecords broke out of the per-topic loop on that and returned NORMALLY, and its outer catch swallowed anything thrown. The caller, DebeziumChangeEventCapture.appendToRecords, has no return value to inspect and no exception to catch, so handleBatch completed as if the batch had been written. The acknowledgement block is skipped, so THIS batch is never committed -- but the engine moves on to the next batch, whose markBatchFinished() commits a HIGHER offset. The unwritten records are then behind the committed offset and are never replayed: permanent loss, while the connector looks healthy. persistRecords now raises BatchPersistenceException instead of returning quietly, so the failure reaches the Debezium engine, which stops and restarts from the last committed offset and replays the records. The outer catch no longer swallows: the only exception it still absorbs is an interrupt, which means the task is shutting down rather than failing to write, and turning that into a persistence failure would make a clean stop trigger a restart loop. Scoped to ClickHouseBatchWriter. ClickHouseBatchRunnable carries the same break but survives it -- it clears currentBatch only when the result is true, so it re-polls and retries the same batch -- and is deliberately left alone. This writer is invoked once per batch and has no retry loop, so failing loudly is the only way for it not to lose data. Test: ClickHouseBatchWriterMissingTableTest drives wasTableMetaDataRetrieved() to false through the same state a missing table produces and asserts that no record is acknowledged AND that the failure reaches the caller. Verified failing first on the unmodified tree: AssertionError: persistRecords returned normally after failing to write the batch: the caller cannot tell the batch was dropped, so the offset advances past records that never reached ClickHouse (issue #1285) Module suite: 226 run / 0 failures / 20 errors before, 228 run / 0 failures / 20 errors after -- the 20 errors are testcontainers tests with no Docker environment on the build host, identical in both runs. * Fix #1213: _version stores UInt64 max when the source has no GTID (#1420) On a MySQL source with GTID disabled, the _version column was filled with 18446744073709551615 and the JDBC driver then refused to read it back ("integer overflow: 18446744073709551615 cannot be presented as long"). Chain: - ClickHouseStruct.version is initialised to UNINITIALIZED_VALUE = -1 (ClickHouseStruct.java:48, :204). - calculateVersion() only assigned a version from gtid, sequenceNumber or lsn (:823-834) and fell through, leaving -1, when all three were absent. setGtid is only reached when source.gtid splits into exactly 2 segments (:477-482), which never happens without GTID; nothing in sink-connector/src/main ever calls setSequenceNumber (only the lightweight DebeziumChangeEventCapture does), so the Kafka Connect path has no sequence number either; and lsn is a Postgres-only field. - handleVersionColumn then executed ps.setLong(index, -1) (PreparedStatementFieldMapper.java:503) into the UInt64 _version column, which stores 18446744073709551615. The unreadable cell was only the visible symptom. UInt64 max is the largest representable version, so under ReplacingMergeTree that row wins every future deduplication permanently and all later updates and deletes for the key are silently discarded on merge -- unrecoverable data loss. Fix, in two parts: 1. calculateVersion() gains a final fallback for the case that previously fell through: SnowFlakeId.generate(ts_ms, kafkaOffset). This reuses the connector's existing, already shipping version formula, substituting the Kafka offset for the absent GTID. Both inputs are populated on exactly the affected path -- ts_ms from source.ts_ms (set in setAdditionalMetaData) and the offset from record.kafkaOffset() -- and both are monotonic per source, so versions order by source commit time first and by offset within the same millisecond, the same ordering the GTID form produces. 2. The two bind sites (handleVersionColumn and insertTombstonePreparedStatement) call rejectUnderivableVersion(), which throws IllegalStateException naming the record and the value that would have been stored. After (1) this is only reachable when the record carries no ordering key AND no source timestamp, i.e. a malformed event. Failing the batch is strictly better than writing a row that can never be superseded. No existing deployment changes: a source supplying a gtid, sequenceNumber or lsn still takes its existing branch untouched, so the 2.8.0-compatible version domain of the lightweight connector and the Postgres LSN domain are unaffected. The new fallback is reachable only from the branch that was previously corrupt, and it is applied regardless of the snowflake.id flag because without a GTID there is no raw transaction id for the non-snowflake branch to use. VersionFallbackWithoutGtidTest pins all of it. Its 5 defect tests fail against the previous logic -- verbatim, before the fix: bindMustNotWriteUint64Max:165 the bind wrote the -1 sentinel, stored as UInt64 18446744073709551615 ==> expected: not equal but was: <18446744073709551615> and its 4 regression tests (GTID snowflake, GTID plain, sequence number, LSN) pass both before and after, pinning that nothing else moved. Regression: sink-connector -am, baseline 226 tests / 0 failures / 20 errors, post-change 235 / 0 / 20. The 20 errors are the pre-existing Docker-dependent tests (DBMetadataTest, DbWriterTest, ClickHouseCreateDatabaseTest, DbKafkaOffsetWriterTest) and the sorted error sets are identical. * Added release notes for 2.10.2 * Fix datatype integration tests * fix: a keyless source table must not prevent the connector from starting (#1426) #1401 added a startup check that refuses to replicate a MySQL source holding a table with no PRIMARY KEY and no non-null UNIQUE key. The diagnosis behind it stands -- such a table has no row identity in the binlog, so its ClickHouse copy silently diverges -- but refusing is the wrong remedy, for two reasons. It is not the connector's decision. A keyless table is often migration bookkeeping nobody reads, or one whose key is already scheduled for a maintenance window. Whether an imperfect copy of it is acceptable in the meantime is a judgement about the data, which the operator has and the connector does not. And the blast radius is wrong. The refusal is per SOURCE, not per table: one keyless table stops the entire pipeline, including every correctly-keyed table on the same server. On a source with dozens of replicated schemas, a single alembic_version row -- three columns, written once per migration -- takes everything else down with it. The cost of refusing lands almost entirely on data that was replicating perfectly well. So the check now reports instead of blocking. Everything that made the refusal actionable is kept: the banner still names every offending table and carries the exact ALTER TABLE that fixes it, still says NOT retroactive so nobody expects sql_generate_invisible_primary_key to repair existing tables, and is still emitted at ERROR at startup, at table creation, and periodically during replication. Only the throw is gone, and the wording now states plainly that replication is continuing -- "REPLICATING ANYWAY -- THESE TABLES WILL DIVERGE" -- so nobody hunts for a startup failure that never happens. Both escape hatches keep working and now mean what their names say: table.exclude.list leaves a table out of replication entirely, and keyless.table.check.skip=true silences the reporting. Changes ------- * UnsupportedSourceException deleted; both throw sites log.error and return. * refusalPreExisting / refusalTooOld renamed to warningPreExisting / warningTooOld, text rewritten from "REFUSING TO REPLICATE" to "REPLICATING ANYWAY -- THESE TABLES WILL DIVERGE". * The rethrow in the catch chain removed, so no path escapes check(). * Class javadoc, the DebeziumChangeEventCapture call-site comment, the MySqlDDLParserListenerImpl note and two parser-test comments corrected: they described a refusal that no longer exists. * The IT config keeps keyless.table.check.skip=true, but its comment no longer claims the ITs need it to run -- they do not, now that startup is never blocked. It only suppresses a banner in a fixture that is keyless on purpose. Tests ----- 30 in KeylessTablePreflightTest (was 28), 160 across the affected suites, all green. Two of the new tests are structural rather than behavioural, and deliberately so. The behavioural test drives every entry path a unit test can reach -- the skip override, a non-MySQL source, an unreachable host, an incomplete config -- and asserts none throws. But the two sites that actually decided to refuse sit AFTER a successful connection and a completed table scan, which no unit test reaches without a live MySQL. Verified by measurement, not assumption: with the refusal reinstated, the behavioural test still PASSED and only the structural check failed. So check() is additionally scanned in source for any throw statement, and the class is reflected over to prove no refusal-era API survives. Failing-first: with the throws restored, testCheckContainsNoThrowStatement fails naming the exact reinstated line, and testNoRefusalApiRemains fails naming the exception type. With the fix in place, both pass. Co-authored-by: OmniWatcher <svc-dba-github@jumptrading.com> Co-authored-by: Claude <noreply@anthropic.com> * Added release notes for 2.10.3 * fix(db_compare): only un-exclude sink-connector metadata columns The un-exclusion rule in get_table_checksum_query() was added so that is_deleted is not dropped when the connector also creates _is_deleted. It matched on the underscore prefix alone, so it fired for ANY excluded column whose _-prefixed name happened to exist on the table. A table with a business column 'version' and the connector's '_version' therefore had 'version' silently re-added to the ClickHouse checksum while mysql_table_checksum.py still honoured the exclusion. The two sides hashed different column sets, so the checksum could never match and no --exclude_columns/ignored_columns value could suppress it. Restrict the rule to columns the connector itself manages. is_deleted remains in the set, so the original is_deleted/_is_deleted behaviour is unchanged; a user-supplied exclusion is now always honoured. Co-Authored-By: Claude <noreply@anthropic.com> --------- Co-authored-by: kanthi subramanian <subkanthi@gmail.com> Co-authored-by: OmniWatcher <svc-dba-github@jumptrading.com> Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: OmniWatcher <dbajobs@jumptrading.com> * fix: eliminate the DDL/DML race that silently drops column values (#1432) * Added release notes for 2.10.1 release * fix: /metrics served with no Content-Type, breaking Prometheus 3.x (#1096) (#1413) The /metrics handler wrote its body without ever setting a Content-Type header. Prometheus 3.x treats that as a non-compliant target and fails the scrape outright: non-compliant scrape target sending blank Content-Type and no fallback_scrape_protocol specified for target so the connector silently stopped being monitored after a Prometheus upgrade. Set the Prometheus text exposition content type, "text/plain; version=0.0.4; charset=utf-8", before sendResponseHeaders, which is the call that commits the headers. Also encode the body once instead of calling response.getBytes() twice. That is a real correctness bug, not tidying: the two calls each used the platform default charset, so the length declared in the headers and the bytes actually written could differ for non-ASCII metric labels under a non-UTF-8 default charset, leaving a scraper short of the promised Content-Length. Encoding once to UTF-8 and using that array length makes the declared length true and matches the charset now advertised. Regression test MetricsContentTypeTest starts the real metrics server on a free port and scrapes it over HTTP, since the header is only observable on the wire. Verified failing first: Content-Type was null ("expected: not <null>"); it is the expected value after the fix. The Content-Length assertion passed before the fix too -- the current scrape output is ASCII -- and is there to pin the invariant, not to reproduce a live failure. * fix: INT8 (Byte) hard-cast to Integer fails every TINYINT batch (#385) (#1414) convert() groups INT8 and INT32 into a single isFieldTypeInt branch whose terminal case was: ps.setInt(index, (Integer) value); An INT8 schema delivers a java.lang.Byte, never an Integer, so this threw ClassCastException: class java.lang.Byte cannot be cast to class java.lang.Integer as reported -- and because it aborts the batch, the connector could not replicate any table holding a MySQL TINYINT. The isWiderIntegerTarget hatch immediately above does not cover it: it only fires for Int32/UInt32/Int64/UInt64 ClickHouse targets, and an INT8 source column maps to ClickHouse Int8 (Int32_SCHEMA -> Int8 in dataTypesMap), or UInt8 when the column is unsigned. Read the value as a Number and bind intValue(), so the boxed type no longer has to be guessed. Byte, Short and Integer all widen correctly and keep their sign. For the INT32 case that works today this is a no-op: Integer.intValue() returns the same int the cast produced, so the same setInt call with the same value is issued. Regression test ClickHouseDataTypeMapperInt8Test pins TINYINT to Int8 and UInt8 targets and negative/boundary bytes (-1, MIN_VALUE, MAX_VALUE), with two controls: INT32 still binds identically via setInt, and the isWiderIntegerTarget path still routes to setObject. Verified failing first: the 3 INT8 tests errored with the ClassCastException above while both controls passed; all 5 pass after the fix. * fix: ARRAY binding hard-casts to ArrayList, breaking every other List (#749) (#1415) The ARRAY branch of ClickHouseDataTypeMapper.convert cast the incoming value to java.util.ArrayList: ps.setArray(index, ps.getConnection().createArrayOf( dt.name(), ((ArrayList) value).toArray())); But the Kafka Connect contract for an array field is java.util.List, not ArrayList. Debezium and the Connect converters may hand over any List implementation, and an empty array in particular arrives as Collections.emptyList(). The cast then throws ClassCastException: class java.util.Collections$EmptyList cannot be cast to class java.util.ArrayList which fails the entire batch -- exactly the frame in the reporter stack trace (ClickHouseDataTypeMapper.convert, from PreparedStatementExecutor.insertPreparedStatement). Bind through java.util.Collection instead. Every value that can reach this branch is a Connect array, i.e. a List, so Collection covers all of them; Collection.toArray() returns the same Object[] the ArrayList cast produced, so the case that worked before is unchanged. Regression test ClickHouseDataTypeMapperArrayTest pins the reporter Collections.emptyList() case plus List.of, Arrays.asList and LinkedList, and keeps an ArrayList control asserting identical bound elements. Verified failing first: 4 of 5 errored with the ClassCastException above (the ArrayList control passed); all 5 pass after the fix. * fix: initial snapshot never finishes -- NPE on every unparseable record (#1416) * fix: initial snapshot never finishes -- NPE on every unparseable record (#1379) Root cause ---------- processEveryChangeRecord dereferenced the parser's return value before testing it for null: chStruct = debeziumRecordParserService.parse(record, recordCommitter, lastRecordInBatch); chStruct.setSequenceNumber(sequenceNumber); try { if (chStruct != null) { Returning null is contract, not failure. SourceRecordParserService.parse returns null for a null SourceRecord or null value (line 57), for a value that is not a Struct (line 62), and for a value the converter cannot map (line 85). The `if (chStruct != null)` on the very next line is the method's own admission that null is expected. Every such record therefore raised java.lang.NullPointerException: Cannot invoke "...ClickHouseStruct.setSequenceNumber(long)" because "chStruct" is null which the method's outer `catch (Exception e)` swallowed. The record was dropped silently and the snapshot loop logged the same stack trace once per record -- the two symptoms reported in #1379: an initial snapshot that never completes, and NPE spam. Fix --- Move setSequenceNumber inside the existing null check and make the null case observable with a WARN naming the record, instead of a swallowed NPE. No other change; the surrounding logic is untouched. Proof ----- NullParsedRecordSkipTest drives processEveryChangeRecord with a parser stubbed to return null (the documented path) and captures what the class logs via a log4j2 appender. Before the fix, the test fails and quotes back the reporter's exception verbatim: [ERROR] Tests run: 1, Failures: 1, Errors: 0, Skipped: 0 java.lang.AssertionError: an unconvertible record must not raise a NullPointerException; issue #1379 logged one per record for the whole snapshot, got: java.lang.NullPointerException: Cannot invoke "com.altinity.clickhouse.sink.connector.model.ClickHouseStruct.setSequenceNumber(long)" because "chStruct" is null After the fix: [INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0 No regression. Module suite (-pl sink-connector-lightweight -am test, -Dmaven.test.failure.ignore=true so the reactor reaches both modules; the 50 errors are *IT tests that need a Docker environment unavailable here, identical in both runs): baseline (stashed): Tests run: 288, Failures: 0, Errors: 50, Skipped: 11 with this change: Tests run: 289, Failures: 0, Errors: 50, Skipped: 11 A per-test-class diff of the two runs shows exactly one line of difference: the added NullParsedRecordSkipTest. Fixes #1379 * test: end-to-end regression coverage for #1379 * fix: MySQL DOUBLE auto-created as Float32, silently truncating precision (#1417) * fix: MySQL DOUBLE auto-created as Float32, silently truncating precision ClickHouseDataTypeMapper.dataTypesMap mapped BOTH FLOAT32 and FLOAT64 to ClickHouseDataType.Float32. That map is what types the columns of auto-created tables, so every replicated My 10BC8 SQL DOUBLE / Kafka Connect FLOAT64 landed in a 32-bit ClickHouse column: ~15 significant decimal digits truncated to ~7. Nothing was logged and row counts still matched, so checksum jobs reported the table clean. The example in the issue, 123456789.123456789, is stored as 123456792. Blast radius of the map, every consumer traced: - ClickHouseTableOperationsBase.getColumnNameToCHDataTypeMapping:127 picks the column type for CREATE TABLE (via ClickHouseAutoCreateTable.createNewTable:68) and for the ADD COLUMN clause of an auto-ALTER (via ClickHouseAlterTable.alterTable:133). This is the defect site, and it only affects tables created from now on -- a wider column for new tables, no rewrite of existing ones. - DataTypeConverter:196 (lightweight DDL path) already returns Float64 for Types.DOUBLE at :189, before this lookup is reached, so that path was already correct and is unchanged. - ClickHouseDataTypeMapper.convert:526 uses it only for the ARRAY element type, where widening a Float64 array element is the same correction. Existing tables are unaffected: the write path does not consult this map. PreparedStatementFieldMapper:274 resolves the target type per column from live table metadata (DBMetadata.getColumnsDataTypesForTable, querying system.columns), so a column already declared Float32 keeps binding exactly as before. The bind path is deliberately left alone. convert():325 narrows with setFloat only for a genuine Float; a Double goes through ClickHouseDoubleValue.of(v).asBigDecimal(), which is lossless -- verified against clickhouse-data 0.9.8: 0.1234567890123456 binds unchanged, where (float) of the same value gives 0.12345679. The narrowing was purely in the declared column type. FLOAT32 stays Float32. Its narrowing already happened inside Debezium, and widening the column would only promise a precision the pipeline cannot deliver -- the same reasoning DataTypeConverter documents for Types.REAL. The defect was invisible to the existing auto-create tests because ClickHouseAutoCreateTableBase.getExpectedColumnToDataTypesMap hand-builds the column map with the CORRECT Float64 entries rather than deriving it from the mapper, so the wrong lookup was never exercised. The new tests drive the real mapper and the real auto-create consumer. Verified failing before the fix: float64MapsToFloat64:57 expected Float64 but was Float32 autoCreatedColumnForDoubleIsFloat64:86 expected Float64 but was Float32 and passing after (4/4). Suite for -pl sink-connector -am, baseline vs post: before Tests run: 226, Failures: 0, Errors: 20 after Tests run: 230, Failures: 0, Errors: 20 The 20 errors are the pre-existing docker-unavailable ones and are an identical set in both runs (diff of the error list is empty); the whole delta is the 4 new tests. Fixes #1262 * test: end-to-end regression coverage for #1262 * fix: support PostgreSQL infinity values in timestamptz (#1418) * fix: support PostgreSQL infinity values in timestamptz (#1231) PostgreSQL timestamptz accepts the special values `infinity` and `-infinity`. Debezium delivers them verbatim as those literal strings (PostgresValueConverter#convertTimestampWithZone returns the constants POSITIVE_INFINITY / NEGATIVE_INFINITY, confirmed in the bytecode of debezium-connector-postgres 3.1.3.Final). ZonedTimestampConverter.convert() initialises `result` to "" and tries each of its 17 patterns in turn. None of them match `infinity`, so the loop falls through to the error log and returns the empty string -- the row is delivered to ClickHouse with an empty value rather than failing, which is exactly the symptom reported in #1231: ERROR - Error parsing zonedtimestamp -infinity ClickHouse DateTime64 cannot represent an actual infinity, so saturate to the bounds of the target type instead. These are the same bounds the method already clamps out-of-range timestamps to a few lines below (BinaryStreamUtils.DATETIME64_MAX / DATETIME64_MIN, i.e. 2299-12-31 23:59:59 and 1900-01-01 00:00:00), so no new sentinel is introduced and the PostgreSQL ordering semantics are preserved: `infinity` sorts after and `-infinity` before every other representable timestamp. The change is deliberately scoped to the two known literals. The wider silent-empty-string behaviour for genuinely unparseable input is left as-is: turning it into a throw would change the failure mode of every existing deployment from degraded to hard-stopped, which does not belong in a targeted bug fix. Test: testZonedTimestampConverterInfinity fails before the change with `expected:<[2299-12-31 23:59:59.000000]> but was:<[]>` and passes after. * test: end-to-end regression coverage for #1231 * fix: batch with a missing ClickHouse table is silently dropped (#1285) (#1419) With auto.create.tables disabled and a table absent from ClickHouse, the single-threaded writer path drops the batch and every topic ordered after the failing one, with nothing but a log line to show for it. processRecordsByTopic returns false once the metadata retries are exhausted for a table it cannot read. persistRecords broke out of the per-topic loop on that and returned NORMALLY, and its outer catch swallowed anything thrown. The caller, DebeziumChangeEventCapture.appendToRecords, has no return value to inspect and no exception to catch, so handleBatch completed as if the batch had been written. The acknowledgement block is skipped, so THIS batch is never committed -- but the engine moves on to the next batch, whose markBatchFinished() commits a HIGHER offset. The unwritten records are then behind the committed offset and are never replayed: permanent loss, while the connector looks healthy. persistRecords now raises BatchPersistenceException instead of returning quietly, so the failure reaches the Debezium engine, which stops and restarts from the last committed offset and replays the records. The outer catch no longer swallows: the only exception it still absorbs is an interrupt, which means the task is shutting down rather than failing to write, and turning that into a persistence failure would make a clean stop trigger a restart loop. Scoped to ClickHouseBatchWriter. ClickHouseBatchRunnable carries the same break but survives it -- it clears currentBatch only when the result is true, so it re-polls and retries the same batch -- and is deliberately left alone. This writer is invoked once per batch and has no retry loop, so failing loudly is the only way for it not to lose data. Test: ClickHouseBatchWriterMissingTableTest drives wasTableMetaDataRetrieved() to false through the same state a missing table produces and asserts that no record is acknowledged AND that the failure reaches the caller. Verified failing first on the unmodified tree: AssertionError: persistRecords returned normally after failing to write the batch: the caller cannot tell the batch was dropped, so the offset advances past records that never reached ClickHouse (issue #1285) Module suite: 226 run / 0 failures / 20 errors before, 228 run / 0 failures / 20 errors after -- the 20 errors are testcontainers tests with no Docker environment on the build host, identical in both runs. * Fix #1213: _version stores UInt64 max when the source has no GTID (#1420) On a MySQL source with GTID disabled, the _version column was filled with 18446744073709551615 and the JDBC driver then refused to read it back ("integer overflow: 18446744073709551615 cannot be presented as long"). Chain: - ClickHouseStruct.version is initialised to UNINITIALIZED_VALUE = -1 (ClickHouseStruct.java:48, :204). - calculateVersion() only assigned a version from gtid, sequenceNumber or lsn (:823-834) and fell through, leaving -1, when all three were absent. setGtid is only reached when source.gtid splits into exactly 2 segments (:477-482), which never happens without GTID; nothing in sink-connector/src/main ever calls setSequenceNumber (only the lightweight DebeziumChangeEventCapture does), so the Kafka Connect path has no sequence number either; and lsn is a Postgres-only field. - handleVersionColumn then executed ps.setLong(index, -1) (PreparedStatementFieldMapper.java:503) into the UInt64 _version column, which stores 18446744073709551615. The unreadable cell was only the visible symptom. UInt64 max is the largest representable version, so under ReplacingMergeTree that row wins every future deduplication permanently and all later updates and deletes for the key are silently discarded on merge -- unrecoverable data loss. Fix, in two parts: 1. calculateVersion() gains a final fallback for the case that previously fell through: SnowFlakeId.generate(ts_ms, kafkaOffset). This reuses the connector's existing, already shipping version formula, substituting the Kafka offset for the absent GTID. Both inputs are populated on exactly the affected path -- ts_ms from source.ts_ms (set in setAdditionalMetaData) and the offset from record.kafkaOffset() -- and both are monotonic per source, so versions order by source commit time first and by offset within the same millisecond, the same ordering the GTID form produces. 2. The two bind sites (handleVersionColumn and insertTombstonePreparedStatement) call rejectUnderivableVersion(), which throws IllegalStateException naming the record and the value that would have been stored. After (1) this is only reachable when the record carries no ordering key AND no source timestamp, i.e. a malformed event. Failing the batch is strictly better than writing a row that can never be superseded. No existing deployment changes: a source supplying a gtid, sequenceNumber or lsn still takes its existing branch untouched, so the 2.8.0-compatible version domain of the lightweight connector and the Postgres LSN domain are unaffected. The new fallback is reachable only from the branch that was previously corrupt, and it is applied regardless of the snowflake.id flag because without a GTID there is no raw transaction id for the non-snowflake branch to use. VersionFallbackWithoutGtidTest pins all of it. Its 5 defect tests fail against the previous logic -- verbatim, before the fix: bindMustNotWriteUint64Max:165 the bind wrote the -1 sentinel, stored as UInt64 18446744073709551615 ==> expected: not equal but was: <18446744073709551615> and its 4 regression tests (GTID snowflake, GTID plain, sequence number, LSN) pass both before and after, pinning that nothing else moved. Regression: sink-connector -am, baseline 226 tests / 0 failures / 20 errors, post-change 235 / 0 / 20. The 20 errors are the pre-existing Docker-dependent tests (DBMetadataTest, DbWriterTest, ClickHouseCreateDatabaseTest, DbKafkaOffsetWriterTest) and the sorted error sets are identical. * Added release notes for 2.10.2 * Fix datatype integration tests * fix: a keyless source table must not prevent the connector from starting (#1426) #1401 added a startup check that refuses to replicate a MySQL source holding a table with no PRIMARY KEY and no non-null UNIQUE key. The diagnosis behind it stands -- such a table has no row identity in the binlog, so its ClickHouse copy silently diverges -- but refusing is the wrong remedy, for two reasons. It is not the connector's decision. A keyless table is often migration bookkeeping nobody reads, or one whose key is already scheduled for a maintenance window. Whether an imperfect copy of it is acceptable in the meantime is a judgement about the data, which the operator has and the connector does not. And the blast radius is wrong. The refusal is per SOURCE, not per table: one keyless table stops the entire pipeline, including every correctly-keyed table on the same server. On a source with dozens of replicated schemas, a single alembic_version row -- three columns, written once per migration -- takes everything else down with it. The cost of refusing lands almost entirely on data that was replicating perfectly well. So the check now reports instead of blocking. Everything that made the refusal actionable is kept: the banner still names every offending table and carries the exact ALTER TABLE that fixes it, still says NOT retroactive so nobody expects sql_generate_invisible_primary_key to repair existing tables, and is still emitted at ERROR at startup, at table creation, and periodically during replication. Only the throw is gone, and the wording now states plainly that replication is continuing -- "REPLICATING ANYWAY -- THESE TABLES WILL DIVERGE" -- so nobody hunts for a startup failure that never happens. Both escape hatches keep working and now mean what their names say: table.exclude.list leaves a table out of replication entirely, and keyless.table.check.skip=true silences the reporting. Changes ------- * UnsupportedSourceException deleted; both throw sites log.error and return. * refusalPreExisting / refusalTooOld renamed to warningPreExisting / warningTooOld, text rewritten from "REFUSING TO REPLICATE" to "REPLICATING ANYWAY -- THESE TABLES WILL DIVERGE". * The rethrow in the catch chain removed, so no path escapes check(). * Class javadoc, the DebeziumChangeEventCapture call-site comment, the MySqlDDLParserListenerImpl note and two parser-test comments corrected: they described a refusal that no longer exists. * The IT config keeps keyless.table.check.skip=true, but its comment no longer claims the ITs need it to run -- they do not, now that startup is never blocked. It only suppresses a banner in a fixture that is keyless on purpose. Tests ----- 30 in KeylessTablePreflightTest (was 28), 160 across the affected suites, all green. Two of the new tests are structural rather than behavioural, and deliberately so. The behavioural test drives every entry path a unit test can reach -- the skip override, a non-MySQL source, an unreachable host, an incomplete config -- and asserts none throws. But the two sites that actually decided to refuse sit AFTER a successful connection and a completed table scan, which no unit test reaches without a live MySQL. Verified by measurement, not assumption: with the refusal reinstated, the behavioural test still PASSED and only the structural check failed. So check() is additionally scanned in source for any throw statement, and the class is reflected over to prove no refusal-era API survives. Failing-first: with the throws restored, testCheckContainsNoThrowStatement fails naming the exact reinstated line, and testNoRefusalApiRemains fails naming the exception type. With the fix in place, both pass. Co-authored-by: OmniWatcher <svc-dba-github@jumptrading.com> Co-authored-by: Claude <noreply@anthropic.com> * Added release notes for 2.10.3 * fix: post-snapshot heartbeat offset is never committed, so the snapshot never finishes (#1379) (#1428) * fix: post-snapshot heartbeat offset is never committed, so the snapshot never finishes (#1379) A Debezium batch carries more than row changes. A heartbeat has a Struct value and no `op` field, so SourceRecordParserService.parse returns null for it -- that null is contract, not failure. Such a record was then dropped outright: never markProcessed, never markBatchFinished. Only records that became a row were ever acknowledged. That silently strands the end-of-snapshot state. Debezium marks the snapshot complete AFTER the last snapshot row is emitted (RelationalSnapshotChangeEventSource#createDataEvents calls preSnapshotCompletion/postSnapshotCompletion once the row loop is done), so every snapshot ROW still carries snapshot=INITIAL, snapshot_completed=false. The completed state rides only on records emitted after the snapshot -- and on an idle source those are exclusively heartbeats. Dropping them leaves replica_source_info pinned at snapshot=INITIAL/snapshot_completed=false forever, which is both halves of the report: - the status view never shows the snapshot finishing, even though the log says "Snapshot completed" and the rows are all in ClickHouse; - on restart InitialSnapshotter#shouldSnapshotData reads snapshotInProgress=true and re-runs the entire snapshot. The reporter's own 2.10.2 log shows the mechanism directly: the dropped heartbeat carries sourceOffset={lsn=74130877960, txId=3322, ...} with no snapshot keys at all (PostgresOffsetContext#getOffset omits them once postSnapshotCompletion has cleared the snapshot type), while the stored offset still has snapshot=INITIAL. Committing that heartbeat is exactly what lets a restart skip the snapshot. This is NOT the defect fixed in #1416. That fix stopped a NullPointerException and turned the drop into a clean, logged skip; the record is still dropped and its offset still never committed, which is why the reporter confirmed the symptom persists on 2.10.2. Ordering safety. A heartbeat carries the connector's CURRENT position, at or after every row already read, so acknowledging it while a row is still unwritten would advance the offset past that row and lose it on a crash -- the failure mode of #1285. The offset is therefore committed only when nothing is in flight: no row …
… value (#1436) * fix: a MATERIALIZED column drives an unbounded metadata re-read storm The stale-cache probe added in #1431 (refreshIfRecordHasUnknownColumn) treats "the record carries a column the cached column map does not have" as proof the cache is stale, and re-reads table metadata. That inference is wrong for a column ClickHouse computes. DBMetadata#getColumnsDataTypesForTable deliberately excludes columns whose default_kind is MATERIALIZED or ALIAS (DBMetadata.java:542), because ClickHouse rejects an INSERT that binds them. A source column that is MATERIALIZED on the ClickHouse side is therefore absent from the map permanently and CORRECTLY. The re-read returns a map that still lacks it, the code bumps the table's invalidation version anyway, and the next record repeats the entire cycle -- forever, at record rate. Each version bump also invalidates every cached DbWriter for the table (ClickHouseBatchRunnable#getDbWriterForTable), so every worker thread rebuilds its writer on its next batch, multiplying the metadata reads by the size of the thread pool. Measured on a production deployment where one replicated table had eight MATERIALIZED columns on the ClickHouse side: - 9,664,873 system.columns queries in a single hour - 15.5% of all queries on the ClickHouse server - the table's invalidation version reached 66,731,063 in three hours, for a table that had had no DDL in over a month - 120,632 "cache is stale" warnings in the connector error log - that connector's binlog offset stopped advancing, while every other connector on the same host reading the same MySQL source stayed current The fix separates the two cases the probe currently conflates. After the re-read, check whether the column actually appeared: - it did -> the cache WAS stale. Bump the version so every other cached writer rebuilds. Unchanged behaviour. - it did not -> the cache was never stale. Record the proof against the current table version, do NOT bump, and skip that column on subsequent records. The proof is version-scoped, so a genuine later DDL discards it and the column is probed exactly once more. A MATERIALIZED column later redefined as an ordinary one is still detected, and the silent column-drop #1431 exists to prevent is not reintroduced. Steady-state cost becomes one metadata query per computed column per DDL generation, instead of one per record. Verification: - CacheInvalidationProvenAbsentTest: 8 new tests, all pass - Mutation-checked: reverting isColumnProvenAbsent to the pre-fix behaviour fails 5 of the 8, so the suite detects the defect - Neighbouring suites green: NullValueColumnDropTest, GroupInsertQueryHistoryMultiRowTest, QueryFormatter*, DDLSchemaChangeWaiter* -- 70 tests, 0 failures - API change is additive only; no existing method signature or behaviour altered * docs+fix: the source is the authority -- report a shadowed source value SOURCE IS THE AUTHORITY: report a shadowed source value The connector replicates a source database into ClickHouse, which makes the source the authority on what the data is; ClickHouse is the replica and has to agree with it. Under that rule ALIAS and MATERIALIZED are not equivalent, even though the writable column map excludes both: - ALIAS stores nothing, so there is no stored value that can disagree with the source. Ignoring it is correct and now logs at debug. - MATERIALIZED DOES store a value, computed by ClickHouse. When the source also supplies that column, the replica silently keeps a DIFFERENT value: no error, no failed batch, identical row counts, detectable only by a value-level checksum. That is a real divergence. The first version of this change called both cases "correctly excluded", which is the same mistake at the log level -- it files a divergence away as correct. DBMetadata#getColumnDefaultKind is added so the two can be told apart, and reportUnwritableColumn now warns for MATERIALIZED, naming the remediation (redefine the column as an ordinary one so the replicated value is stored, or exclude it from replication), and warns for an undeterminable kind rather than assuming it is benign. The reporting is purely diagnostic and never changes what is written: binding a MATERIALIZED column would make ClickHouse reject the whole batch, turning a silent divergence into a stalled pipeline. Surfacing it is what lets an operator fix the schema. AGENTS.md / CLAUDE.md The source-of-truth rule was not written down anywhere, so every contributor rediscovered it or failed to. AGENTS.md now states it, with the column-kind rules, the failures it prevents (all of which happened), the fail-loudly rules, and the working rules for changes here. CLAUDE.md points at it and repeats the two rules that are missed most. Verification: - CacheInvalidationProvenAbsentTest: 8 tests - UnwritableColumnReportingTest: 6 tests covering MATERIALIZED, ALIAS, ordinary, missing column, metadata failure, and null arguments - Mutation-checked: reverting isColumnProvenAbsent to the pre-fix behaviour fails 5 of the 8 storm tests - 76 tests, 0 failures across the neighbouring suites (NullValueColumnDrop, GroupInsertQueryHistoryMultiRow, QueryFormatter*, DDLSchemaChangeWaiter*) - API change is additive only; no existing method signature or behaviour altered * enforce: the connector makes ClickHouse conform, it does not just report The source is the authority on what the data is, and this connector is what ENFORCES that on the replica. The previous version only reported a divergence and left ClickHouse wrong. A column defined MATERIALIZED on the ClickHouse side while the source also supplies it is the case that matters: ClickHouse stores its own computed value, the source value is never written, and the two sides disagree silently. The connector now corrects it with ALTER TABLE MODIFY COLUMN, restating the declared type with no default expression -- metadata-only, on the same DDL path the connector already uses for schema evolution. On success the caller re-reads the schema so the very batch that hit the problem writes the source value, and deliberately skips the proven-absent entry that would suppress it. ALIAS is unchanged. ENFORCEMENT IS VERIFIED: executeSystemQuery returns normally once its retry budget is spent, so no exception does not mean it worked; makeColumnWritable re-reads default_kind after the ALTER and only reports success when the column is genuinely writable. Code 497 ACCESS_DENIED added to NON_RETRYABLE_ERROR_CODES: retrying a denied ALTER blocks the CDC thread for the whole budget and still fails (rejected-ALTER test 45.0s to 0.03s). Enforcement fixes the write path FORWARD; rows written earlier need a backfill, stated in the log and in AGENTS.md. 91 tests, 0 failures. --------- Co-authored-by: minguyen9988 <minguyen9988@users.noreply.github.com>
…letes (#1379) (#1438) * fix: post-snapshot heartbeat offset is never committed, so the snapshot never finishes (#1379) A Debezium batch carries more than row changes. A heartbeat has a Struct value and no `op` field, so SourceRecordParserService.parse returns null for it -- that null is contract, not failure. Such a record was then dropped outright: never markProcessed, never markBatchFinished. Only records that became a row were ever acknowledged. That silently strands the end-of-snapshot state. Debezium marks the snapshot complete AFTER the last snapshot row is emitted (RelationalSnapshotChangeEventSource#createDataEvents calls preSnapshotCompletion/postSnapshotCompletion once the row loop is done), so every snapshot ROW still carries snapshot=INITIAL, snapshot_completed=false. The completed state rides only on records emitted after the snapshot -- and on an idle source those are exclusively heartbeats. Dropping them leaves replica_source_info pinned at snapshot=INITIAL/snapshot_completed=false forever, which is both halves of the report: - the status view never shows the snapshot finishing, even though the log says "Snapshot completed" and the rows are all in ClickHouse; - on restart InitialSnapshotter#shouldSnapshotData reads snapshotInProgress=true and re-runs the entire snapshot. The reporter's own 2.10.2 log shows the mechanism directly: the dropped heartbeat carries sourceOffset={lsn=74130877960, txId=3322, ...} with no snapshot keys at all (PostgresOffsetContext#getOffset omits them once postSnapshotCompletion has cleared the snapshot type), while the stored offset still has snapshot=INITIAL. Committing that heartbeat is exactly what lets a restart skip the snapshot. This is NOT the defect fixed in #1416. That fix stopped a NullPointerException and turned the drop into a clean, logged skip; the record is still dropped and its offset still never committed, which is why the reporter confirmed the symptom persists on 2.10.2. Ordering safety. A heartbeat carries the connector's CURRENT position, at or after every row already read, so acknowledging it while a row is still unwritten would advance the offset past that row and lose it on a crash -- the failure mode of #1285. The offset is therefore committed only when nothing is in flight: no row from this batch was handed off, both handoff queues are empty, and no batch is awaiting persistence (DebeziumOffsetManagement.hasUnwrittenBatches). On a busy pipeline that check fails and behaviour is exactly as before -- no regression -- because a busy pipeline commits its offsets through the rows themselves. It is precisely the idle case, where no row is coming, that needed this. The batch handler is extracted from the anonymous ChangeConsumer into handleChangeEventBatch so the batch semantics can be tested without standing up an engine; the consumer is now a single delegation. The extraction is otherwise behaviour-preserving. One incidental change: isDDLRecord is now evaluated once per record rather than only when rows are pending, because the result also selects the control-record branch; it is documented side-effect free. Verification: - ControlRecordOffsetCommitTest drives the real handleChangeEventBatch with a heartbeat built to Debezium's own heartbeat value schema. Failing-first verified: neutralising only the new acknowledgement reproduces the defect (expected:<1> but was:<0> markProcessed calls), and the other three cases pin the ordering-safety direction so the fix cannot pass by acknowledging unconditionally. - Full unit suite, both modules, against an unmodified develop baseline: 295 -> 299 tests, the delta being exactly the four new tests; errors unchanged at 73 with byte-identical error sets, all of them pre-existing Docker-environment failures on the build host. Docker-dependent integration suites were not run locally; CI exercises them. * fix: offset table must deduplicate on offset_key, not on the random id (#1379) The reporter tested the previous commit and the snapshot state still read back as snapshot_completed=false. That is a second, independent defect, and it is in the configs this repository ships. Debezium's JdbcOffsetBackingStore.save() performs a delete-then-insert, and the value it binds to `id` is UUID.randomUUID().toString() -- a fresh value on every flush. On ClickHouse the configured "delete" is `select * from %s`, a SELECT that removes nothing, because ClickHouse has no cheap row DELETE. Those two facts together mean the offset table only ever accumulates rows. So the sort key is the only thing that can collapse an old offset into the new one. With ORDER BY id every row is unique by construction: ReplacingMergeTree can never merge them and FINAL is a no-op, so every offset ever written stays queryable. The read path then orders by record_insert_ts, record_insert_seq -- but record_insert_ts is only second-resolution and record_insert_seq restarts at 1 on every connector start, so after a restart a STALE row can sort last and win. The connector resumes from an old position, and a snapshot that has already finished reads back as still in progress. ORDER BY offset_key makes the newest write for a given connector replace the previous one. That is what k8s/config.yml, docker/config-2.yml, docker/config_local.yml and the ansible template already used; the other copies had drifted. The most consequential one is src/main/resources/config.properties. That is not a sample -- it is the shipped runtime default, loaded as the base layer by both ClickHouseDebeziumEmbeddedApplication#loadPropertiesFile and EnvironmentConfigurationService#parse, so every deployment that does not override the DDL itself inherits it. The reporter's config came from docker/config_postgres.yml, which had the same defect. All 18 remaining files are corrected in one pass, so no copy is left to be rediscovered later: the runtime default, the docker samples (config, config_oracle, config_postgres, config_postgres_local), the helm chart configmap (both occurrences), the test resources, and the integration-test environments and helper. Files already using offset_key are untouched, as are the KeeperMap DDLs (PRIMARY KEY offset_key, already correct) and the schema-history table, which legitimately orders by id because JdbcSchemaHistory writes a stable id. Note this changes only how a NEW offset table is created. An existing table keeps its old sort key, so a deployment hitting this must drop the offset table and let the connector recreate it; the stale rows would otherwise keep shadowing the current offset. Verification: - OffsetTableDdlSortKeyTest asserts over EVERY config.properties on the classpath rather than the first match, because the test resources shadow the shipped defaults -- checking only the resolved one would silently stop covering src/main/resources. It matches the DDL as raw text, so a value written across multiple unescaped lines is still checked, and it fails if no DDL is found at all so it cannot pass vacuously. - Failing-first verified: reverting only the runtime default to ORDER BY id fails the test with expected:<offset_key> but was:<id>, naming the offending resource. - Full unit suite, both modules: 299 -> 301 tests, the delta being exactly the two new tests; errors unchanged at 73 with byte-identical error sets, all pre-existing Docker-environment failures on the build host. Docker-dependent integration suites were not run locally; CI exercises them. Jinja-Render-Check: rendered=4; formats=yaml; result=pass * fix: close the poll-then-register window in the quiescence check (#1379) Review of the previous commit raised a check-then-act race in isPipelineQuiescent(), and it is real. A consumer polls a batch off the handoff queue and only registers it in inFlightBatches once it reaches ClickHouseBatchRunnable#processBatch -- after addRecordsToHistoryTable has run. Between the poll and that registration the batch is in NEITHER the queue nor either map, so a quiescence check evaluated in that window sees everything empty and concludes nothing is outstanding. That is exactly the predicate the control-record offset commit uses to decide whether committing a heartbeat is safe. A heartbeat carries the connector's CURRENT position, so committing one while a polled but unwritten batch exists advances the committed offset past those rows and loses them on a crash -- the failure mode of #1285, which the check exists to prevent. The queue and map checks cannot close the window on their own, because the gap is precisely the handoff between them. A counter can: it is incremented by the PRODUCER before the batch becomes visible to any consumer, and released only after the batch has been acknowledged. The producer is also the thread that reads it, so its own increments happen-before its own read and no batch it has handed off can be missed. The maps are kept as a 10BC8 second line of defence rather than replaced. Details that matter for correctness: - Every routed group is counted separately. Hash routing splits one list into N groups that are acknowledged independently, so counting the caller's list once would under-count and let a commit through while sibling groups were still unwritten. - Single-threaded mode is not counted at all. persistRecords runs inline on the Debezium thread, so the rows are written (or have thrown) before control returns and nothing is ever outstanding across a quiescence check. - A batch that fails to reach its queue releases its registration, since no consumer will ever acknowledge it; without that the counter would stay above zero forever and no control-record offset could be committed again for the life of the process. - The release floors at zero. A batch parked in completedBatches and later retried can reach the acknowledgement path more than once, and a negative counter would make the pipeline read as quiescent while work is outstanding -- the one direction that is unsafe. Every remaining error in the counter therefore withholds a commit rather than permitting one. Verification: - HandedOffBatchVisibilityTest covers the window itself, the per-routed-group counting, and the zero floor. - Failing-first verified: dropping the counter from the predicate -- leaving the original map-only check -- fails all three, the first with "a batch handed to the consumers must read as unwritten immediately". - The test clears both static maps and the counter before each case. Surefire shares one JVM across the module and DebeziumOffsetManagementTest leaves entries in inFlightBatches, so without that isolation these assertions depend on execution order; that is what made them fail on the first full-suite run. - Full unit suite, both modules: 550 -> 559 tests against an unmodified develop baseline, the delta being exactly the nine tests added across this PR. Zero failures. Errors unchanged at 73 with byte-identical error sets, all pre-existing Docker-environment failures on the build host. Docker-dependent integration suites were not run locally; CI exercises them. * fix: enable heartbeats, without which the snapshot never completes (#1379) The reporter still sees snapshot_completed=false after three fixes for this issue, and the reason is that none of them could ever fire on his setup. Debezium marks a snapshot complete only AFTER the last snapshot row is emitted, so every snapshot ROW carries snapshot=INITIAL, snapshot_completed=false. The completed state rides exclusively on records emitted after the snapshot. On a source that goes idle once the snapshot ends -- the normal case for the small database someone first tries the connector on -- the only such records are heartbeats. commitControlRecordOffset was added to commit the offset from exactly those control records. But it can only act on a heartbeat that is actually emitted, and Debezium heartbeat.interval.ms defaults to 0, which disables heartbeats entirely. Nothing in this connector ever set it. So the machinery was correct and unreachable: no heartbeat, no control record, no commit, snapshot_completed=false forever. That is destructive rather than cosmetic. On restart Debezium reads the persisted offset, sees a snapshot that never completed, and re-runs the whole snapshot from the beginning (InitialSnapshotter#shouldSnapshotData keys off exactly this state). The connector can never make forward progress past its first snapshot and the target is rewritten every time the process restarts -- which is precisely what the reporter describes when he says interrupting and restarting starts from fresh. ensureHeartbeatInterval sets heartbeat.interval.ms to 5000 when the user has not configured one. A user-supplied value always wins, and an explicit 0 is honoured with a warning, so the escape hatch stays open for anyone who wants heartbeats off and accepts the consequence. A blank value counts as unset, so a config template rendering an empty string does not silently reinstate the bug. Tests. PostgresSnapshotCompletionIT is end-to-end on purpose: every unit-level piece of this already passed while the bug was live -- the parse correctly returns null for a heartbeat, the commit helper correctly commits when handed a control record, the quiescence check correctly gates it. The defect was that no heartbeat ever arrived, which is only observable by running a real connector against a real source and reading the persisted offset. It snapshots a small Postgres database, leaves the source ENTIRELY IDLE, asserts the persisted offset flips to snapshot_completed=true, then restarts the connector against the same offset store and asserts the snapshot is NOT re-run (row count unchanged). It deliberately does not set heartbeat.interval.ms, so it would pass against a broken build if it did. HeartbeatIntervalDefaultTest pins the property contract cheaply: default applied, user value wins, explicit 0 honoured, blank treated as unset, null inert, the default is a positive sane duration, and the key matches Debezium exactly (a typo there would make the default silently do nothing). Verification: 11 tests pass (7 new + the 4 existing #1379 control-record tests). Mutation-checked: disabling the setProperty call fails 2 of the 7, so the suite detects the defect rather than passing vacuously. JDK 17, Maven 3.9.11. --------- Co-authored-by: minguyen9988 <minguyen9988@users.noreply.github.com>
The step is `mvn --quiet surefire-report:report`. That is a REPORTING goal: invoked bare it does not bind to the lifecycle, so the test phase never runs. It only formats surefire XML that already exists, and on a clean runner none does. The job therefore produced an empty report and passed. Evidence from run 34307223948 job 102327628230: the step ran 03:43:01 to 03:43:56 -- 55 seconds, zero tests -- while the junit-report step printed "No annotations found for JUnit Test Report" and the check stayed green. That warning has been printing on every run. Found while verifying #1379: PostgresSnapshotCompletionIT sleeps well over 100 seconds by design, so a 55-second green job could not have run it. Fix: run `mvn test` first, then `surefire-report:report-only` to format what it produced. -Dmaven.test.failure.ignore=true keeps the existing junit-report step (fail_on_failure: true) as the gate, so a real failure is annotated on the PR instead of aborting the run before the report is written. The surefire config already includes **/*Test.java and **/*IT.java, so no include changes are needed. Expect the first honest run to be much slower than 55s and possibly red -- that is the point. Co-authored-by: minguyen9988 <minguyen9988@users.noreply.github.com>
* fix: post-snapshot heartbeat offset is never committed, so the snapshot never finishes (#1379) A Debezium batch carries more than row changes. A heartbeat has a Struct value and no `op` field, so SourceRecordParserService.parse returns null for it -- that null is contract, not failure. Such a record was then dropped outright: never markProcessed, never markBatchFinished. Only records that became a row were ever acknowledged. That silently strands the end-of-snapshot state. Debezium marks the snapshot complete AFTER the last snapshot row is emitted (RelationalSnapshotChangeEventSource#createDataEvents calls preSnapshotCompletion/postSnapshotCompletion once the row loop is done), so every snapshot ROW still carries snapshot=INITIAL, snapshot_completed=false. The completed state rides only on records emitted after the snapshot -- and on an idle source those are exclusively heartbeats. Dropping them leaves replica_source_info pinned at snapshot=INITIAL/snapshot_completed=false forever, which is both halves of the report: - the status view never shows the snapshot finishing, even though the log says "Snapshot completed" and the rows are all in ClickHouse; - on restart InitialSnapshotter#shouldSnapshotData reads snapshotInProgress=true and re-runs the entire snapshot. The reporter's own 2.10.2 log shows the mechanism directly: the dropped heartbeat carries sourceOffset={lsn=74130877960, txId=3322, ...} with no snapshot keys at all (PostgresOffsetContext#getOffset omits them once postSnapshotCompletion has cleared the snapshot type), while the stored offset still has snapshot=INITIAL. Committing that heartbeat is exactly what lets a restart skip the snapshot. This is NOT the defect fixed in #1416. That fix stopped a NullPointerException and turned the drop into a clean, logged skip; the record is still dropped and its offset still never committed, which is why the reporter confirmed the symptom persists on 2.10.2. Ordering safety. A heartbeat carries the connector's CURRENT position, at or after every row already read, so acknowledging it while a row is still unwritten would advance the offset past that row and lose it on a crash -- the failure mode of #1285. The offset is therefore committed only when nothing is in flight: no row from this batch was handed off, both handoff queues are empty, and no batch is awaiting persistence (DebeziumOffsetManagement.hasUnwrittenBatches). On a busy pipeline that check fails and behaviour is exactly as before -- no regression -- because a busy pipeline commits its offsets through the rows themselves. It is precisely the idle case, where no row is coming, that needed this. The batch handler is extracted from the anonymous ChangeConsumer into handleChangeEventBatch so the batch semantics can be tested without standing up an engine; the consumer is now a single delegation. The extraction is otherwise behaviour-preserving. One incidental change: isDDLRecord is now evaluated once per record rather than only when rows are pending, because the result also selects the control-record branch; it is documented side-effect free. Verification: - ControlRecordOffsetCommitTest drives the real handleChangeEventBatch with a heartbeat built to Debezium's own heartbeat value schema. Failing-first verified: neutralising only the new acknowledgement reproduces the defect (expected:<1> but was:<0> markProcessed calls), and the other three cases pin the ordering-safety direction so the fix cannot pass by acknowledging unconditionally. - Full unit suite, both modules, against an unmodified develop baseline: 295 -> 299 tests, the delta being exactly the four new tests; errors unchanged at 73 with byte-identical error sets, all of them pre-existing Docker-environment failures on the build host. Docker-dependent integration suites were not run locally; CI exercises them. * fix: offset table must deduplicate on offset_key, not on the random id (#1379) The reporter tested the previous commit and the snapshot state still read back as snapshot_completed=false. That is a second, independent defect, and it is in the configs this repository ships. Debezium's JdbcOffsetBackingStore.save() performs a delete-then-insert, and the value it binds to `id` is UUID.randomUUID().toString() -- a fresh value on every flush. On ClickHouse the configured "delete" is `select * from %s`, a SELECT that removes nothing, because ClickHouse has no cheap row DELETE. Those two facts together mean the offset table only ever accumulates rows. So the sort key is the only thing that can collapse an old offset into the new one. With ORDER BY id every row is unique by construction: ReplacingMergeTree can never merge them and FINAL is a no-op, so every offset ever written stays queryable. The read path then orders by record_insert_ts, record_insert_seq -- but record_insert_ts is only second-resolution and record_insert_seq restarts at 1 on every connector start, so after a restart a STALE row can sort last and win. The connector resumes from an old position, and a snapshot that has already finished reads back as still in progress. ORDER BY offset_key makes the newest write for a given connector replace the previous one. That is what k8s/config.yml, docker/config-2.yml, docker/config_local.yml and the ansible template already used; the other copies had drifted. The most consequential one is src/main/resources/config.properties. That is not a sample -- it is the shipped runtime default, loaded as the base layer by both ClickHouseDebeziumEmbeddedApplication#loadPropertiesFile and EnvironmentConfigurationService#parse, so every deployment that does not override the DDL itself inherits it. The reporter's config came from docker/config_postgres.yml, which had the same defect. All 18 remaining files are corrected in one pass, so no copy is left to be rediscovered later: the runtime default, the docker samples (config, config_oracle, config_postgres, config_postgres_local), the helm chart configmap (both occurrences), the test resources, and the integration-test environments and helper. Files already using offset_key are untouched, as are the KeeperMap DDLs (PRIMARY KEY offset_key, already correct) and the schema-history table, which legitimately orders by id because JdbcSchemaHistory writes a stable id. Note this changes only how a NEW offset table is created. An existing table keeps its old sort key, so a deployment hitting this must drop the offset table and let the connector recreate it; the stale rows would otherwise keep shadowing the current offset. Verification: - OffsetTableDdlSortKeyTest asserts over EVERY config.properties on the classpath rather than the first match, because the test resources shadow the shipped defaults -- checking only the resolved one would silently stop covering src/main/resources. It matches the DDL as raw text, so a value written across multiple unescaped lines is still checked, and it fails if no DDL is found at all so it cannot pass vacuously. - Failing-first verified: reverting only the runtime default to ORDER BY id fails the test with expected:<offset_key> but was:<id>, naming the offending resource. - Full unit suite, both modules: 299 -> 301 tests, the delta being exactly the two new tests; errors unchanged at 73 with byte-identical error sets, all pre-existing Docker-environment failures on the build host. Docker-dependent integration suites were not run locally; CI exercises them. Jinja-Render-Check: rendered=4; formats=yaml; result=pass * fix: close the poll-then-register window in the quiescence check (#1379) Review of the previous commit raised a check-then-act race in isPipelineQuiescent(), and it is real. A consumer polls a batch off the handoff queue and only registers it in inFlightBatches once it reaches ClickHouseBatchRunnable#processBatch -- after addRecordsToHistoryTable has run. Between the poll and that registration the batch is in NEITHER the queue nor either map, so a quiescence check evaluated in that window sees everything empty and concludes nothing is outstanding. That is exactly the predicate the control-record offset commit uses to decide whether committing a heartbeat is safe. A heartbeat carries the connector's CURRENT position, so committing one while a polled but unwritten batch exists advances the committed offset past those rows and loses them on a crash -- the failure mode of #1285, which the check exists to prevent. The queue and map checks cannot close the window on their own, because the gap is precisely the handoff between them. A counter can: it is incremented by the PRODUCER before the batch becomes visible to any consumer, and released only after the batch has been acknowledged. The producer is also the thread that reads it, so its own increments happen-before its own read and no batch it has handed off can be missed. The maps are kept as a second line of defence rather than replaced. Details that matter for correctness: - Every routed group is counted separately. Hash routing splits one list into N groups that are acknowledged independently, so counting the caller's list once would under-count and let a commit through while sibling groups were still unwritten. - Single-threaded mode is not counted at all. persistRecords runs inline on the Debezium thread, so the rows are written (or have thrown) before control returns and nothing is ever outstanding across a quiescence check. - A batch that fails to reach its queue releases its registration, since no consumer will ever acknowledge it; without that the counter would stay above zero forever and no control-record offset could be committed again for the life of the process. - The release floors at zero. A batch parked in completedBatches and later retried can reach the acknowledgement path more than once, and a negative counter would make the pipeline read as quiescent while work is outstanding -- the one direction that is unsafe. Every remaining error in the counter therefore withholds a commit rather than permitting one. Verification: - HandedOffBatchVisibilityTest covers the window itself, the per-routed-group counting, and the zero floor. - Failing-first verified: dropping the counter from the predicate -- leaving the original map-only check -- fails all three, the first with "a batch handed to the consumers must read as unwritten immediately". - The test clears both static maps and the counter before each case. Surefire shares one JVM across the module and DebeziumOffsetManagementTest leaves entries in inFlightBatches, so without that isolation these assertions depend on execution order; that is what made them fail on the first full-suite run. - Full unit suite, both modules: 550 -> 559 tests against an unmodified develop baseline, the delta being exactly the nine tests added across this PR. Zero failures. Errors unchanged at 73 with byte-identical error sets, all pre-existing Docker-environment failures on the build host. Docker-dependent integration suites were not run locally; CI exercises them. * fix: enable heartbeats, without which the snapshot never completes (#1379) The reporter still sees snapshot_completed=false after three fixes for this issue, and the reason is that none of them could ever fire on his setup. Debezium marks a snapshot complete only AFTER the last snapshot row is emitted, so every snapshot ROW carries snapshot=INITIAL, snapshot_completed=false. The completed state rides exclusively on records emitted after the snapshot. On a source that goes idle once the snapshot ends -- the normal case for the small database someone first tries the connector on -- the only such records are heartbeats. commitControlRecordOffset was added to commit the offset from exactly those control records. But it can only act on a heartbeat that is actually emitted, and Debezium heartbeat.interval.ms defaults to 0, which disables heartbeats entirely. Nothing in this connector ever set it. So the machinery was correct and unreachable: no heartbeat, no control record, no commit, snapshot_completed=false forever. That is destructive rather than cosmetic. On restart Debezium reads the persisted offset, sees a snapshot that never completed, and re-runs the whole snapshot from the beginning (InitialSnapshotter#shouldSnapshotData keys off exactly this state). The connector can never make forward progress past its first snapshot and the target is rewritten every time the process restarts -- which is precisely what the reporter describes when he says interrupting and restarting starts from fresh. ensureHeartbeatInterval sets heartbeat.interval.ms to 5000 when the user has not configured one. A user-supplied value always wins, and an explicit 0 is honoured w 10BC8 ith a warning, so the escape hatch stays open for anyone who wants heartbeats off and accepts the consequence. A blank value counts as unset, so a config template rendering an empty string does not silently reinstate the bug. Tests. PostgresSnapshotCompletionIT is end-to-end on purpose: every unit-level piece of this already passed while the bug was live -- the parse correctly returns null for a heartbeat, the commit helper correctly commits when handed a control record, the quiescence check correctly gates it. The defect was that no heartbeat ever arrived, which is only observable by running a real connector against a real source and reading the persisted offset. It snapshots a small Postgres database, leaves the source ENTIRELY IDLE, asserts the persisted offset flips to snapshot_completed=true, then restarts the connector against the same offset store and asserts the snapshot is NOT re-run (row count unchanged). It deliberately does not set heartbeat.interval.ms, so it would pass against a broken build if it did. HeartbeatIntervalDefaultTest pins the property contract cheaply: default applied, user value wins, explicit 0 honoured, blank treated as unset, null inert, the default is a positive sane duration, and the key matches Debezium exactly (a typo there would make the default silently do nothing). Verification: 11 tests pass (7 new + the 4 existing #1379 control-record tests). Mutation-checked: disabling the setProperty call fails 2 of the 7, so the suite detects the defect rather than passing vacuously. JDK 17, Maven 3.9.11. * ci: the java-lightweight job never ran any tests The step is `mvn --quiet surefire-report:report`. That is a REPORTING goal: invoked bare it does not bind to the lifecycle, so it never runs the test phase. It only formats surefire XML that already exists, and on a clean runner none does. So the job produced an empty report and passed. On this PR run it finished in 55 seconds (03:43:01 to 03:43:56) having executed zero tests, while the junit-report step printed "No annotations found for JUnit Test Report" on every run and the check stayed green. Every java-lightweight pass has been vacuous for as long as the step has looked like this. Found while trying to verify #1379 end to end: PostgresSnapshotCompletionIT sleeps well over 100 seconds by itself, so a 55-second green job could not possibly have run it. Fix: run `mvn test` first, then `surefire-report:report-only` to format what it produced. -Dmaven.test.failure.ignore=true lets the existing junit-report step be what fails the job, so a real failure is annotated on the PR instead of aborting the run before the report is written -- fail_on_failure is already true there. Jinja-Render-Check: rendered=1; formats=yaml; result=pass --------- Co-authored-by: minguyen9988 <minguyen9988@users.noreply.github.com>
…IZED (#1442) A MySQL generated column (GENERATED ALWAYS AS (expr) VIRTUAL|STORED) is currently emitted as a ClickHouse MATERIALIZED column. MATERIALIZED cannot receive the value the binlog carries, so the replicated value is discarded and the replica silently keeps whatever its own translated expression produced. ClickHouse rejects an INSERT naming a MATERIALIZED column outright: Code: 44 DB::Exception: Cannot insert column c, because it is MATERIALIZED column (ILLEGAL_COLUMN). Verified on 24.8.14. Debezium puts generated columns in the row image, so the connector must strip them from every INSERT to avoid failing the batch, and the source value can never reach the replica. That is only invisible while the two expressions agree. They can diverge when a MySQL function has no exact ClickHouse equivalent, when NULL or overflow semantics differ, when the value is session-dependent, or when the generated column is redefined on the source without the replica DDL being regenerated. The divergence raises no error, fails no batch and leaves row counts matching, so only a value-level checksum would ever surface it. DEFAULT has the same derive-when-omitted behaviour while also accepting an explicit value. Verified on 24.8.14: with c DEFAULT <expr>, INSERT (id, base) computes c and INSERT (id, base, c) stores the supplied value. The source stays authoritative and the replica still fills the column on its own. DEFAULT columns are also included in SELECT *, whereas MATERIALIZED columns are not. Constants.ALIAS is deprecated rather than removed so any other caller keeps compiling; the emitter now uses Constants.GENERATED_COLUMN_KIND. Tests: three existing expectations updated in MySqlDDLParserListenerImplTest (testGeneratedColumn, testPartitionByRange, isNotNullCreateTable), plus a new generatedColumnsAreDefaultNotMaterialized covering VIRTUAL, STORED and the bare AS (...) shorthand. It asserts on the emitted DDL rather than on a constant, so the property survives refactoring of the emitter. MySqlDDLParserListenerImplTest on this branch: 109 tests found, 108 successful, 1 skipped, 0 failures. Mutation-checked: reverting the emitter to MATERIALIZED fails 4 of them, including the new one. Note that mvn test currently discovers ZERO tests in this module on an unmodified 2.11.0 checkout too -- surefire 2.22.0 throws NoSuchMethodError EngineDiscoveryRequest.getDiscoveryListener() during discovery and reports Tests run: 0. That is pre-existing and unrelated to this change; the counts above were obtained by driving the JUnit 5 platform launcher directly. Built with JDK 17. Co-authored-by: OmniWatcher <omniwatcher@jumptrading.com>
…rows (#1445) drainBeforeDDL() pauses the writer pool BEFORE draining the handoff queue. pause() parks every pool thread inside beforeExecute(), so while it is in effect nothing can dequeue. The drain loop that follows then waits on a queue that cannot shrink: it burns the full DDL_DRAIN_TIMEOUT_MS (60s) and throws. That throw is not a safe fallback. The queued batches are discarded along with the aborted DDL attempt, so rows the source committed never reach the target. Because the loss is a tail of whatever was in flight, row counts differ by an arbitrary amount on exactly one day and every other day matches -- it does not self-heal and it does not look like lag. Seen in the field as a one-day, tail-shaped shortfall in a single table, concentrated in the highest id buckets of one partition, with no is_deleted tombstones and identical min/max ids on both sides. The deadlock requires the pool threads to be parked in beforeExecute() at the moment pause() takes effect. ClickHouseBatchRunnable. runLegacyMode keeps polling inside its own while loop once it is already inside a task body, so a batch that is mid-run when the pause lands can still drain the queue. It is therefore a race rather than a certainty -- but a likely one on a busy table, which is exactly where the queue is deep and threads cycle through beforeExecute constantly. The pause-first ordering was introduced to close a 'the queue never reaches empty on a busy table' race. That race cannot occur here. The queue has exactly one producer -- appendToRecords(), reached only from handleChangeEventBatch(), which runs on the same Debezium thread that executes the drain. No batch can be appended while the drain is running, so the queued set is already fixed and the pool is free to consume it to empty. Restore drain -> pause -> awaitQuiescent. The safety property is unchanged: the pool is still paused before awaitQuiescent(), and the check-then-act window between them is already closed inside ClickHouseBatchExecutor, where the pause test and the in-flight increment share one monitor. The abort on a genuinely undrainable queue is deliberately kept -- applying an ALTER over records captured under the previous schema is the silent corruption that guard exists to prevent. Adds DdlDrainDeadlockTest, which drives drainBeforeDDL() with a real executor, a real queue and a consumer that only runs while the pool is unpaused: - a non-empty queue drains instead of timing out (the regression); - the writer is still paused AND quiescent when the drain returns (the safety property is not traded away); - a genuinely stuck queue still aborts, and does not discard its batches. Mutation-checked: with the fix reverted, tests 1 and 2 fail with the exact error seen in the field ('DDL drain: N record batch(es) still queued after 60000 ms'), while test 3 still passes -- so the tests detect the defect rather than merely relaxing the guard.
#1446) * fix(build): the lightweight test suite has been executing zero tests in CI The java-lightweight CI job has been green while running no tests at all. Every test class dies during DISCOVERY, surefire reports 'Tests run: 0', no surefire XML is written, and the job exits SUCCESS. Two independent version conflicts, either of which is sufficient: 1. junit-bom was imported in the plain <dependencies> block. A <scope>import</scope> BOM is only honoured inside <dependencyManagement>; elsewhere it is inert. The hardcoded junit-platform-launcher / junit-platform-engine 1.8.2 pins therefore won, while junit-jupiter 5.9.1 requires platform 1.9.1. The Jupiter engine calls EngineDiscoveryRequest#getDiscoveryListener(), added in 1.9.0. 2. Both POMs hardcoded maven-surefire-plugin 2.22.0 while declaring an unused \plugin.version = 3.0.0-M7 property. 2.22.0 bundles its own pre-1.9 platform launcher, so it reproduces the same NoSuchMethodError even once the dependency versions are correct. Both are fixed: the BOM moves into <dependencyManagement>, the now-redundant junit version pins are dropped so the BOM governs, and both surefire declarations use the property already provided. Verified on the 2.11.0 tree: before: Tests run: 0 (96x NoSuchMethodError, BUILD SUCCESS) after: Tests run: 337, Failures: 4, Errors: 1 Enabling the suite surfaces pre-existing failures that were masked. They are NOT introduced here; they are repaired in the following commit so this change stays reviewable on its own: - PostgresSchemaReconcilerTest (3): expected Nullable(DateTime64(6)), got Nullable(DateTime64(6, 'UTC')) -- 3 of 9 variants. - NullParsedRecordSkipTest (1): NPE on pgConfig -- the very regression that test was added to prevent (#1379). - MySQLGenerateColumnsTest (1): environmental, needs Docker. Why this matters beyond the test count: the DDL drain regression fixed in the companion PR shipped through this pipeline. The suite that should have caught it could not fail, because it could not run. Evidence from the run that merged it (run 34325668417, job 102384143622, 2026-09-09): java-lightweight 'succeeded' in 1m41s over ~28 container-backed IT classes, with 96 NoSuchMethodError lines and 'No annotations found for JUnit Test Report'. * fix: repair the failures unmasked by enabling the test suite Enabling the suite in the previous commit surfaced 4 real failures that had been hidden behind "Tests run: 0". Fixing them here so the job goes from false-green to genuinely green rather than red on merge. PostgresSchemaReconcilerTest (3 failures) -- STALE TEST, not a code bug. ClickHouseDataTypeMapper#mapDebeziumSchemaToDDL emits an explicit 'UTC' zone for the timestamp logical types, and documents it in its own Javadoc table, so the column's zone is pinned instead of inheriting the server's local timezone. The assertions still expected the pre-#1435 zone-less form: expected: Nullable(DateTime64(6)) actual: Nullable(DateTime64(6, 'UTC')) expected: Nullable(DateTime64(3)) actual: Nullable(DateTime64(3, 'UTC')) expected: Nullable(DateTime64(9)) actual: Nullable(DateTime64(9, 'UTC')) `git log -S` puts the mapper change in 078d9ce (#1435) -- the same commit that shipped the DDL drain regression. The expectations were never updated with it, and CI could not object because it ran nothing. The ZonedTimestamp case in the same list already expected the 'UTC' form, which is why only 3 of the 9 variants failed. NullParsedRecordSkipTest (1 failure) -- REAL PRODUCT BUG. processEveryChangeRecord dereferenced `pgConfig` unguarded: if (pgConfig.getSchemaChangeDetector() != null) `pgConfig` is assigned in setup(), so it is null on any path reaching record processing without it. The NPE was thrown BEFORE parse(), which re-broke the very contract this branch exists to preserve (#1379): the unparseable record was dropped without being acknowledged, instead of being skipped with a warning. Schema drift detection is PostgreSQL-only and optional, so its absence must skip the check, never fail the record. Guarded with a null check. That is the second time a guard test for #1379 was red while CI was green. MySQLGenerateColumnsTest -- environmental only; it needs Docker and errors with "Could not find a valid Docker environment" where none is present. It passes on a runner with Docker and is unchanged here. Verified on the 2.11.0 tree with a container runtime available: before this commit: Tests run: 337, Failures: 4, Errors: 1 after this commit: Tests run: 337, Failures: 0, Errors: 0, Skipped: 1 The single skip is a pre-existing @disabled case in MySqlDDLParserListenerImplTest. * ci: fail the job when the test suite does not actually run A green job proves nothing unless tests EXECUTED. This job has now passed while running zero tests twice, for two unrelated reasons: 1. `surefire-report:report` never bound to the lifecycle, so the test phase never ran (fixed earlier in this workflow file). 2. A junit-platform version conflict killed every test class during DISCOVERY: 96x NoSuchMethodError, "Tests run: 0", BUILD SUCCESS (fixed in the preceding commits). Both were invisible for the same reason: nothing asserted a floor. A suite that silently stops running looks exactly like a suite that passes. Adds a "Verify tests actually ran" step that reads the surefire XML and fails the job when: - no surefire XML was produced at all; - fewer than MIN_TOTAL_TESTS (250) tests executed; - no *IT class produced results. The unit and integration counts are checked SEPARATELY on purpose. The end-to-end tests are the ones that would have caught the DDL drain regression, and an IT-only collapse -- containers unavailable on the runner, for instance -- would otherwise hide behind a healthy unit count. Requiring IT results makes "the E2E suite did not run" a build failure rather than a silent omission. Floors are deliberately set below current levels (337 tests observed) so that ordinary test removal does not trip them; they catch collapse, not drift. The error message says not to lower the floor to make the job pass, because that is the obvious wrong fix. Runs with `if: always()` so it still fires when tests failed -- `-Dmaven.test.failure.ignore=true` means the mvn step itself does not fail the job. Verified against real surefire output and all four cases: no XML at all -> rc=1 "test suite did not run" XML present, tests="0" -> rc=1 both floors reported 300 unit tests, zero ITs -> rc=1 IT floor catches it 305 tests including one IT class -> rc=0 passes The third case is the one that matters: a healthy-looking unit count still fails when the end-to-end suite is absent. The fourth confirms this is a real gate and not an always-fail.
…he flush semaphore (#1447) acknowledgeRecords() serializes only markBatchFinished() (via markBatchFinishedSafely) and leaves markProcessed() outside the lock. That is not sufficient: markProcessed() mutates the OffsetStorageWriter's pending-offset map, which is exactly the state beginFlush() snapshots, and neither is thread-safe. Debezium also builds a NEW RecordCommitter per batch (EmbeddedEngine.buildRecordCommitter), so the committer's own synchronized methods lock different monitors for different batches and provide no mutual exclusion across worker threads. Once two worker threads collide, EmbeddedEngine.commitOffsets returns early when doFlush returns null WITHOUT calling cancelFlush, permanently leaking the flushInProgress semaphore. Every subsequent beginFlush() in the JVM throws ConnectException: OffsetStorageWriter is already flushing. Two further behaviours turn a stuck writer into SILENT DIVERGENCE rather than a visible failure: - ClickHouseBatchRunnable routes the failure through ClickHouseErrorClassifier, which sees no ClickHouse error code and defaults it to UNKNOWN/retriable. With errors.max.retries = -1 it retries forever. - JdbcOffsetBackingStore.set is asynchronous, so ClickHouse inserts keep succeeding against a frozen binlog position. Observed in production on a MySQL -> ClickHouse deployment running 2.11.0: the connector logged healthy batches for ~8h while its committed offset never advanced past an event 8 hours old, and on restart the binlog client resumed nine binlog files (~10 GiB) behind the position the dying process had been reading. 3,930 occurrences of the exception in that window. Three sibling connectors on 2.9.1 against the same ClickHouse host logged zero, as did a second 2.11.0 connector running roughly a tenth of the volume -- the defect needs two batch completions in flight. Changes: - DebeziumOffsetManagement: markProcessed() and markBatchFinished() now run inside ONE OFFSET_COMMIT_LOCK critical section in both acknowledgeRecords variants. markBatchFinishedSafely is replaced by a public acknowledgeRecord entry point so every offset-committing path funnels through the same lock. - ClickHouseBatchWriter: its second, unserialized acknowledgement path -- which called markProcessed()/markBatchFinished() directly, bypassing the lock entirely -- now routes through acknowledgeRecord. A poisoned writer is rethrown rather than logged and swallowed, and an InterruptedException re-sets the interrupt flag instead of continuing to acknowledge. - ClickHouseBatchRunnable: a poisoned writer is detected BEFORE the ClickHouse classifier and stops the task, so it can never be retried forever as UNKNOWN. A restart resumes from the last committed offset. No lock-order change: acknowledgeRecord takes only OFFSET_COMMIT_LOCK, never the class monitor. Testing (2.11.0 @ af198d9, java-17, maven 3.9.11): OffsetCommitSerializationTest reproduces the defect deterministically with a committer that fails if any two offset-mutating calls overlap. with this fix: 3 run, 0 failures acknowledgeRecord reverted to the partial-lock shape: 3 run, 2 FAILURES, both naming markProcessed as the overlapping call Full module suite, same tree, no Docker available on the runner: unmodified 2.11.0: Tests run: 344, Failures: 0, Errors: 20 with this fix: Tests run: 347, Failures: 0, Errors: 20 The 20 errors are identical in both runs and are all "Previous attempts to find a Docker environment failed" from the container-backed tests; they are pre-existing and unrelated. Co-authored-by: svc_dba_github <svc_dba_github@fpif1-vcsl1.w2k.jumptrading.com>
…se the derivation if it were not (#1448) * fix(ddl): the MATERIALIZED-to-writable ALTER is a no-op, and would lose the derivation if it were not makeColumnWritable() exists to stop a ClickHouse MATERIALIZED column shadowing the value the source supplies. It emits ALTER TABLE `db`.`t` MODIFY COLUMN `c` <type> on the belief that restating the column without a default clause drops the MATERIALIZED expression. It does not. ClickHouse reads an omitted default clause as "leave the existing default alone": the statement is accepted, returns no error, and default_kind stays MATERIALIZED. The source value goes on being discarded, and the post-ALTER verification added alongside it is the only reason this surfaces at all -- it logs "Enforcement did not take effect" forever, once per DDL generation, and the divergence is never corrected. Verified in an isolated clickhouse-local sandbox, reproduced identically on 24.8.14, 25.8.12 and 26.1.6: CREATE TABLE t (id Int64, base Nullable(Int64), c Nullable(Int64) MATERIALIZED ifNull(base,0)*2) ALTER TABLE t MODIFY COLUMN c Nullable(Int64) -> default_kind: MATERIALIZED (unchanged, no error) ALTER TABLE t MODIFY COLUMN c Nullable(Int64) DEFAULT ifNull(base,0)*2 -> default_kind: DEFAULT Making that statement effective by stripping the default instead would be worse than leaving it broken. A MATERIALIZED column is derived: the replica computes it whenever the connector omits it. A bare column derives nothing, so every row written without that column would store a type zero -- trading a value divergence for a data-loss path, on tables whose source declares the column generated and therefore frequently does omit it. DEFAULT is the kind that satisfies both halves. It accepts an explicit value, so the binlog value lands as sent and the source becomes authoritative; and it re-derives the value when the column is absent, exactly as MATERIALIZED did. Same sandbox, same builds: INSERT INTO t (id, base) VALUES (1, 5) -> c = 10 (derived) INSERT INTO t (id, base, c) VALUES (2, 5, 999) -> c = 999 (source wins) This is the same conclusion #1442 reached for the CREATE TABLE path, where a MySQL generated column is now translated to DEFAULT rather than MATERIALIZED. This commit brings the ALTER path into agreement with it; until now the two disagreed, so a table created before #1442 could never be corrected into the shape #1442 creates. Changes: - DBMetadata#makeColumnWritable now emits MODIFY COLUMN <type> DEFAULT <expr>. The expression is read back from system.columns and restated verbatim rather than reconstructed, so the replica keeps deriving precisely what it derived before. CODEC, COMMENT and column TTL are carried across by ClickHouse itself -- verified on the same builds by comparing create_table_query either side of the conversion. - DBMetadata#getColumnDefaultExpression: new read-only introspection, the counterpart of the existing getColumnType/getColumnDefaultKind pair. - When the expression cannot be read the conversion is abandoned and no DDL is issued. Emitting the bare form in that situation is the no-op above, and any statement that did take effect would remove the derivation. - The post-ALTER verification now asserts the column reads back as DEFAULT, rather than asserting it is no longer MATERIALIZED. The negative form also accepts a column left with no default at all, which is the outcome this change exists to avoid. - Caller log messages and javadoc in GroupInsertQueryWithBatchRecords updated: they described dropping the expression, which is neither what happens nor what should. Behaviour is unchanged for ALIAS (still skipped, stores nothing) and for ordinary columns. makeColumnWritable has exactly one production caller and its signature and return contract are unchanged. Testing (2.11.0 @ 50cd189, JDK 17, maven 3.9.11): UnwritableColumnReportingTest: 16 run, 0 failures. Mutation-checked, each mutation caught by a different test: emitter reverted to the bare MODIFY COLUMN -> 3 failures verification reverted to "not MATERIALIZED" -> 1 failure unreadable-expression guard disabled -> 1 failure Full module suite, same tree: unmodified 2.11.0: Tests run: 372, Failures: 0, Errors: 20 with this change: Tests run: 376, Failures: 0, Errors: 20 The 20 errors are identical in both runs -- "Previous attempts to find a Docker environment failed" from the container-backed tests. Pre-existing and unrelated; no Docker on the build host. * fix: bind the identifiers in the default-expression lookup Review follow-up. getColumnDefaultExpression interpolated databaseName, tableName and columnName into single-quoted SQL literals. They are replicated identifiers, so one containing a single quote makes the query malformed -- and the failure is invisible: it is caught, logged, and reported as "no expression", which the caller reads as "abandon the conversion". The column then stays MATERIALIZED with no indication why. Bound as parameters instead. The statement is also now closed: the previous try-with-resources held only the ResultSet, and closing a ResultSet does not close the Statement that created it, so every enforcement attempt leaked one. Both were introduced by the preceding commit; neither exists elsewhere in this change. A third review finding -- that a '?' inside the default expression would be parsed as a bind placeholder by prepareStatement, failing the DDL -- was checked and is not real. Driving clickhouse-jdbc 0.6.0's own JdbcParameterizedQuery over the emitted DDL reports zero parameters for DEFAULT concat(name, '?') (quoted literal) DEFAULT if(amount > 0, ..., 0) (how ClickHouse renders a ternary) DEFAULT x ? y : z (raw, never produced by system.columns) against a positive control of INSERT ... VALUES (?, ?) reporting 2 and SELECT ... WHERE id = ? reporting 1, so the probe does detect placeholders when they are present. ClickHouse also normalises a ternary to if() in default_expression, verified on 24.8.14/25.8.12/26.1.6, so the raw form cannot reach this path in the first place. No change made. Testing: UnwritableColumnReportingTest 17 run, 0 failures. The new testExpressionLookupBindsIdentifiersRatherThanInterpolating asserts the SQL carries placeholders and not the identifier values; mutation-checked by restoring the interpolated form, which fails it. Full module suite 377 run, 0 failures, same 20 pre-existing Docker errors as the unmodified base. --------- Co-authored-by: OmniWatcher <omniwatcher@users.noreply.github.com>
…erts (#1449) testModifyColumn issues six ALTERs against the source and then polls ClickHouse until the schema has caught up. The poll predicate checks two columns: class_name == Nullable(Int32) && col2 == Nullable(Int32) and the assertions that follow check five, including ship_class.tonange == Nullable(Decimal(10, 10)). The ALTERs are replicated one at a time, so the replica passes through intermediate states where some are applied and others are not. Waiting on a subset and asserting on a superset makes the outcome depend on where the replication stream happened to be when the last poll ran, which is why this test fails intermittently. Observed in CI (run 34733893151, job java-tests-lightweight): 04:43:30.987 Executed Source DB DDL: alter table ship_class modify column class_name int 04:43:30.988 ClickHouse DDL: ALTER TABLE ship_class MODIFY COLUMN class_name Nullable(Int32) 04:43:31.756 testModifyColumn FAILED at line 89 (tonange) Only the first of the six ALTERs had been replicated when the loop exited. tonange was still Nullable(Decimal(10, 2)) -- its ALTER, and the remaining four, arrived afterwards (the ship_class tonange conversion is visible later in the same log at 05:19:56.186). The failure is also unreadable. Assert.assertTrue over equalsIgnoreCase reports exactly "java.lang.AssertionError": not which column failed, not what it held. And if the column is absent rather than stale, .equalsIgnoreCase() is invoked on null, so the test errors with NullPointerException instead of failing with a diagnosis. Changes (test-only; no production code is touched): - The expected columns are declared once, in a map, and the poll predicate is driven from that same map. The loop now waits for exactly the set the assertions check, so it cannot exit while any of them is still stale. - Assertions report the table, the column, the expected value and the OBSERVED value, and a missing column fails with "the ALTER was not replicated" plus the columns that are present, rather than throwing NPE. - Timeout is unchanged (10 x 5s). A genuinely broken replication path still fails, and now says what was wrong. Verification: the new predicate and the old one were both driven against the exact schema state recorded in the failing run. At that state the old predicate returns true (the loop exits -> the failure) and the new one returns false (the loop keeps waiting). Against the fully converged state the new predicate returns true, so the loop still terminates rather than always spending its full budget. Against a state with the column absent it returns false instead of raising NullPointerException. A positive control confirms the old predicate can return false, so its true above is a real result. Built with JDK 17, maven 3.9.11; sink-connector-lightweight test-compile passes. The four other failures in that CI run (PostgresInitialDockerIT, PostgresInitialDockerWKeeperMapStorageIT, PostgresSnapshotCompletionIT, MySQLDemoIT) are present on the unmodified base commit 50cd189 and on af198d9 before it; they are untouched here. Co-authored-by: OmniWatcher <omniwatcher@users.noreply.github.com>
…d commits ranks above the earlier write of its key (#1452) * fix(version): a commit that reaches the binlog after newer-timestamped commits ranks above the earlier write of its key On MySQL, source.ts_ms of a row event is the timestamp of the STATEMENT that produced it, not of the commit. A transaction that stays open while others commit reaches the binlog after them, carrying an OLDER timestamp. The _version sequence (formula ts*1e6+counter, counter reset on a >1s advance of the source clock) versioned such a late event by its own older timestamp; when the newer-timestamped commits in between had reset the counter, the late UPDATE ranked BELOW the earlier write of the same key (same source second, higher counter). ReplacingMergeTree kept the stale row and discarded the newer one; row counts matched on both sides and only a value-level checksum reported it (observed on a production replica running 2.11.0 with gtid_mode=OFF: rows left open after the source closed them, rows with NULL blobs the source had populated). The counter reset is visible in the stored versions of consecutive keys: ...36254, ...35957, ...36064, then ...00208 - the UPDATE events for the middle keys landed after that reset and lost. #1346 (redelivery-stable versions anchored to the source timestamp) is kept. What was missing is the distinction between a first delivery and a redelivery, which the binlog position gives for free: Debezium delivers in log order, log order IS commit order, so the only time a lower position follows a higher one is a redelivery after an offset rewind. - SourcePosition (new): comparable (binlog file number, pos, row) for MySQL/MariaDB - a rotation resets pos but the file number orders correctly - and lsn for PostgreSQL. Records without coordinates (heartbeats, MongoDB, synthetic records) have none. - ClickHouseStruct: read the position from a change event source struct and from the struct's own file/pos/row/lsn fields. - DebeziumChangeEventCapture.nextSequenceNumber: one assignment used by the streaming loop and addVersion. A first delivery (position above the run's high-water mark) has its timestamp component floored at the newest timestamp already versioned, so _version stays monotonic in commit order; every record that goes through the sequence raises that floor (a heartbeat with a newer envelope timestamp resets the counter exactly like a newer commit does). A redelivery (position at or below the mark) and a record without a position keep the #1346 assignment unchanged. Formula, SEQUENCE_START/SEQUENCE_START_INITIAL domains and anchor semantics are untouched, so upgrade and downgrade stay safe. Tests: CommitOrderVersionClampTest reproduces the production sequence at unit level (fails on the previous code with T*1e6+1e9+1 against T*1e6+1e9+50), the late DELETE variant, the heartbeat-driven counter reset, in-batch ordering, the #1346 redelivery case (unchanged result asserted to the exact value), binlog rotation, multi-row events, counter reset on the floored clock, no-position records (unchanged) and PostgreSQL LSNs. LateCommitVersionOrderIT is the end-to-end reproduction against real MySQL (gtid_mode=OFF, the sequence-number path) and ClickHouse containers: SET TIMESTAMP pins the statement time MySQL writes into the binlog header; a 50-row transaction plus the key's write in second T; a second session executing the key's UPDATE and a DELETE in second T that commits after an unrelated T+5 commit. Verified value-level against MySQL over every row, plus the _version ordering and proof that the pinned timestamps reached the connector. SourcePositionTest covers ordering, rotation, numeric vs lexical file names, missing coordinates, LSNs and struct extraction. SourceTsVersionAnchorTest (the #1346 pins) is unchanged. * test: make the late-DELETE, rotation and multi-row cases fail without the fix Mutation check with the floor disabled: three cases placed the key's earlier write at counter +0 of its second, which the unfixed code still out-ranks by +1, so they passed the mutant. The earlier write now sits at a higher counter in each (other writes of the same second precede it); all three are red without the floor and green with it. * fix(position): a total order for mixed binlog file names; IT uses the container's own readiness check Comparing numerically-suffixed names numerically and everything else lexically is not transitive once the two kinds are mixed (log.20 before log.100, log.100 before log.15a, but log.20 after log.15a). SourcePosition now orders lexicographically on (file prefix, numeric-suffixed first, sequence or full name, pos, row) - a total order; LSN positions carry no file name (fileSequence -1, empty prefix). SourcePositionTest pins the transitivity case. The HttpWaitStrategy override on the MySQL container (copied from the sibling ITs) was inert - MySQLContainer runs its own JDBC probe regardless - and is removed. --------- Co-authored-by: minguyen9988 <minguyen9988@users.noreply.github.com>
…ey assert (#1453) The java-lightweight job has executed the suite for real since the surefire fix, and four classes fail deterministically, in CI and locally alike (449 run, 2 failures, 2 errors on the 2.11.0 head). None of them is a product regression; each reads state before the pipeline could have produced it. - PostgresInitialDockerIT / PostgresInitialDockerWKeeperMapStorageIT: read the persisted offset once, right after the snapshot rows became visible, and NPE'd on null. The offset is flushed on Debezium's schedule, and on an idle source the end-of-snapshot state travels on the first heartbeat, emitted only once streaming has started. Both now poll (ITCommon.waitForOffset) until the offset says the snapshot is finished, then assert the position fields every PostgreSQL offset carries. The old assertions on last_snapshot_record / snapshot keys could never hold once the offset had moved to its streaming form. - PostgresSnapshotCompletionIT: waited for the literal snapshot_completed:true. Debezium writes the snapshot keys only while the offset context is in snapshot mode; the offset the first post-snapshot heartbeat persists is a plain streaming offset (lsn/txId/ts_usec) with no snapshot keys, which a restarted connector resumes from without re-snapshotting. The test now accepts any offset that does not report snapshot_completed:false (ITCommon.offsetSaysSnapshotFinished) - the state issue #1379 left behind forever is still rejected - and keeps its restart/no-re-snapshot proof. - MySQLDemoIT.testCustomersAndPayments: waited for the post-delete counts with waitForRowCount(..., 0, ...), which returns as soon as the count is at or above its floor - i.e. immediately - and then compared MySQL's 1/1 against ClickHouse's still undeleted 4/3. New ITCommon.waitForRowCountEquals waits for the exact post-delete counts. Co-authored-by: minguyen9988 <minguyen9988@users.noreply.github.com>
) The three Java jobs resolve 17 against the Temurin metadata API on every run: the runners are ephemeral, so the tool cache is empty and setup-java always goes to the network before a single line of Java executes. One dropped TLS handshake there fails the whole job. On the 2.11.0 pull-request pipeline, java-tests-kafka died 32 seconds in with: Client network socket disconnected before secure TLS connection was established (run 35231232254), on a head whose kafka suite had passed on the run immediately before and whose lightweight suite passed in the very same run. actions/setup-java@v3 has no retry around that request, so retry the step itself: the first attempt carries continue-on-error, and an identical second attempt runs only when the first did not succeed. steps.setup-jdk.outcome is the step result BEFORE continue-on-error is applied, so it is the right signal; the job still fails on a genuine second failure, and a successful first attempt skips the retry entirely. The post-step of a failed first attempt cannot fail the job either: the setup-java cleanup only saves the Maven cache when the job status is success and wraps the save in a best-effort warning. Applied to the three workflows that call setup-java (docker-build, sink-connector-kafka-tests, sink-connector-lightweight-tests) so the pipeline outcome does not depend on which job the blip lands in. No test or product code is touched. Co-authored-by: minguyen9988 <minguyen9988@users.noreply.github.com>
#1455) A bitemporal migration that adds a surrogate AUTO_INCREMENT PRIMARY KEY and tightens just-added columns to NOT NULL produced ClickHouse DDL the server rejects. Because DDL is retried indefinitely, each rejection stalled the whole replication stream. 1. ADD PRIMARY KEY produced a malformed ALTER. A ClickHouse ReplacingMergeTree fixes its sorting key at CREATE, so ADD PRIMARY KEY has no ALTER equivalent and the translator emits nothing for it. But the comma separators around the dropped clause were still written, yielding a leading comma ("ALTER TABLE t, MODIFY ...") or a bare "ALTER TABLE t" -- both Code: 62 Syntax error. Handle AlterByAddPrimaryKey like the ALGORITHM/LOCK hints (emit nothing, drop the separator), emit clause separators lazily so a no-op clause cannot leave a leading/doubled comma, and skip the statement entirely when every specification is a no-op. The surrogate id COLUMN is still added and replicated by the accompanying ADD COLUMN clause, so no source value is lost. 2. MODIFY/CHANGE COLUMN ... NOT NULL emitted a non-Nullable MODIFY. When the column already exists as Nullable in ClickHouse -- which is what this translator emits for a preceding ADD COLUMN in the same migration -- converting Nullable -> non-Nullable requires a DEFAULT expression, so the server rejects it with Code: 36 BAD_ARGUMENTS. Honor NOT NULL only for ADD COLUMN (a new column has no existing rows to violate it); for MODIFY/CHANGE keep the column Nullable. Nullable(T) is a superset of T, so no source value is lost, no fabricated DEFAULT overwrites existing rows, and the change is checksum-safe (comparison is value-level). 3. The DDL error logger inserted into system.`null`. The DDL path passed the raw ERROR_TABLE_NAME property (null when unset) into the INSERT while createErrorTable() defaulted it, so logging one failure raised a second Code: 60 UNKNOWN_TABLE "Table system.`null` does not exist" and the real error was never recorded. Centralize the default in ErrorLogger.resolveErrorTableName and use it on both paths. Adds unit regression tests for all three (parser output and error-table resolution). Two existing MODIFY/CHANGE NOT NULL assertions expected the non-Nullable form that triggered defect 2; they are updated to the corrected Nullable form.
…Y and MODIFY COLUMN NOT NULL (#1457) Verify end-to-end replication against MySQL and ClickHouse containers: - ADD PRIMARY KEY alone is skipped without producing Code: 62 syntax error. - ADD PRIMARY KEY as first clause does not leave a leading comma. - MODIFY COLUMN ... NOT NULL keeps the ClickHouse column Nullable, avoiding Code: 36 (Please specify DEFAULT expression) and avoiding stream stalls. - Data inserted after the DDLs replicates successfully.
…ementations and fix drift cache invalidation (#1458)
* test: fix the two CI failures on 2.11.0 left by #1455 and #1457 java-tests-lightweight: AlterTableModifyColumnIT.testAlterAddPrimaryKeyAndModifyNotNull (added in #1457) fails inside MySQL before anything reaches the connector: java.sql.SQLSyntaxErrorException: Multiple primary key defined at AlterTableModifyColumnIT.testAlterAddPrimaryKeyAndModifyNotNull(AlterTableModifyColumnIT.java:154) Step 3 issues "ALTER TABLE office ADD PRIMARY KEY (office_id), ..." but alter_ddl_modify_column.sql already creates office with "office_id INT PRIMARY KEY", so MySQL rejects the second key and the multi-clause translation the step exists to exercise never runs. Add a keyless table (branch) to the init script and run the step against it, and assert the outcome on the ClickHouse side as well: branch_name stays Nullable(String) (MODIFY ... NOT NULL on an existing column), branch_status is String (ADD COLUMN ... NOT NULL is honored), and the ADD PRIMARY KEY clause emitted nothing. testflows (lightweight, hikari-pool and replicated): the features "alter/modify column not null" and "alter/change column name to new name not null" time out after 100s waiting for DESC to show "x","String" / "x2","String". Since #1455, MODIFY/CHANGE COLUMN ... NOT NULL keeps the existing ClickHouse column Nullable (converting Nullable -> non-Nullable needs a DEFAULT expression, otherwise Code: 36 stalls the stream), so the column now reads Nullable(String). #1455 updated the Java unit assertions to the Nullable form but not these two testflows expectations; align them and record the rationale in the feature docstrings. The 2x4 failing examples across the three testflows jobs were exactly these two features, and no other suite asserts a non-Nullable result after MODIFY/CHANGE ... NOT NULL. * test: make DECIMAL sample values in-range so is_deleted stops flaking The `is deleted different datatypes / check is_deleted with DECIMAL(2,1) NOT NULL` example failed in only one of the four testflows-lightweight jobs on 2.11.0 (x86 replicated), passing in the other three, and failed fast (~6.8s) at the INSERT rather than timing out on replication -- the signature of a bad generated value, not infrastructure flakiness. generate_sample_mysql_value() drew DECIMAL values from round(random.uniform(-(10**(p-s)), 10**(p-s)), s) but DECIMAL(p, s) admits |value| <= (10**p - 1)/10**s, i.e. 9.9 for DECIMAL(2,1), not 10.0. Any draw >= max - 0.5*10**-s rounds to exactly 10**(p-s), one unit of the last place out of range. Under MySQL 8.0 strict mode the INSERT then fails with "Out of range value", which fails the test at random -- about a 0.5% chance per run for DECIMAL(2,1), matching one-in-four jobs. Generate an integer number of least-significant-digit units over the exact valid range [-(10**p - 1), 10**p - 1] and place the decimal point by string. Every value is in range by construction, so the test is deterministic; building the string from the int also avoids scientific notation (which MySQL rejects) for high-scale types and the default Decimal context's 28-digit precision limit for DECIMAL(30, 10). Verified over 300k draws each for DECIMAL(2,1), DECIMAL(30,10), DECIMAL(10,2), DECIMAL(5,0) and DECIMAL(1,1): zero out-of-range, zero scientific-notation. The only callers are is_deleted.py (all DECIMAL datatypes) and multiple_databases.py (INT/VARCHAR only, unaffected).
…rmal Lean verification (#1460) * feat(specs): transition repository to spec-driven development with formal Lean verification Establish a ground-truth Specification-Driven Development (SDD) framework across the repository using the Smart Ralph protocol, and introduce a complete formal verification model in the Lean 4 proof assistant. Add CONSTITUTION.md, SMART_RALPH_PROTOCOL.md, component specs 001-009 matching 2.11.0, Lean 4 formal verification suite under formal_specs/lean/, and validation tooling. * feat(specs): modularize specifications into 11 encapsulated domains with 57 micro-specs Decompose coarse component specs into smaller, highly encapsulated micro-specifications across 11 architectural domains (01-cdc-engine through 11-verification-tooling). Update validate_specs.py to recursively validate all 57 micro-specs, governance documents, and Lean formal model linkages.
A DDL change event is handled inside processEveryChangeRecord, whose
trailing catch-all (catch (Exception e) { log.error("Exception
processing record", e); }) exists to keep one malformed DML record from
killing the stream. A DDL failure fell into that same catch and was
swallowed: drainBeforeDDL() aborting, or performDDLOperation() exhausting
its retries, threw, was logged, and the method returned normally. The
batch loop then continued, committed offsets past the unapplied DDL, and
every later row was written against a ClickHouse schema that no longer
matched the source -- a silent, count-clean divergence. In
single-threaded mode the same path dereferenced a null executor and
dropped the first DDL with an NPE.
Fix:
- Add DDLReplicationException, raised for any DDL-path failure and
re-thrown ahead of the catch-all so it leaves handleBatch and halts the
engine. Offsets are not advanced past the DDL; a restart re-delivers it
from the last committed position -- the genuine retry the drain abort
was always meant to enable.
- performDDLOperation now throws DDLReplicationException on retry
exhaustion (was a plain RuntimeException the catch-all absorbed).
- drainBeforeDDL returns immediately when there is no worker pool
(executor == null, single-threaded mode): rows are persisted inline
before the DDL record, so there is nothing to drain -- no NPE, no
dropped DDL. The finally resume() is guarded the same way.
Spec-first per the SDD mandate: specs 06.08 and 10.04 updated to declare
the loud-failure contract for the DDL path.
Tests (DdlFailureLoudTest, no engine required):
- ddlFailurePropagatesInsteadOfBeingSwallowed: a DDL whose drain aborts
throws DDLReplicationException out of processEveryChangeRecord instead
of returning null. Mutation-checked: removing the re-throw makes it
fail with the original swallow behaviour.
- drainIsNoOpWhenExecutorIsNull: single-threaded drain is a no-op, not an
NPE.
DdlDrainDeadlockTest still passes unchanged (drain still aborts a
genuinely undrainable queue and stays quiescent+paused on return).
… not fatal (#1462) ClickHouseErrorClassifier put 252 TOO_MANY_PARTS in FATAL_ERROR_CODES, under "resource limits (deterministic for a given batch)". 252 is not deterministic: it is ClickHouse insert backpressure raised when the active part count exceeds parts_to_throw_insert, and it clears on its own as background merges catch up -- the SAME batch succeeds on retry once the part count falls. Classifying it FATAL made ClickHouseBatchRunnable rethrow and stop the task, halting replication for EVERY table on a transient, self-healing condition and requiring a manual restart. The correct behaviour is to let it fall to the RETRIABLE default and be retried with backoff on the next scheduled run. Retrying is safe: offsets never advance past an unwritten batch (checkIfBatchCanBeCommitted only commits on success), so there is no divergence risk. 241 MEMORY_LIMIT_EXCEEDED is deliberately left FATAL -- a batch too large for the memory budget will deterministically fail again; whether it should be retriable under transient concurrent memory pressure is a separate decision, noted in the spec. Spec-first per the SDD mandate: spec 10.01 rewritten to match the real classifier -- correct source path, the real enum name (ErrorCategory, not CLICKHOUSE_ERROR_SEVERITY), the actual FATAL set, and the 252 rationale. The prior 10.01 listed a wrong path, a wrong enum, and codes (44, 36, 159) that the classifier does not contain. Tests (ClickHouseErrorClassifierTest): - testTooManyPartsIsRetriableBackpressure (new): 252 -> RETRIABLE and isFatal(252) == false. Mutation-checked: restoring 252 to FATAL fails this and testIsFatal. - testClassifyFatal / testIsFatal updated to drop 252 from the fatal set. 7 tests pass on JDK 17; scripts/validate_specs.py passes (exit 0).
…theorems (#1463) The formal verification suite never compiled and proved nothing: lakefile.lean did not parse (invalid Lake fields), Invariants.lean used the reserved keyword "prefix" as a binder, Proofs.lean carried 8 sorry across 5 of 6 theorems, and the model was unfaithful (translateEvent truncate produced no records while the source empties MySQL, so convergence was false for any stream with a table-clear; and encodeVersion overflowed its fields so version monotonicity was not true). Changes: lakefile parses and builds; rename the reserved binder to "replayed"; name the max-version fold step and inline chFinalView; make CHTable an abbrev; version by strictly increasing stream ordinal (live 2*i, relocation tombstone 2*i-1) so convergence needs no monotonic-position hypothesis; the table-clear event now resets the accumulated table in lock-step with the source; encodeVersion kept only as a documented order-witness proved monotonic within a new BinlogPos.WellFormed domain, not used by the engine; complete sorry-free proofs of all six theorems on self-contained core-only helper lemmas plus a general convergence lemma. Verification: lake build compiles with zero errors and zero sorry; print axioms on every theorem lists only [propext, Quot.sound] (no sorryAx); validate_specs.py now rejects any sorry/admit in the Lean sources; spec 11.03 and the Lean README updated to the real theorem names and the ordinal-versioning model. Destructive-Op-Check: sites=3; result=pass
…stamp range (#1464) DebeziumOffsetManagement keyed inFlightBatches and completedBatches by the batch (minTs,maxTs) timestamp range. Two DISTINCT batches routinely share a range - a single multi-row statement split across batches, or two batches whose rows all fall in the same millisecond. Keying by the range collided them: put overwrote the earlier batch entry, remove deleted the wrong one, and the overlap check "skip the same batch" step (also range-based) treated a distinct sibling as itself. An unwritten older batch therefore stopped blocking the offset commit, so the committed binlog position could advance past rows not yet in ClickHouse and lose them on a crash. Fix: introduce BatchKey, an identity wrapper over the batch List (equals/hashCode by System.identityHashCode), and key both maps by it. The overlap check now skips the current batch by identity (==) and compares timestamp ranges only to decide overlap. addToBatchTimestamps/checkIfBatchCanBeCommitted/acknowledgeRecords and the unused accessors are retyped accordingly; timestamp ranges are still derived for the overlap comparison, never used as a key. Tests (OffsetBatchIdentityTest): two same-range batches are tracked as two entries; a distinct same-range sibling still blocks the commit; a batch does not block itself; identity removal leaves the sibling. Mutation-checked: making BatchKey compare by range reproduces the collision and fails 3 of the 4 tests. HandedOffBatchVisibilityTest still passes. Spec 09.01 updated to the identity-key design and its corrected source path.
…dering (#1465) * fix(offset): key in-flight/completed batch maps by identity, not timestamp range DebeziumOffsetManagement keyed inFlightBatches and completedBatches by the batch (minTs,maxTs) timestamp range. Two DISTINCT batches routinely share a range - a single multi-row statement split across batches, or two batches whose rows all fall in the same millisecond. Keying by the range collided them: put overwrote the earlier batch entry, remove deleted the wrong one, and the overlap check "skip the same batch" step (also range-based) treated a distinct sibling as itself. An unwritten older batch therefore stopped blocking the offset commit, so the committed binlog position could advance past rows not yet in ClickHouse and lose them on a crash. Fix: introduce BatchKey, an identity wrapper over the batch List (equals/hashCode by System.identityHashCode), and key both maps by it. The overlap check now skips the current batch by identity (==) and compares timestamp ranges only to decide overlap. addToBatchTimestamps/checkIfBatchCanBeCommitted/acknowledgeRecords and the unused accessors are retyped accordingly; timestamp ranges are still derived for the overlap comparison, never used as a key. Tests (OffsetBatchIdentityTest): two same-range batches are tracked as two entries; a distinct same-range sibling still blocks the commit; a batch does not block itself; identity removal leaves the sibling. Mutation-checked: making BatchKey compare by range reproduces the collision and fails 3 of the 4 tests. HandedOffBatchVisibilityTest still passes. Spec 09.01 updated to the identity-key design and its corrected source path. * fix(routing): wire hash routing to per-thread queues for per-table ordering The hash-routing path was dead: the routed queue was never assigned, so run() always fell back to runLegacyMode and every worker thread polled ONE shared records queue. Same-table batches were therefore processed concurrently by different threads, out of source order — losing per-table ordering (Invariant I1), which matters for TRUNCATE, CollapsingMergeTree, and the relocation tombstone tie. Fix: give each worker thread its OWN routed queue (routedQueues, index == thread id). appendToRecordsWithHashRouting routes every batch for a table to that tables single owning threads queue by a stable hash; that thread (runWithHashRouting) drains only its own queue in FIFO order. run() now selects runWithHashRouting when a routed queue is present, else runLegacyMode (single-thread / legacy mode keeps the shared records queue). isPipelineQuiescent checks every routed queue. Per-thread queues are deliberate: a single shared queue that workers filter by thread id (putting non-matching batches back at the tail) reorders same-table batches under contention — re-enqueuing a sibling batch moves it behind later batches for the same table. Per-thread queues remove that race, so the old put-back branch is gone. Tests (HashRoutingPerTableOrderingTest): a table routes to exactly its one owning queue; successive batches for a table stay on the same queue (FIFO); each table routes to its hash-assigned queue. Mutation-checked: routing to the wrong queue fails all three. Specs 03.02 and 03.03 updated to the per-thread queue design and corrected mode selection.
…1466) Guarantees that a newer connector is a drop-in replacement for an older one: upgrading in place 2.8.0 / 2.9.1 / 2.10.x -> 2.11.0 (same ClickHouse tables, same persisted offset store, same schema history, same config) does NOT ruin data already in ClickHouse. Rows written by the old version and rows written by the new version coexist in one ReplacingMergeTree table and FINAL still equals the source. Formal verification (formal_specs/lean/Replication/Upgrade.lean): - Generalise the engine to an arbitrary version scheme v : Nat -> Nat (live version at ordinal i; relocation tombstone v i - 1) and prove replicate_convergesV: convergence holds for ANY gap-monotone scheme. So the absolute _version numbers a given connector emits are irrelevant to correctness - only their order matters. - upgrade_safe: replicating a stream whose first n ordinals use the OLD version scheme and the rest use the NEW scheme converges, provided the combined scheme stays gap-monotone (the new version continues strictly above the last version the old one wrote). That is the drop-in-replacement guarantee. - liveVersion_gapMono: the shipped ordinal scheme 2*i is gap-monotone. All three depend only on [propext, Quot.sound] (no sorryAx), verified via #print axioms. Compatibility analysis backing the theorem hypothesis (2.8.0/2.9.1/2.10.3 vs 2.11.0), recorded in specs/02-versioning/06-upgrade-compatibility.md: - _version formula precedence and arithmetic preserved (gtid > sequenceNumber > lsn > SnowFlakeId(ts,offset)); 2.11.0 only ADDS branches. The no-GTID sequence anchor moved from envelope ts to source ts (#1346), which cannot invert a post-upgrade row against a pre-upgrade one given the restart gap. - Debezium is 3.1.3.Final in all four versions, so the offset-store and schema-history serialisation is byte-compatible; 2.11.0 reads the committed position written by 2.8.0/2.9.1/2.10.x. - No config key removed or renamed; hardcoded replacingmergetree.delete.column default unchanged ("sign") - existing deployments keep their own config. Spec/governance: add Invariant I11 (Drop-in Upgrade Safety) to CONSTITUTION.md, add spec 02.06, wire Upgrade.lean into the lakefile roots and the validate_specs.py Lean module + no-sorry check, and document the theorems in the Lean README. Destructive-Op-Check: sites=3; result=pass
…and of DBMetadata's pooled reconnect (Spec 06.09 section 2) PrimaryKeyRebuild and PrimaryKeyBackfill each carried their own copy of the same plain-Statement exec/scalar/column/columns helpers (including an identical system.columns query), differing only in the exception they raise and the log prefix. They now share RebuildQueries, parameterised by the caller's failure; logging, failure types and messages are unchanged, and the backfill resume scan keeps propagating its raw exception. DBMetadata repeated the reconnect-if-pooled block ten times; it is now reconnectIfPooled(conn, database), same condition and same call. Behaviour-preserving: PrimaryKeyRebuildTest (29), MySqlDDLParserListenerImplTest (162) and the DBMetadata unit tests (19) pass unchanged.
…spec-exempt: no behaviour change; the declared config variables become the single source of the two prefix strings ColumnTypeOverrideConfig restated] ClickHouseSinkConnectorConfigVariables.COLUMN_TYPE_OVERRIDE_DIRECT_PREFIX and COLUMN_TYPE_OVERRIDE_ALIAS_PREFIX were declared but never read, while ColumnTypeOverrideConfig hardcoded the same literals. The config class now takes its prefixes from the variables, so the two cannot drift.
…he exhausted-budget test stops relying on the swallowed interrupt (Spec 06.08 section 3.2) DdlFailureLoudTest.ddlExecutionFailureAfterRetriesExhaustedIsLoud interrupted its thread to skip the 10 s back-off, which only worked because the retry loop swallowed the interrupt. With the interrupt now honoured it halted with the interrupt instead of the exhausted budget. The back-off becomes DebeziumChangeEventCapture.ddlRetryBackoffMs (10 s, package-private like stopDrainTimeoutMs) and the test sets it to 0; its assertions (Max retries, the ClickHouse error as cause) are unchanged.
Collaborator
Author
|
Code review of this PR (gate-based: contract, impacted surface, failure and divergence, evidence, then quality), with fixes pushed as commits b326b6e..d9750c6 on this branch. Summary: the write path, offset handoff, loud-failure ordering and the MySQL primary-key rebuild hold up under tracing. The review found defects in newer surface (REST flush/resume, history-mode key predicates, PostgreSQL DDL, one reconciler query) and one structural problem in the python toolset. Every fixed item below has a regression test that fails when the fix is reverted (mutation-checked); the full Java unit suite shows no new failure versus the PR head (Docker-only ITs excluded locally, CI runs them). FIXED in this branch
OPEN, being fixed next on this branch OPEN, needs a maintainer decision Blind spots: the preflight classes, payload/* and the Docker-based ITs were not reviewed line by line or run locally. Verdict: request changes until items 8 and 9 land. |
…eEventCapture into DdlTableNames (Spec 08.02 section 3.4) The rename patterns and the tableChanges/source.table/rename-text resolution were ~150 lines of stateless string handling inside the 4,266-line engine class. They move unchanged to the package-private DdlTableNames (affected, renamed, fromRecord, bareName); the three call sites use DdlTableNames.affected and the existing rename tests call DdlTableNames.renamed with their assertions unchanged. Behaviour-preserving: DebeziumChangeEventCaptureTest, DdlFailureLoudTest, DdlFailureModesTest, DdlIgnoreRulesTest and EngineRestartFifoResetTest (53) pass.
…RN, not DEBUG (Spec 10.04 section 3.9) ClickHouseConverter.extractDebeziumSchema caught every exception, logged its message at DEBUG and returned null; PostgresSchemaChangeDetector then skips the record, so schema-drift detection could stop for a table with nothing visible in a production log. The failure is now logged at WARN with the topic and the stack trace; the null return and the caller are unchanged. Test: ClickHouseConverterSchemaExtractionTest; mutation-checked: fails on the DEBUG version. PostgresSchemaChangeDetectorTest (16) unchanged.
…ssword; neither loader prints it on failure (Spec 11.05 FM-11.05-1, FM-11.05-2) FM-11.05-1: REDACTED-SYSTEM/db_load/clickhouse_loader.py (the pyproject entry point and the ch-mysql-resync default) spliced the password as --password with single quotes into a command run by bash, so a quote in the password ended the word and the rest ran as shell syntax, and execute_load logged the whole command, password included. Both load paths now build the argument with shell_password_arg / shell_config_file_arg (shlex.quote, the password registered for redaction), every log line and exception carries redact_password(cmd), and load_data_mysqlshell uses the resolved password instead of the raw CLI value, as the legacy copy does. FM-11.05-2: both loaders raised AssertionError with the raw command and logged it unredacted at DEBUG in run_command/run_quick_command; both now redact. Tests: the two DEFECT-skipped tests in test_loader_failure_modes.py are live and pass, plus test_failure_message_is_redacted and test_password_with_shell_metacharacters_stays_one_word for the packaged copy; mutation-checked: all four fail on the previous loaders. Full python unit run: 770 passed, 2 skipped; the one failure (test_checksum_connector_fence TestSnapshotSideProtocol.test_no_pipe_is_left_open, order-dependent) fails identically on the PR head and passes alone.
Collaborator
Author
|
Review follow-up: three more commits on this branch (head 1137a45).
Still open: merging the two python trees into one implementation, plus the CI step that runs every python unit-test file (items 8 and 9). Items 11, 12 and 14 still need a maintainer decision. |
…the code, the legacy db/db_compare/db_load/db_dump modules are shims (Spec 13.01 section 3.16, I-13.01-1; Spec 11.05 FM-11.05-3) The MySQL tools existed twice and had drifted: fixes landed in the legacy copy only (datetime clamp and saturation, the source column-set / replica-only check, exact is_binary_datatype, quote_plus of MySQL credentials) while run-from-any-directory launching and mysql_resync existed only in the packaged copy that pyproject installs. Each packaged module is now the union of both copies, checksum_common moves into the packaged db package and owns the single DateTime64/Date32 range constants, and the 13 hand-written legacy modules become shims that alias the packaged module object (sys.modules), so imports, patches and module globals are shared, and that run the tool's main() when started as a script, with no PYTHONPATH or cwd requirement. CI: python-toolset-unit-tests.yml runs the whole offline suite (db_compare, db_load, db_dump, tests) on every PR and push to 2.11.0; before, one test file ran. Tests adapted only where they encoded the old side launcher (python db_compare/x.py from PATH instead of interpreter -m module) or the old packaged arities; assertions unchanged. test_no_pipe_is_left_open now collects earlier garbage before recording, so it measures only the code under test (it failed in a full run on the PR head). New guard: tests/test_single_source.py (identity, no logic in shims, launchers run from / without PYTHONPATH). The two checksum Dockerfiles copy the packaged tree. Verification: python -m pytest -q db_compare/tests db_load/tests db_dump/tests tests: 805 passed, 2 skipped (PR head: 766 passed, 4 skipped, 1 failed); scripts/validate_specs.py passed. The flagged template-like files are a GitHub Actions workflow (parsed with a duplicate-key-detecting YAML loader) and Python modules (f-string braces; imported and exercised by the 805 tests); none is Jinja. Jinja-Render-Check: rendered=7; formats=yaml,text; result=pass
Collaborator
Author
|
Review follow-up: items 8 and 9 are done in db0330a (head).
Verdict update: no blocker left from this review. Items 11, 12 and 14 remain open for a maintainer decision. |
…-only column scope (follow-up to #1437) (#1559) * fix(sink): a failed ALIAS override reconciliation now halts like a direct-override mismatch (Spec 08.05 s3.3.2) ColumnTypeOverrideReconciler#reconcileAliasOverrides auto-applies a missing or drifted column_type_override.alias.* entry via ALTER TABLE ... ADD/MODIFY COLUMN ... ALIAS. Three ways this could fail left the table silently out of sync with the declared override, unlike a direct override, which already halts via ColumnTypeOverrideMismatchException: - a rejected ADD/MODIFY COLUMN ... ALIAS statement was only logged, and the drift was never retried; - an unreadable system.columns silently skipped both the direct and the ALIAS checks for the whole table; - the alias column's configured name already belonging to a non-ALIAS column produced no DDL and no log line at all. All three now throw ColumnTypeOverrideMismatchException (reused, not a new exception type; a new (String, Throwable) constructor preserves the underlying SQLException as the cause). It carries no ClickHouse error code and is not a terminal type for ClickHouseErrorClassifier, so it classifies UNKNOWN and ClickHouseBatchRunnable retries the same batch with exponential backoff -- the same mechanism a direct-override mismatch already relies on, now applied symmetrically, satisfying Invariant I15 (bounded, declared recovery) and I9 (no silent continuation past a condition that leaves ClickHouse different from the source). Per a scope ruling from the coordinator: parity is required only for columns that match the source, not for columns a user added to ClickHouse by hand. Both override kinds (this fix touches only the ALIAS path; the direct path already had the equivalent check) are skipped silently -- no DDL, no halt -- when the override's source column is absent from the table being reconciled. This keeps a wildcard override that legitimately does not apply to a given table from halting it, and guarantees a ClickHouse-only column is never inspected or altered, even one that happens to carry the exact name a generated ALIAS column would have. reconcile()/getExistingColumns()/reconcileAliasOverrides() no longer declare a checked exception (every failure path in this class now throws the unchecked ColumnTypeOverrideMismatchException), so ClickHouseAutoCreateTable createNewTable()'s try/catch around the reconcile() call -- which rethrew the mismatch exception but silently swallowed anything else via catch (Exception e) { log.error(...) } -- is removed; there is nothing left to catch, and removing it also closes that latent catch-all for any other exception the reconciler could throw. Spec: 08-schema-catalog/05-auto-table-creation.md section 3.3 gains a cross-reference, new section 3.3.2 documents the halt and the scope guard, and new FM-08.05-5 (Failure Modes & Recovery) is added and marked FIXED. Verification Criteria lists the new/changed tests. Tests (ColumnTypeOverrideReconcilerTest, all new except the first): - addColumnDdlFailureHalts(), modifyColumnDdlFailureHalts(), nonAliasColumnCollisionHalts(), existingColumnsReadFailureHalts() pin the four halt triggers. - wildcardOverrideForAbsentSourceColumnIsIgnored(), clickHouseOnlyAliasColumnForAbsentSourceColumnIsNeverTouched() pin the source-matching scope guard from the coordinator's ruling. - addColumnDdlSucceedsForSourceMatchingColumn() confirms the existing auto-add behaviour for a source-matching column is unaffected. No existing test encoded the old log-and-continue behaviour, so none needed to be rewritten; columnMetadataQueryBindsTheNames() (section 3.3.1, pre-existing) is unchanged and still passes. Mutation-checked: reverted the three main-source files (the test file and spec left in place) and re-ran the suite -- the build failed to compile (three of the new test methods call reconcile() directly, and the pre-fix signature still declares "throws Exception"), confirming the new tests cannot pass without the fix. Restored the fix; recompiled clean. Verification (boxcar sandbox, docker.io/jenkins/jenkins:lts-jdk17, mvnw -o): mvn test -Dtest=ColumnTypeOverrideReconcilerTest Tests run: 8, Failures: 0, Errors: 0, Skipped: 0 mvn test -Dtest=ClickHouseAutoCreateTableTest,ColumnTypeOverrideReconcilerTest Tests run: 9, Failures: 0, Errors: 0, Skipped: 0 -- in ClickHouseAutoCreateTableTest Tests run: 8, Failures: 0, Errors: 0, Skipped: 0 -- in ColumnTypeOverrideReconcilerTest python3 scripts/validate_specs.py -- Validation PASSED (83 specifications, 1215 verification citations checked). Destructive-Op-Check: sites=1; result=pass Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * fix(postgres): halt on schema-drift failure instead of logging and continuing (spec 10.04 s3.9) Defect: a PostgreSQL source column that could not be confirmed present on ClickHouse was a silent divergence, not a loud failure. Three places caught and logged past exactly this condition instead of halting, breaking parity with the MySQL path (whose SQL thread stops and does not advance its position on an error): - `ClickHouseConverter.extractDebeziumSchema` caught any exception while reading a record's own schema and returned `null` after a WARN, so the caller treated "I could not tell what columns this row has" the same as a legitimate no-row record. - `PostgresSchemaChangeDetector.checkAndReconcile` wrapped its whole body in a catch-all that logged and returned for a null writer/connection, a failed `system.columns` fetch, and a failed `addMissingColumns` call. Worse, because `lastReconcileAttempt` was recorded unconditionally before the reconciliation call, a failure was also masked for the following 10s cooldown window: every record in that window passed through unchecked. - `PostgresSchemaReconciler.addMissingColumns` logged each failed `ALTER TABLE ... ADD COLUMN` and returned normally, so the caller believed every missing column had been added. - The call site in `DebeziumChangeEventCapture#processEveryChangeRecord` caught whatever the detector threw and logged "Schema drift detection threw unexpectedly; continuing replication". Fix: let all of the above propagate. `addMissingColumns` still attempts every missing column (one bad column must not block the others) but now throws afterward naming every column that failed, with the first rejection as cause. `checkAndReconcile` no longer catches; it still returns normally only for the two legitimate non-failure outcomes (a record extraction that correctly returns null, and a table confirmed absent from ClickHouse, which the connector is expected to auto-create). It now also tracks the last reconciliation failure per table so a cooldown-window check re-throws the stored failure instead of passing records through silently. The `processEveryChangeRecord` call site wraps whatever `checkAndReconcile` throws in `RecordReplicationException` -- the same exception type and wrap-and-rethrow shape already used a few lines below for a parser failure -- so it escapes the generic catch-all and halts the engine the same way an unconvertible row does: offset not committed, retry/terminal-exit policy applies on restart. Parity scope (ruling applied here and stated in spec 10.04 s3.9): parity is required only for columns that exist in the source and for tables the connector manages. A column a user has added to a ClickHouse table that does not exist in the source is never flagged, never reconciled, and never halts the pipeline -- `findMissingColumns` inspects only the Debezium-side field list, by construction. Spec: specs/10-resilience-monitoring/04-loud-failure-guarantee.md section 3.9 rewritten (previously described the WARN-and-skip behaviour as correct); FM-10.04-8 failure mode entry added; Verification Criteria updated to the five tests below. Tests (all new or changed to assert the halt instead of the skip): - `ClickHouseConverterSchemaExtractionTest.unreadableSchemaPropagates()` (rewritten from `unreadableSchemaIsLoggedAtWarn`, which asserted `assertDoesNotThrow`); `nullValueSchemaReturnsNullWithoutThrowing()` pins the still-legitimate null case. - `PostgresSchemaChangeDetectorTest`: `genuineRowWithUnreachableWriterThrows()` and `cdcInternalColumnsEnvelopeStillThrows()` (rewritten from two tests that asserted `assertDoesNotThrow` for the same inputs); `cooldownAfterAFailureStillThrows()` (new, pins the cooldown-masking fix via reflection-prepopulated cache state); `extraClickHouseOnlyColumnNeverHalts()` (new, pins the parity-scope ruling: a ClickHouse-only column never halts even with no writer wired). - `PostgresSchemaReconcilerAddColumnFailureTest.failedAlterThrowsAfterAttemptingAllColumns()` (new): a rejecting JDBC connection proxy confirms all 3 missing columns are attempted and the thrown exception names all 3. - `SchemaDriftFailureIsTerminalTest.schemaDriftFailurePropagatesThroughProcessEveryChangeRecord()` (new): exercises the real `processEveryChangeRecord` seam with a rejecting-connection-backed detector wired through `PostgresConnectorConfig`; confirms `RecordReplicationException` naming "schema-drift" escapes. Mutation-check: each of the four production edits (ClickHouseConverter.java, PostgresSchemaChangeDetector.java, PostgresSchemaReconciler.java, DebeziumChangeEventCapture.java) was individually reverted in isolation (git stash of that one file) and the corresponding test(s) re-run in an isolated container sandbox; every one went red with the expected failure/assertion message, then the file was restored and the suite re-confirmed green. Verification (isolated container sandbox, offline `-o` Maven reactor build `-pl sink-connector,sink-connector-lightweight -am`): Tests run: 22, Failures: 0, Errors: 0, Skipped: 0 (ClickHouseConverterSchemaExtractionTest: 2; PostgresSchemaChangeDetectorTest: 18; PostgresSchemaReconcilerAddColumnFailureTest: 1; SchemaDriftFailureIsTerminalTest: 1) BUILD SUCCESS. `python3 scripts/validate_specs.py`: Validation PASSED (83 specifications, 1217 verification citations checked). Not touched: PreparedStatementExecutor, ColumnTypeOverrideReconciler, and the alias-override code in PostgresSchemaReconciler (ensureAliasColumns/ensureAllAliasColumns/fetchColumnNames) -- owned by a different work item. Docker-dependent integration tests were not run (no Docker in the sandbox); this change only touches unit-testable code paths. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * fix(batch): a batch retry resends only what ClickHouse has not yet acknowledged, not the whole record list (Spec 03.06 section 3.5) A chunked batch's failure was retried from scratch: when chunk k of a BatchChunker split threw, the worker retried the whole batch (spec 03.03 FM-03.03-2) with the same retained record list, and nothing distinguished a record whose write ClickHouse had already acknowledged from one it had not. Every retry therefore resent chunks 1..k-1 again, resent a flushStagedRows() flush that had already landed, re-ran an inline SCD2 UPDATE/DELETE history statement that had already run, and re-ran a TRUNCATE/bulk-close that had already applied. Destructive-Op-Check: sites=4; result=pass * fix(python): tolerate ClickHouse-only columns in the checksum tool instead of warning and failing (Spec CONSTITUTION I6 parity scope, Spec 11.02 sections 3.3, 3.9, FM-11.02-9, FM-11.02-10) A later ruling narrows what the replication contract covers: parity is required only for columns that exist on the MySQL source (or that the connector itself creates and manages for replication bookkeeping), not for every column a ClickHouse table happens to carry. A column added on the ClickHouse replica alone, with no column of that name on the source table, has no source value to conform to or diverge from, so it is tolerated. This reverses the previous default, which named such a column at WARNING, failed the scheduled job's "no WARNING in the output" contract, and exited 1 unless --allow_replica_only_columns was passed. - specs/CONSTITUTION.md: new "Parity scope" paragraph under Invariant I6, the single place this rule is stated; it does not relax Invariant I9 (an unapplied or lost source change is still a loud halt) because a ClickHouse-only column has no source value to apply or lose. - clickhouse_table_checksum.py restrict_to_source_columns(): the replica-only branch now logs once per table at INFO ("... tolerated, not compared)") instead of WARNING. The opposite case -- a source column the replica lacks -- is unchanged: still a WARNING, still a real difference, since parity is in scope for it. - top_level_table_checksum.py: record_replica_only() always logs the tolerated finding at INFO (no more flag-gated WARNING branch); relay_side_messages() now checks record_replica_only() ahead of, and independently of, the generic WARNING-marker branch, so the report is recognised regardless of which level the side emits it at; report_run_summary()'s end-of-run line is INFO and no longer sets exit code 1 for these findings. --allow_replica_only_columns stays an accepted flag (Invariant I11: no flag is removed) but is now a no-op; its --help text says so. Tests: renamed/rewrote the three tests that pinned the old WARNING/exit-1 default to pin the new tolerated/INFO/exit-0 default instead -- TestReplicaOnlyColumns (test_checksum_fidelity.py), TestSourceColumnSetReachesTheReplicaSide (test_checksum_verdicts.py), TestScheduledJobLogContract (test_checksum_job_log_contract.py). The "source column missing on the replica" tests in the same classes, which exercise the still-in-scope case, are unchanged. Mutation-checked: reverting just the two source files (git stash) turns 6 tests red -- test_replica_only_column_is_left_out_and_named_once_at_info, test_a_name_with_a_comma_or_a_space_survives_the_driver_to_side_trip, test_by_default_a_replica_only_column_is_tolerated_at_info, test_the_allow_flag_is_a_no_op, test_replica_only_column_is_tolerated_by_default, test_allow_replica_only_columns_flag_is_a_no_op; restoring the fix turns them green again. Verification: python3 -m pytest -q db_compare/tests db_load/tests db_dump/tests tests: 805 passed, 2 skipped, 30 subtests passed (unchanged from the pre-change baseline -- the renamed tests replace the old ones one-for-one, no existing test weakened or dropped). scripts/validate_specs.py: PASSED (83 specs, 1212 citations). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * refactor(cdc): extract version sequencer out of DebeziumChangeEventCapture into VersionSequencer (Spec 02.01 section 2, Spec 02.02 section 2, Spec 02.03 section 2) DebeziumChangeEventCapture.java had grown to ~4,100 lines even after commit 19c1bea extracted DdlTableNames. The `_version` sequencer -- the `effectiveTs * 1_000_000 + counter` formula, its process-wide floor, its intra-window counter, its source-log high-water mark, and the restart-floor seeding -- was ~300 lines of self-contained, stateless-logic-over-shared- statics with no dependency on the rest of the engine object. It moves unchanged to the package-private VersionSequencer, next to its user: - The 7 static fields (SEQUENCE_START, SEQUENCE_START_INITIAL, sequenceNumber, sequenceAnchorTs, sequenceHighWaterPosition, sequenceHighWaterEffectiveTs, sequenceMaxSourceTs) and their javadoc, verbatim. - nextSequenceNumber, the VersionAssignment inner class, nextVersionAssignment, seedVersionFloor, raiseVersionFloor and addVersion, verbatim. No logic changes, no renamed config keys, same log text, same exceptions. addVersion's visibility is narrowed from public to package-private: it has no caller outside the cdc package (confirmed by grep across both modules), matching the all-package-private convention of DdlTableNames. seedVersionFloorFromDurableMark and handleChangeEventBatch stay on DebeziumChangeEventCapture (they are entangled with the engine's own Properties/config lifecycle) and now call into VersionSequencer. The statics remain process-wide mutable state shared across engine restarts within one JVM -- this extraction only moves which class declares them, it does not change their lifetime or visibility to the rest of the process (still package-private to `cdc` via the class, but the fields themselves stay `public static` so the existing test hooks that reset them via direct field access keep working unchanged, just via a new qualifier). Formal model: formal_specs/lean/Replication/VersionFloor.lean names `DebeziumChangeEventCapture.nextSequenceNumber` / `DebeziumChangeEventCapture.seedVersionFloor` in two doc comments; both now read `VersionSequencer.*`. The Lean state machine itself is unchanged. Specs updated to point Codebase Mapping / qualified references at the new file (Primary Source and/or inline qualified mentions only -- specs that mention the moved methods/fields in unqualified prose are left as-is, per the precedent set by 19c1bea's own spec diff): - specs/02-versioning/01-version-formula-encoding.md - specs/02-versioning/02-commit-monotonicity-floor.md - specs/02-versioning/03-sequence-counter-lifecycle.md - specs/02-versioning/04-redelivery-stability.md - specs/02-versioning/06-upgrade-compatibility.md - specs/01-cdc-engine/04-transaction-boundaries.md - specs/01-cdc-engine/02-binlog-coordinates.md - specs/01-cdc-engine/08-compressed-transaction-payloads.md - specs/09-offset-management/03-clickhouse-offset-store.md Tests: the 17 test files that reset the statics or construct/invoke VersionAssignment / nextSequenceNumber / nextVersionAssignment / seedVersionFloor / raiseVersionFloor / addVersion via reflection or direct package-private access are updated to the DebeziumChangeEventCapture.X -> VersionSequencer.X qualifier; no assertion changed. No new regression test is added -- this is a pure move, and the existing suite already pins every behaviour of the sequencer; mutation check is therefore N/A in the usual revert-the-fix sense (there is no fix, only a move), and is instead a move check: `git diff` shows the moved bodies are byte-identical to the pre-move text modulo the class qualifier at call sites, and the test run below is unchanged in count from the pre-move baseline. Verification: - `python3 scripts/validate_specs.py` -> PASSED (83 specifications, 1212 verification citations). - `./mvnw -pl sink-connector-lightweight test`: before and after this change, Tests run: 784, Failures: 0, Errors: 76, Skipped: 28 -- identical counts; all 76 errors are the known Docker-environment baseline (testcontainers *IT classes cannot reach a Docker daemon in this environment). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * test(java): pin that a ClickHouse-only column is tolerated, not written (Spec 04.03 FM-04.03-4, Spec 11.02 FM-04.03-4 Blast radius) A later ruling narrows what the replication contract covers (Constitution I6, parity scope): a column that exists on the ClickHouse replica alone, with no column of that name on the MySQL source and not created or managed by the connector for replication bookkeeping, is tolerated rather than a replication gap. The Python checksum side of this ruling was already fixed in a prior commit on this branch; this commit covers the Java connector's own write path, per the follow-up audit: does the connector ever try to supply a value for such a column, warn about it, or ALTER it? Investigation (QueryFormatter.createColumns, PreparedStatementFieldMapper. insertPreparedStatement, GroupInsertQueryWithBatchRecords, DbWriter, DBMetadata.getColumnsDataTypesForTable) found the connector already does the right thing, by the same mechanism the #1389 pre-ALTER fix put in place: - A ClickHouse-only column can never be a member of any record's schema (no source column of that name ever produces one), so QueryFormatter.createColumns() never gives it a placeholder in the generated INSERT -- in every batch, not just the first one after an ADD COLUMN, because there is no ADD COLUMN on the source side to catch up to. - At bind time, PreparedStatementFieldMapper.insertPreparedStatement finds no entry for it in columnNameToIndexMap, finds it is not isUnboundByDesign, finds recordCarries(fields, colName) false, and takes the existing by-design branch: DEBUG log, continue. No value is bound, no exception is thrown, no ALTER is issued; ClickHouse keeps whatever value the column already holds. This is the #1389 mechanism's Case B, applied to a column that is absent *permanently* (no source counterpart will ever arrive) rather than *transiently* (pending an ALTER) -- mechanically identical, different rationale, so it gets its own named failure mode and its own test rather than extending the pre-ALTER one. No production code changes: this is a "pin existing-correct-behaviour" change (outcome (a) of the audit), not a bug fix. - specs/04-query-generation/03-field-membership-rules.md: added a paragraph to section 3.1 distinguishing Case B's transient (pre-ALTER) and permanent (replica-only) triggers; added FM-04.03-4 ("A column only ClickHouse has"); added the two new tests to section 5 Verification Criteria; updated the failure-mode count (3 -> 4). - specs/11-verification-tooling/02-db-compare-checksums.md: corrected FM-11.02-10's "Blast radius" sentence, which previously only pointed at spec 04.03 without saying what it found there, to state plainly that the connector's write path already excludes such a column from every generated INSERT, so nothing worse than an unjudgeable column happens. - sink-connector/src/test/java/.../db/batch/ClickHouseOnlyColumnTest.java (new): two tests through the two real entry points -- testClickHouseOnlyColumnIsOmittedFromGeneratedInsert (query-formatting, through GroupInsertQueryWithBatchRecords.updateQueryToRecordsMap) and testClickHouseOnlyColumnRecordWritesWithoutError (bind-time, through PreparedStatementFieldMapper.insertPreparedStatement, using a JDK dynamic proxy PreparedStatement stand-in -- this module has no mocking framework on its test classpath, following the pattern already used by NullValueColumnDropTest). The proxy records every touched index from any setXxx(int, ...) call rather than naming individual JDBC setter methods: a first version that matched only setNull/setString/setInt/setLong missed that an Int32 column binds via setObject (ClickHouseDataTypeMapper's isWiderIntegerTarget treats Int32 itself as a possible post-ALTER widening target, so it is not routed through setInt), which silently dropped the "id" column's own bind from the recorded indices and failed the test for a reason that had nothing to do with the behaviour under test. Mutation-checked: since no production code changed, the check instead confirms the two new tests actually exercise the mechanism being pinned. Temporarily disabled, in the two production methods the javadoc names, only the specific condition this change pins (QueryFormatter.createColumns's record-field-membership check at the "omitted from INSERT" branch; and PreparedStatementFieldMapper.insertPreparedStatement's `recordCarries` branch), confirmed both new tests red: ClickHouseOnlyColumnTest.testClickHouseOnlyColumnIsOmittedFromGeneratedInsert: AssertionError -- ch_added_col got a placeholder in the generated INSERT. ClickHouseOnlyColumnTest.testClickHouseOnlyColumnRecordWritesWithoutError: StaleSchemaCacheException -- "Column ch_added_col is present in the ClickHouse table and carried by the record, but has no placeholder in the generated INSERT ... Failing the batch instead." Restored both methods; both tests green again. No other test's behaviour depends on this change, so no test was weakened, deleted, or had its assertions loosened. Verification (isolated container, JDK 17, offline Maven; Docker-backed testcontainers/*IT tests cannot run in this environment, the known baseline): ./mvnw -o -pl sink-connector -am -Dtest=ClickHouseOnlyColumnTest, NullValueColumnDropTest,QueryFormatterPreAlterRecordTest, PreparedStatementFieldMapperRecordCarriesTest, PreparedStatementFieldMapperUnboundColumnTest test Tests run: 23, Failures: 0, Errors: 0, Skipped: 0 (2 new + 21 pre-existing, unchanged). ./mvnw -o -pl sink-connector -am test (full module) Tests run: 770, Failures: 0, Errors: 20, Skipped: 41 -- all 20 errors are DBMetadataTest/DbKafkaOffsetWriterTest/DbWriterTest/ ClickHouseCreateDatabaseTest failing with "IllegalState: Previous attempts to find a Docker environment failed", the pre-existing testcontainers/ Docker-unavailable baseline for this environment, unrelated to this change. python3 scripts/validate_specs.py: Validation PASSED (83 specifications, 1214 verification citations). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * test(postgres): reference VersionAssignment at its new home VersionSequencer (fixes testCompile on the combined head) SchemaDriftFailureIsTerminalTest was written against DebeziumChangeEventCapture.VersionAssignment; the version-sequencer extraction (4f3f809) moved the nested class to VersionSequencer, and the two commits were combined without a compile in between. No behaviour change. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * fix(postgres): a PostgreSQL DDL translation failure halts instead of being swallowed (Spec 10.04 section 3.10, FM-10.04-9; precedent Spec 06.03 FM-06.03-1) The PostgreSQL DDL parse path (PostgreSQLDDLParserService#parseSql and all five enter*() callbacks in PostgreSQLDDLParserListenerImpl) wrapped translation in try/catch and logged-and-continued on any error, including statements the grammar could not parse. MySQL has no such catch: a syntax error propagates via ErrorListenerImpl#syntaxError (a RuntimeException), is wrapped in DDLReplicationException, and halts replication with retry/backoff instead of advancing the offset past an untranslated change (Spec 06.03 FM-06.03-1). Fix: reuse MySQL's own ErrorListenerImpl unchanged, registered on the PostgreSQL lexer and parser in a new PostgreSQLDDLParserService#runAntlrPipeline helper, and remove every try/catch from the five enter*() methods in PostgreSQLDDLParserListenerImpl (enterCreatestmt, enterAltertablestmt, enterRenamestmt, enterDropstmt, enterTruncatestmt). The bespoke LenientErrorListener that only logged is deleted; no second halting mechanism was invented, reusing MySQL's exact type, listener, and propagation path. Five things remain intentionally skipped without throwing (spec 10.04 section 3.10's allowed-skip list, each a dedicated early-return or reporting call decided before translation, never a caught exception): 1. Statement kinds with no listener callback (CREATE INDEX, CREATE SEQUENCE, CREATE VIEW, COMMENT ON, ...) parse fine and produce an empty translation plus a WARN log, mirroring MySQL's own FM-06.03-3. 2. DROP of a non-TABLE object (early return in enterDropstmt). 3. An unrecognized alter_table_cmd keyword (default branch in translateAlterTableCmd, DEBUG log). 4. An untranslatable constraint clause, reported via reportUntranslatedConstraint (Spec 06.09 section 3.7), unchanged by this fix. 5. ClickHouse-only tables/columns, tolerated by design under Parity scope (Spec 10.04 section 3.9). Spec: updated specs/10-resilience-monitoring/04-loud-failure-guarantee.md - Codebase Mapping cites the PostgreSQL parse path and its reuse of ErrorListenerImpl. - New section 3.10 'A PostgreSQL DDL translation failure halts, not merely warns': before/after behavior and the five-item allowed-skip list above. - Verification Criteria cites the two new tests plus the three existing allowed-skip-kind tests and PostgreSQLConstraintClauseReportTest. - New failure mode FM-10.04-9, following the sibling entries' Trigger/Behaviour/Detection/Blast radius/Recovery/RTO/Test structure. - Summary count updated to 9 failure modes. python3 scripts/validate_specs.py passes (83 specs, 1229 verification citations checked). Tests: added unparseablePostgresDdlPropagatesInsteadOfBeingSwallowed and unparseablePostgresDdlPropagatesThroughAtomicBooleanOverload to PostgreSQLDDLParserServiceTest; both feed an unparseable statement to the two parseSql overloads and assert a RuntimeException containing 'Error parsing DDL' escapes with nothing appended to the output buffer. Negative proof (reverted both source files to HEAD via git stash, reran the class in an isolated offline test environment, then restored the fix): pre-fix: Tests run: 38, Failures: 2, Errors: 0 -- both new tests failed: Expected RuntimeException to be thrown, but nothing was thrown. post-fix: both tests pass; log shows 'ERROR - Error parsing' for each unparseable statement; the other 36 tests in the class unaffected. Full build (build.sh, RUN_TESTS=1, isolated offline test environment, sink-connector installed -DskipTests then both modules tested) versus the e9068e8 baseline: sink-connector-lightweight: tests=790 (788+2 new) failures=0 errors=76 skipped=28 sink-connector: tests=780 failures=0 errors=20 skipped=41 Error counts match baseline; every failing/erroring testcase (verified via surefire XML) is a Testcontainers/Docker *IT class or container-startup failure, expected offline, not a regression. No existing test was weakened, skipped, or deleted. Destructive-Op-Check: sites=4; result=pass Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * refactor(cdc): extract system-database connection retry out of DebeziumChangeEventCapture into SystemDbConnectionRetry (Spec 01.01 section 3.5) DebeziumChangeEventCapture.java was still ~3,800 lines after commits 19c1bea (DdlTableNames) and 4f3f809 (VersionSequencer). The initial system/offset database connection retry -- SYSTEM_DB_CONNECT_ATTEMPTS, SYSTEM_DB_CONNECT_RETRY_MS, createSystemDbConnectionWithRetry and connectWithRetry -- was ~100 lines of self-contained, stateless retry logic with no dependency on the rest of the engine object beyond the credentials/config it is handed. It moves unchanged to the package-private SystemDbConnectionRetry, next to its caller: - SYSTEM_DB_CONNECT_ATTEMPTS and SYSTEM_DB_CONNECT_RETRY_MS, and their javadoc, verbatim. - createSystemDbConnectionWithRetry and connectWithRetry, verbatim, including log levels/messages. setSystemDbConnection on DebeziumChangeEventCapture now delegates to SystemDbConnectionRetry.createSystemDbConnectionWithRetry instead of calling the method on itself; no other behavior changes. SystemDbConnectRetryTest moves its reflection/reference target from DebeziumChangeEventCapture to SystemDbConnectionRetry (qualifier only; no assertion changed). Spec 01.01 (embedded engine lifecycle) gets a new Codebase Mapping bullet, a new section 3.5 describing the retry behavior, and the existing SystemDbConnectRetryTest citation is repointed at the new class. python3 scripts/validate_specs.py passes (83 specs, 1227 verification citations). Pure move: both the production code and the test were otherwise byte-identical before and after save for the DebeziumChangeEventCapture -> SystemDbConnectionRetry qualifier swap; no behavior change. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * refactor(ddl): extract the MySQL DEFAULT-literal translator out of MySqlDDLParserListenerImpl into DefaultLiteralTranslator (Spec 06.04 section 3.2.1, 3.2.2 rules 1-2) MySqlDDLParserListenerImpl.java was still 3,226 lines after commits 19c1bea (DdlTableNames) and 4f3f809 (VersionSequencer). The literal DEFAULT rewrite -- literalDefault, bytesLiteral, clickHouseString, stringBody, isCurrentTimestampDefault and firstEnumMember, plus the NUMERIC_CH_TYPE pattern they share -- is ~160 lines of stateless, side-effect-free parse-tree logic with no dependency on the listener's own state (config, database/table names, the DDL event timestamp, the logger). It moves unchanged to the package-private DefaultLiteralTranslator, next to its caller: - NUMERIC_CH_TYPE and its javadoc, verbatim. - literalDefault, bytesLiteral, clickHouseString, stringBody, isCurrentTimestampDefault and firstEnumMember, verbatim, including their javadoc. literalDefault, isCurrentTimestampDefault and firstEnumMember become package-private statics (resolveDefault calls them); bytesLiteral, clickHouseString and stringBody stay private, since only the moved methods call them. resolveDefault on MySqlDDLParserListenerImpl now delegates to DefaultLiteralTranslator.literalDefault/isCurrentTimestampDefault/ firstEnumMember instead of calling the methods on itself; no other behavior changes. eventInstantLiteral and backfillNotReproducible stay behind -- they need the listener's own ddlEventTimestampMs/ databaseName/cleanTableName/originalSql/log -- along with the DATETIME64_SCALE/UTC_SECONDS constants eventInstantLiteral uses. The no-longer-used java.math.BigInteger import is dropped from the old file; DefaultLiteralTranslator has no logger, since none of the moved code logs (matching the DdlTableNames precedent). None of literalDefault/bytesLiteral/clickHouseString/stringBody/ isCurrentTimestampDefault/firstEnumMember/NUMERIC_CH_TYPE is referenced by any test, directly or via reflection: every existing DEFAULT test in MySqlDDLParserListenerImplTest exercises this logic through the public translate(...) entry point, so no test file changes are needed. Spec 06.04 (ALTER TABLE clause translation) gets a new Codebase Mapping bullet naming DefaultLiteralTranslator and the methods/sections it implements, and resolveDefault is added to the Primary Source bullet's method list. python3 scripts/validate_specs.py passes (83 specs, 1227 verification citations). Pure move: the production code is byte-identical before and after save for the MySqlDDLParserListenerImpl -> DefaultLiteralTranslator qualifier swap at the three call sites in resolveDefault; no behavior change. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * refactor(ddl): extract the ReplacingMergeTree sorting-key/engine policy out of MySqlDDLParserListenerImpl into ReplacingMergeTreeKeyPolicy (Spec 06.05 section 3.6) MySqlDDLParserListenerImpl.enterColumnCreateTable carried the full sorting-key derivation for a CREATE TABLE -- declared PRIMARY KEY, then a NOT NULL UNIQUE key, then every stored column as a last resort, with the allow_nullable_key determination, the is_deleted/_is_deleted collision check, the ReplacingMergeTree/ReplicatedReplacingMergeTree engine-clause selection, and the ORDER BY index-prefix clean-up -- all inline in one already-large method. This is stateless, logic determining the table's identity and engine from the source key, not column/type emission, so it moves to its own package-private class, next to its caller: - resolveSortingKey: the PRIMARY KEY / UNIQUE-key-fallback / all-columns -fallback derivation (moved verbatim, including every rationale comment and the INFO/WARN/ERROR logging and the DDLReplicationException refusal for a table with no stored column), now taking orderByColumns/uniqueKeyColumns/orderedColumnNames/notNullColumnNames/ databaseName/tableName/overridePrimaryKey/originalSql as parameters and returning the nullableSortingKey flag. - resolveIsDeletedColumnName: the is_deleted/_is_deleted disambiguation loop, moved verbatim. - engineClause: the four-way ENGINE=... text selection, moved and restructured into one method returning the clause text (leading space preserved) for the caller to append. - sanitizeOrderBy: the id_registro(10)-style suffix clean-up regex, moved verbatim into a function returning the sanitized string. - splitIndexColumns: moved in full (not just visibility-widened), since grep confirms its only call site is inside the sorting-key derivation that is itself moving; its one internal call becomes MySqlDDLParserListenerImpl.stripBackticks(part). stripBackticks loses its `private` modifier (package-private now) so the new class can call it across the package boundary; no other change to it. The now-unused KeylessTableWarning import is dropped from the old file (the one call that used it moved to the new class). The schema-override primary_key/overridePrimaryKey lookup stays in enterColumnCreateTable -- it is passed into resolveSortingKey as a parameter -- along with the column/engine text emission, the PARTITION BY/TTL/SETTINGS emission and the appendOrderBy calls themselves, since all of those need the listener's own query/config state. parseCreateTable/parseColumnDefinitions are untouched: they are general column/type emission entangled with the listener's query buffer, not part of the key/engine policy. None of the moved methods is referenced by any test, directly or via reflection: every existing sorting-key/engine test in MySqlDDLParserListenerImplTest (including testCreateTableKeylessOrdersByAllColumns) exercises this logic through the public translate(...)/listener entry point, so no test file changes are needed. Spec 06.05 gets a new Codebase Mapping bullet naming ReplacingMergeTreeKeyPolicy and the methods/sections it implements. python3 scripts/validate_specs.py passes (83 specs, 1227 verification citations). Pure move: production code is byte-identical before and after for the MySqlDDLParserListenerImpl -> ReplacingMergeTreeKeyPolicy qualifier swap at the four call sites in enterColumnCreateTable; no behavior change. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * fix(batch): the all-applied short-circuit still advances the durable offset watermark (Spec 03.06 section 3.5) Section 3.5 says the watermark used for the offset commit reflects every row that is actually durable, whichever attempt wrote it, and the grouping path honours that by folding an already-applied record's offset into partitionToOffsetMap. The short-circuit in ClickHouseBatchRunnable.processRecordsByTopic (every record of the topic is already marked appliedToClickHouse) returned true before grouping, so it never reached the durablyInsertedOffsets merge. If the attempt that wrote those rows returned false before its own merge, the rows were durable in ClickHouse but outside the committed offset, and a restart would redeliver and re-insert them. Fix: the short-circuit merges each record's offset into durablyInsertedOffsets (max per TopicPartition, the same merge the normal path uses) before returning true. ClickHouseBatchWriter has no durable watermark, so its short-circuit is unchanged. Spec: 03.06 section 3.5 states the fold for the short-circuit; Verification Criteria cites the new tests. Tests: AllAppliedTopicWatermarkTest (no server; connections are null and the short-circuit never uses one). - anAllAppliedTopicAdvancesTheDurableWatermark: without the fix it fails with "partition 0 must advance from 5 to the highest durable offset, 9 ==> expected: <9> but was: <5>"; with the fix it passes. - anAllAppliedTopicNeverMovesTheWatermarkBackwards: a lower offset never lowers the watermark. ClickHouseBatchRunnableTest, WorkerFailureModesTest, ParkedBatchWrittenOnceTest, PreparedStatementExecutorFailureModesTest and PreparedStatementExecutorTruncateTest pass unchanged (their skipped tests are the existing @disabled ones). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * fix(build): remove duplicate dependency declarations that Maven 3.10 rejects [spec-exempt: build metadata only; the resolved dependency set is unchanged] The CI build broke on the GitHub runner image update from ubuntu24/20260927.320 to ubuntu24/20261004.327, which moves Maven from 3.9.16 to 3.10.0. The first build step runs `mvn -q ... exec:exec` in sink-connector and writes its output into GITHUB_ENV; Maven 3.10 now reports duplicate dependency declarations as an ERROR that -q does not silence, so the step wrote "[ERROR] 'dependencies.dependency.(groupId:artifactId:type:classifier)' must be unique: org.apache.kafka:connect-json:jar -> duplicate declaration of version ${version.kafka} @ line 425" into GITHUB_ENV and failed with "Unable to process file command 'env' successfully". Every job after it was skipped. The same poms passed on the previous image (run on dcfb9e0: 15 of 15 green). Duplicates removed (a scan of every pom.xml finds no others): - sink-connector/pom.xml: org.apache.kafka:connect-json, declared twice, both compile scope. - sink-connector-lightweight/pom.xml: org.apache.kafka:connect-json (test, then compile), org.apache.commons:commons-lang3 3.12.0 (test, then compile), org.apache.logging.log4j:log4j-api (2.17.1, then 2.23.1 compile). In each pair the LAST declaration is kept, because that is what Maven 3.9 resolved: the jar built from the unchanged poms shades log4j-api 2.23.1, commons-lang3 3.12.0 and org/apache/kafka/connect/json, i.e. the compile-scope and later-version entries. Keeping them leaves the effective dependency set exactly as it was. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * fix(build): declare maven-compiler-plugin once in the lightweight pom (Maven 3.10 rejects duplicate plugins) [spec-exempt: build metadata only; the plugin configuration the build uses is unchanged] After the duplicate dependencies were removed, the same runner-image update (Maven 3.9.16 -> 3.10.0) failed the next step that reads sink-connector-lightweight/pom.xml (`mvn versions:set-property`): "'build.plugins.plugin.(groupId:artifactId)' must be unique but found duplicate declaration of plugin org.apache.maven.plugins:maven-compiler-plugin @ line 629". Maven 3.10 promotes duplicate plugin declarations from a warning to an error (release notes: "duplicated plugin entries will not be tolerated anymore"). The pom declared maven-compiler-plugin twice: 3.15.0 with lombok 1.18.34 as the annotation processor, then 3.11.0 with lombok 1.18.30. Maven 3.9 used the last one: the earlier CI log on this branch shows "maven-compiler-plugin:3.11.0:testCompile ... on project clickhouse-debezium-embedded". The 3.11.0 declaration is kept and the unused 3.15.0 one removed, so the compiler and annotation-processor versions the build runs with do not change. A scan of every pom.xml (dependencies, plugins, plugin executions, modules) now finds no duplicate declarations. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * test(e2e): run every test the pipeline's testflows suites load - no Skip, no XFail [spec-exempt: test-suite selection only, no product code] The connector must replicate with source-to-destination parity, so the end-to-end suite has to pass in every test, not only in the ones it chooses to run. Until now the two suites the pull-request pipeline runs (regression_auto.py, regression_auto_replicated.py) disabled a large part of themselves: on the 2.11.0 base and on this branch alike each run reported "91 features (74 ok, 14 skipped, 3 xfail)" (auto) and "89 features (72 ok, 14 skipped, 3 xfail)" (replicated). Disabled until now, now enabled: - ffails Skip entries (both regressions, 33 entries each): the whole insert feature, partitions, schema only, cli, truncate of tables without a primary key, the multiple parallel add/modify/drop column alter, the no-primary-key variants of insert/delete/update, primary keys/no primary key, partition limits. - inline xfail() in modules the pipeline loads: truncate.py complex primary key (2), insert.py DEFAULT-with-NULL (2), INSERT with a SELECT constant / SELECT from a table (2), one million datapoints, the no-primary-key examples of many-partitions-one-part / one-million / parallel (3), parallel_alters.py multiple parallel add and rename column. Several of these reasons describe parity defects (for example "doesn't work without primary key as only last row of insert is replicated"). This commit only stops hiding them; the failures it surfaces are fixed in the commits that follow on this branch. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * test(e2e): fix the previously skipped tests so they test what they claim [spec-exempt: end-to-end test harness only, no product code] With every Skip/XFail removed (previous commit) the pipeline's testflows suites failed the same set of tests in all four jobs. Each failure is in the test harness: the statement the test sends is rejected by MySQL itself, or the test races itself, before the connector is involved. Fixed without weakening any assertion: - insert/default with null, default with null and non null, use select constant as value, use select from table as value: the table's primary key `id` is INT NOT NULL with no default and the INSERT named only (col1,col2,col3), so MySQL answered "ERROR 1364 (HY000): Field 'id' doesn't have a default value" (quoted from the CI log). The INSERT now names id. - insert/parallel (all four table shapes): the three concurrent inserts used overlapping ids (1, 1..100, 2..1001), so MySQL rejected the later ones with duplicate keys and the step retried for its whole 300 s; keyless tables are a single column `x` but were sent two values per row. The inserts now use disjoint ranges (1, 2..101, 102..1101) and one value per row for keyless tables (distinct values, since identical whole rows collapse under FINAL in a keyless ReplacingMergeTree, spec 08.05). - truncate/complex primary key (both variants): create_mysql_table turned primary_key="id,k" into "id,k INT NOT NULL," - a column without a type, and k twice. It now defines only the key columns that the column list does not already define. - truncate/no primary key (both variants): four values were inserted into the three-column (k, c, pad) table; keyless tables now get three. - parallel alters/multiple parallel add modify drop column: ADD, MODIFY and DROP of the same column ran in parallel, so MODIFY/DROP could reach MySQL before ADD ("Unknown column"). They now run in order per column; columns still run in parallel with each other. - schema only: the test inserted a row into a table created after the connector was already streaming and expected it not to replicate, but snapshot.mode=no_data only skips the initial snapshot. In the replicated suite it also wrote schema_only.yml to env/auto/configs, which that suite does not mount, so the connector never restarted. The scenario now stops the connector, creates a table that already holds a row, starts the connector in no_data mode with offset and schema-history tables of its own (so a new snapshot really happens), checks the table is created with no rows, checks a row written afterwards is replicated, and restarts the connector with the suite's configuration, restoring the shared in-memory config. - cli (replicated suite) and multiple databases (replicated suite): cascades of the dead connector left by schema only. The client calls now assert exit code 0, so an unreachable connector fails at the call instead of as a later timeout. CI: a failing testflows job now prints every failing test's steps and messages (tfs transform nice-fails on the raw log); --output new-fails printed details only for the first failure and a bare "AssertionError" for the rest. Jinja-Render-Check: rendered=1; formats=yaml; result=pass Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * test(e2e): fix the three remaining harness defects in insert and schema only [spec-exempt: end-to-end test harness only, no product code] The previous commit fixed truncate, parallel alters, cli and multiple databases: they pass in all four testflows jobs of run 37722722453. Three harness defects remained, now visible thanks to the per-failure log step: - insert/default with null, default with null and non null, use select constant as value, use select from table as value: the rows now reach ClickHouse, but select() compared the test's comma-separated expectation (e.g. "\N,5,333") with `SELECT ... FINAL` in ClickHouse's default tab-separated format, so it could never match and timed out after 300 s. When a test supplies its expected output, the FINAL query now asks for FORMAT CSV, as the two other branches of select() already do. Comparisons against the MySQL client's own (tab-separated) output are unchanged. - insert/parallel, keyless shapes: create_mysql_to_clickhouse_replicated_table gives every shape an `id` column, so keyless tables are (id, x), not (x); the one-value rows from the previous commit were rejected by MySQL. All shapes are back to two values per row, still with disjoint id ranges. - schema only: errored at once with "'TestStep' object has no attribute 'deepcopy'" because the star imports bind a test step named `copy` over the standard-library module. deepcopy is now imported after them under its own name. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Collaborator
PR #1437: 2.11.0 (architecture spec + replication hardening)This PR bundles the 2.11.0 line. It changes 631 files across 139 commits, including about 27k lines of production Java in 95 files, plus specs, Lean proofs, Python tools and tests. The offset and handoff code is careful, and most of its edge cases are already covered by tests. I found one High issue in how a coalesced write reports its groups, plus a few smaller items. Recommendation: approve with fixes, and split future work of this size into separate PRs (offset FIFO, versioning, PK rebuild, Postgres DDL, Python tools). The review is against Reviewed in full (tier 1):
Read in part: Not reviewed:
Findings[High] An offset-ack failure in a coalesced write orphans the remaining groups and stalls offsets until the hard cap stops the engineWhere: Invariant / area: Offsets trail data, and failures don't silently skip. Problem: After a successful write,
Suppose the first of N groups triggers the drain and the ack throws. Groups 2..N were written to ClickHouse but never reported, so their units keep
This is not data loss. It is an unbounded replay, which leaves duplicate rows on engines that don't deduplicate by version, plus a misleading log trail. Fix: Register every written group before any acknowledgement can throw. For example, add one synchronized entry point and call it once from // DebeziumOffsetManagement
static synchronized void reportWritten(List<List<ClickHouseStruct>> groups) throws InterruptedException {
for (List<ClickHouseStruct> g : groups) {
markGroupWritten(g); // today's decrement / completedUnits.put, no drain
}
drainCompletedUnits(); // may throw; every group is already recorded
}If you'd rather not change the API, keep the loop but catch per group, remember the first exception, and rethrow it after the loop. Test: Extend [Low] Smaller items
Questions
Reviewed with the |
…ctor JARs, THIRD_PARTY_NOTICES.md and license texts (#1560) * build(license): Apache-2.0 compliance - no GPL/LGPL code in the connector JARs, THIRD_PARTY_NOTICES.md, license texts The 2.11.0 lightweight JAR shaded in code under licenses that may not be bundled into an Apache-2.0 work (ASF Category X) or that do not belong in a production JAR: - MySQL Connector/J 9.1.0 (GPL-2.0 + Universal FOSS Exception): 1,127 entries - MariaDB Connector/J 2.7.12 (LGPL-2.1): 283 entries - JUnit 4.10 (CPL-1.0, via json-simple's mis-scoped dependency): 242 entries The org.honton.chas license-maven-plugin 0.0.6 "compliance" goal ran in every build and never failed it. Java - mysql-connector-j is "provided": on the compile/test class path, not shaded. Debezium's MySQL connector links directly against com.mysql.cj.jdbc.Driver, com.mysql.cj.CharsetMapping and com.mysql.cj.protocol.a.NativeConstants, so no other driver can replace it; it is supplied at run time instead. The JAR manifest carries Class-Path: mysql-connector-j.jar lib/mysql-connector-j.jar, and mvn package copies the unmodified driver and its sources JAR to target/mysql-driver/. -Pbundle-mysql-driver restores the old shading for builds that are never redistributed. - MySqlJdbcDriver.check refuses a MySQL start without the driver and names the three ways to supply it; PostgreSQL is never affected. - ConnectionTimeZonePreflight calls Connector/J's own zone resolver by reflection: no project source links against GPL classes. - mariadb-java-client -> test scope (unused by src/main); junit excluded from json-simple; unused org.glassfish:javax.json removed. - maven-enforcer bannedDependencies replaces the no-op plugin: the build fails if Connector/J, the MariaDB driver or JUnit 4 reaches compile/runtime scope. Python (ch-sink-tools) - psycopg2-binary (LGPL-3.0) -> pg8000 (BSD-3-Clause, pinned >=1.30,<2). Dict rows from cursor.description; explicit cursor close; the 20 s timeout bounds only the connect (pg8000 would keep it on every socket read, so it is cleared after connecting); statement_timeout=0 still sent; the checksum snapshot uses SET SESSION CHARACTERISTICS ... REPEATABLE READ, READ ONLY before the implicit BEGIN and now verifies READ ONLY too. - mysql-connector-python (GPL-2.0) -> PyMySQL (MIT) in sink-connector/tests and generate_15k_tables.py. Attribution and packaging (per the 2.10.3 third-party license review) - THIRD_PARTY_NOTICES.md for both JARs, the Go client, the Docker bundles and ch-sink-tools, keeping the review's license elections. The Maven table is generated from mvn dependency:tree -Dscope=runtime. - licenses/: GPL-2.0 + UFE, MPL-2.0, EPL-2.0, CDDL-1.0, CDDL-1.1 texts. - Both JARs carry THIRD_PARTY_NOTICES.md and licenses/ under META-INF/. - Lightweight images ship the driver as a separate file next to /app.jar, plus /licenses/ with the notices, the license texts and the driver's complete corresponding source. - The release JAR zip includes the notices and licenses/; the notices are also attached as a release asset. - Kafka images no longer download io.confluent:kafka-schema-registry (the registry server, Confluent Community License): the Avro converter needs only kafka-avro-serializer and kafka-schema-registry-client. - sink-connector/deploy/libs/mysql-connector-java-8.0.27.jar (GPL) removed: the Kafka sink never connects to MySQL. - e2e podman scripts mount the driver next to /app.jar; doc/licensing.md documents the policy, the gate and how to supply the driver. Checks: publish.yml parses with a duplicate-key-detecting YAML loader; the edited shell scripts pass bash -n; postgres_dumper.py passes the Python unit suite (810 passed); README text only. Jinja-Render-Check: rendered=6; formats=yaml,shell,text; result=pass * spec(01.01, 07.03): declare the run-time MySQL JDBC driver contract; pin pg8000 to the verified 1.31 line Spec-first gate: the license change touched spec-governed sources (MySqlJdbcDriver, DebeziumChangeEventCapture, ConnectionTimeZonePreflight) without a spec update. - Spec 01.01: new section 3.2.1 (the driver is provided-scope, supplied at run time via the manifest Class-Path; MySqlJdbcDriver.check runs first and refuses a MySQL start without it; non-MySQL connectors untouched), Codebase Mapping entry, start-sequence step, FM-01.01-2 trigger and test, and the MySqlJdbcDriverTest verification citations. - Spec 07.03: the session-zone preflight resolves with the driver's own TimeUtil.getCanonicalTimeZone, now invoked by reflection through MySqlJdbcDriver.canonicalTimeZone. - pg8000 pinned to >=1.31,<1.32: the connect-timeout fix reads pg8000's private _usock, verified on 1.31.5; it still fails loudly (with a test) if the attribute is ever missing. * fix(python): pg8000 sessions honour the libpq environment (PGTZ, PGDATESTYLE, PGGEQO, PGSSLMODE) python-toolset-e2e-postgres failed on test_zone_less_timestamp_keeps_its_wall_clock_in_a_non_utc_zone (D-13.05-8): the dumper detected the server zone as UTC instead of America/Chicago and created DateTime64(6, 'UTC'). Root cause: psycopg2 is libpq-based and libpq sends PGTZ (and PGDATESTYLE, PGGEQO) from the environment as session parameters; pg8000 is pure Python and reads none of them, so the session kept the server default. get_postgres_connection now passes those variables as pg8000 startup parameters (TimeZone, DateStyle, geqo) and maps PGSSLMODE to pg8000's ssl_context (prefer/allow -> None, which is pg8000's TLS-if-offered default and equals libpq's default; disable -> False; require -> True; verify-ca / verify-full -> a verifying context using PGSSLROOTCERT). PGCLIENTENCODING is not mapped: pg8000 decodes UTF-8 only. Verified: - the PostgreSQL e2e suite against a real PostgreSQL 16 + ClickHouse 24.8 (external-stack mode): before the fix 1 failed, 30 passed, 2 xfailed (the failure above); after it 31 passed, 2 xfailed. - python unit suite (db_compare, db_dump, db_load, tests): 818 passed. - new unit tests for the environment and PGSSLMODE mapping.
Collaborator
Author
|
@subkanthi thanks for the review. #1566 addresses it. High: reportWritten records every group of a coalesced write before the single drain, and still rethrows the first failure; there is a regression test for your scenario and for a mark failure on the first group. Low 1: offset-acknowledgement failures get their own exception and log line. Low 2: topics split on the first two dots only, with a dotted-table test. Question: the completeness check is safe. Its retired side joins the key map like the copy does, and its rebuilt side is SELECT DISTINCT over T without FINAL, so a tombstoned key counts as present and the backfill cannot loop on it. The code review skills and CI checks are in #1565. |
…c 11.06) (#1565) Adds the review methodology and a CI gate so contributors and agents review PRs the same way: - .claude/skills/: pr-self-review (five review gates: contract, impacted surface, failure and divergence, evidence, lower-priority quality) with a project-specific checklist; code-quality-{jvm,python,go,bash,sql,architecture}; code-reuse-review, code-reuse-first, design-note-debt, mandatory-code-change-validation, destructive-operation-safety, repro-harness-negative-proof and sink-connector-mysql-source-of-truth. Index in .claude/skills/README.md, pointer in AGENTS.md. - scripts/review_gates.py: stdlib-only checks over the lines a PR adds. destructive: every DROP/TRUNCATE/DELETE/DETACH/rm -rf site needs a DESTRUCTIVE: comment and a matching Destructive-Op-Check commit trailer. merge-stop: no code may stop or pause ClickHouse merges. license: no Category X dependency at a runtime scope, and every new runtime dependency is in THIRD_PARTY_NOTICES.md. hygiene: conflict markers, swallowed exceptions, debugger leftovers, credentials, YAML/JSON syntax. Exit 0/1/2. - scripts/tests/test_review_gates.py (25 offline tests), .github/workflows/review-gates.yml, doc/review_gates.md, specs/11-verification-tooling/06-review-gates.md. Jinja-Render-Check: rendered=1; formats=yaml; result=pass Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…y acknowledgement can throw; routing and error-category fixes (PR #1437 review; Spec 09.01 section 3.2, 03.07, 10.01) (#1566) High: processBatch reported the groups of a coalesced write with one checkIfBatchCanBeCommitted call per group. That call drains the FIFO, and the drain can throw (Debezium markBatchFinished timeout, FIFO consistency checks). A throw on group 1 left groups 2..N written to ClickHouse but never recorded, so their units kept remainingGroups above zero, hasUnwrittenBatches() stayed true, heartbeats stopped committing and the engine eventually stopped at handoff.wait.timeout.ms, replaying everything since the frozen offset. DebeziumOffsetManagement.reportWritten now marks every group first (a per-group mark failure is remembered, never stops the loop), drains once, and rethrows the first failure. Tests: CoalescedQueuedBatchesTest committerFailureOnFirstOfTwoCoalescedGroupsDoesNotOrphanTheSecond and groupThatFailsToMarkDoesNotOrphanSiblingGroups (fails without the fix: the sibling is never acknowledged). Low 1: a failure to report written rows is wrapped in OffsetAcknowledgementException and logged as an offset acknowledgement error in run(), instead of Retriable ClickHouse error (Code: -1, Category: UNKNOWN). Retry behaviour is unchanged. Low 2: RoutedBatch.extractTableName and createRoutingKey split the topic on the first two dots only, so a table named a.b keeps its own routing group instead of sharing one with table a. Test: RoutedBatchTest.testDottedTableNameIsNotSplitFurther. Question (PK rebuild completeness check): not a bug. The retired side joins the key map exactly as the copy does, so a row deleted at the source before the key map was loaded is excluded from both, and the rebuilt side is SELECT DISTINCT over T without FINAL or a delete filter, so a key present only as a tombstone counts as present (PrimaryKeyBackfill.completenessCheck). Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
com.google.protobuf:protobuf-java:3.19.6 (BSD-3-Clause) is a compile-scope dependency of io.debezium:debezium-connector-postgres 3.1.3.Final. sink-connector-lightweight/pom.xml neither excludes it nor filters it out of the shaded JAR, so it ships in the lightweight connector JAR. The 2.10.3 notice in #1561 lists it; the notice regenerated for 2.11.0 in #1560 does not. Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…es per commit (unblocks #1437) (#1567) * fix(review): the PR checks are not retroactive and attest destructive sites per commit (Spec 11.06 sections 3.1, 3.2) The release PR for the 2.11.0 line (#1437, base develop) failed the new Review gates check: its range holds 139 commits merged before the check existed, so it re-judged 6 old destructive lines and demanded one trailer covering all 35 sites in the line. - When the commit that introduced scripts/review_gates.py lies inside merge-base..head, checking starts at that commit, so nothing reviewed before the checks existed is re-judged, and every change from the introducing commit onward still is. - Attestation is per commit: each commit that adds sites carries its own Destructive-Op-Check trailer for its own count (several trailers in one message, as a squash merge collects them, are summed). A single cumulative trailer cannot work for a release branch whose range holds many already-attested commits. - Warning comments are still checked on the head version of every added line. Tests: 5 new (per-commit attestation, missing per-commit trailer, squash summing, pre-check changes ignored, changes from the introducing commit checked). The #1437 range (3c0759b..3fcf20d) now reports 0 findings; the same range without the introducing commit still reports the 6 unwarned sites. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * fix(review): squash merges may over-attest destructive sites; e2e test data is exempt (Spec 11.06 section 3.2) A squash merge concatenates the trailers of the PR's commits, each already checked exactly in that PR; when the PR reworked a destructive line their sum exceeds the squashed diff's net count. For a commit whose subject ends in (#N) the sum may exceed, never fall short of, its sites; every other commit still attests exactly. tests_e2e/ joins the test directories. 4 new tests (91 total, 4 skipped). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
… checks (Spec 11.06 section 2) (#1568) The workflow ran scripts/review_gates.py and its tests from the PR head. A branch cut before the checks existed has neither, so every such PR failed with ImportError: Start directory is not importable: scripts/tests (seen on #1561 and #1563). The workflow now takes the script and its self-tests from origin/<base>, runs them from the runner temp directory against the PR head, and reports a notice when the base has no script. It also means a PR cannot weaken the checks that judge it; a PR's own copy of the tests still runs in spec-governance. Emulated locally: the base copy's 34 self-tests pass from outside the repo, and the checks report 0 findings on the #1561 and #1563 heads. Jinja-Render-Check: rendered=1; formats=yaml; result=pass Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
* ADded claude skill for code review * Move the review skill to .claude/skills/clickhouse-sink-review/SKILL.md Claude Code discovers a project skill only at .claude/skills/<name>/SKILL.md; a SKILL.md directly under .claude/skills is never loaded. The directory name matches the skill's name field. No content change. This push also re-runs CI: the earlier build-kafka-lightweight failure was a transient Docker Hub error (auth.docker.io returned 500 while pulling debezium/connect-base:1.9.5.Final), not a problem in this PR. * ci: re-run checks after #1568 The Review gates job failed on this branch because the workflow ran the check script from the PR head, which predates it; #1568 makes it use the base branch copy. No content change. --------- Co-authored-by: minguyen9988 <136291932+minguyen9988@users.noreply.github.com>
… the PR head (Spec 11.01 section 3.7) (#1569) The job checked out pull_request.head.sha. A branch cut before formal_specs/ existed has no lean-toolchain, so lake build failed with no default toolchain configured, and the job validated a tree that is not what merges (#1561 is such a branch). It now checks out github.sha, the synthetic merge commit of the PR head into the current base, i.e. the tree that would land. The [spec-exempt: ...] marker keeps working: validate_specs.py already searches every commit message in BASE..HEAD, and the merge commit's second parent is the contributor's head. Push events are unchanged (github.sha is the pushed commit). Jinja-Render-Check: rendered=1; formats=yaml; result=pass Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
* Add THIRD_PARTY_NOTICES.md * THIRD_PARTY_NOTICES.md: align with the notice regenerated for 2.11.0 (#1560) This branch and #1560 both add THIRD_PARTY_NOTICES.md, so the PR cannot merge into 2.11.0 (add/add conflict). Take the 2.11.0 version as the base of this PR; the next commit adds the one runtime component this branch listed that the 2.11.0 notice lacks. * THIRD_PARTY_NOTICES.md: align with 2.11.0 again after #1564 #1564 added the protobuf-java row to the 2.11.0 notice, which re-created the add/add conflict with this branch. The file here is now identical to the 2.11.0 version, so the PR merges cleanly. * ci: re-run checks after #1568 The Review gates job failed on this branch because the workflow ran the check script from the PR head, which predates it; #1568 makes it use the base branch copy. No content change. * ci: re-run checks after #1569 Lean proofs and spec validator failed here because the job checked out this branch's head, which predates formal_specs/; #1569 makes it validate the merge commit with 2.11.0. No content change. --------- Co-authored-by: minguyen9988 <136291932+minguyen9988@users.noreply.github.com>
* fix: bound REPORTED_NULL_SCHEMA_MISMATCH_COLUMNS cache size * fix(test): make MAX_REPORTED_NULL_SCHEMA_MISMATCH_COLUMNS package-private NullSchemaMismatchBoundTest reads the cap from the same package, but it was declared private, so test compilation failed with 'has private access' and the build job stopped before any test ran. Drop the private modifier to match REPORTED_NULL_SCHEMA_MISMATCH_COLUMNS, which is already package-private for the same reason. Also fix a missing semicolon in the spec sentence describing the bound.
Collaborator
PR #1437 Review — 2.11.0
VerdictBugbot found no bugs in the covered files, and targeted manual verification of the highest-risk changed code paths found no correctness defects. Residual risk is coverage, not findings — see Coverage. Review Method
Manual Verification of Critical Paths
Notes (non-blocking)
Coverage
ConclusionNo bugs found in the reviewed changes. The critical replication paths (multi-row events, UPDATE handling, TRUNCATE ordering, offset commit serialization, version monotonicity) are correct at the PR head, with several long-standing silent-failure modes converted to loud ones. Merge risk from the reviewed scope is low; the dominant residual risk is the unreviewed long tail, itemised above. |
subkanthi
approved these changes
Oct 15, 2026
Collaborator
There was a problem hiding this comment.
Ran claude code skill to review, there are lot of changes that will take a while to review manually
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Architecture & End-to-End Flow Specification: ClickHouse Sink Connector (2.11.0)
Executive Summary
This document provides a comprehensive, ground-truth architectural specification and method-by-method operational mapping of the
clickhouse-sink-connectoron branch2.11.0. It covers both operating runtimes:sink-connector-lightweight): In-process Debezium engine streaming MySQL/PostgreSQL binlog/WAL directly to ClickHouse without Kafka brokers.sink-connector): Standard Kafka ConnectSinkTaskconsuming from Kafka partitions and writing batches to ClickHouse.The purpose of this specification is to establish a rigorous, idea-level mapping of all control paths, data pipelines, state machines, concurrency models, and failure domains. This mapping serves as the foundation for identifying structural incoherencies, race conditions, and design complexity traps that have led to recurring operational defects, and provides the architectural blueprint required for a coherent system redesign.
1. Prime Invariant & Architectural Topology
1.1 The Prime Directive
The connector is a replication engine from transactional source databases (MySQL, PostgreSQL) to an analytical columnar store (ClickHouse).
MATERIALIZEDcolumn shadowing source values), the connector automatically alters the ClickHouse schema to enable ingestion.1.2 Architectural Dualism
2. Core Component Directory & Responsibilities
ClickHouseDebeziumEmbeddedApplicationlightweightAppInjector); coordinates lifecycle.DebeziumChangeEventCapturelightweightClickHouseSinkTasksink-connectorSinkRecordbatches, invokes deduplication, tracks durable offsets inpreCommit().ClickHouseBatchExecutorsink-connectorScheduledThreadPoolExecutor; introduces thread-safe gate monitor withpause(),resume(), andawaitQuiescent()for DDL barrier synchronization.ClickHouseBatchRunnablesink-connectorDbWriter, formats queries, and executes JDBC flushes.ClickHouseBatchWritersink-connectorthreadPoolSize == 1); processes batches inline without thread dispatch.DbWritersink-connectorGroupInsertQueryWithBatchRecordssink-connectorrefreshIfRecordHasUnknownColumn); enforces column writability; handles UPDATE splitting.QueryFormattersink-connectorINSERT INTO table(cols) VALUES (?,?,...)queries; builds column-to-index parameter maps.PreparedStatementFieldMappersink-connector_version,_sign,is_deleted).PreparedStatementExecutorsink-connectorPreparedStatement; detects sorting key mutations on UPDATEs and synthesizes delete tombstones; triggersexecuteBatch().ClickHouseDataTypeMappersink-connectorCacheInvalidationManagersink-connectorDebeziumOffsetManagementsink-connectorRecordCommitterunderOFFSET_COMMIT_LOCK; checks pipeline quiescence.PostgresSchemaChangeDetectorlightweightsystem.columnsusing TTL-cached metadata.PostgresSchemaReconcilerlightweightALTER TABLE ... ADD COLUMNstatements on ClickHouse when PostgreSQL schema drift is detected.DDLParserFactory/ ServiceslightweightMySQLDDLParserService,PostgreSQLDDLParserService) that translate source DDL into ClickHouse DDL.BinLogHistorysink-connector_valid_from,_valid_to,_operation,is_deleted) inbinlog_historytables.3. End-to-End Execution Flow (Detail to the Tee)
3.1 Phase 1: Initialization & Engine Startup
ClickHouseDebeziumEmbeddedApplication.main()is invoked.config.properties, YAML files, or environment variables (EnvironmentConfigurationService).SinkConnectorLightWeightConfigandClickHouseSinkConnectorConfigvalidate settings (database overrides, credentials, thread pool size).KeylessTablePreflight.checkKeylessTables()queries source metadata (information_schema.tables,innodb_indexes) to identify tables lacking primary keys. Warns that keyless tables will collapse to a single row under ClickHouseReplacingMergeTreeunless generated invisible primary keys (GIPK) are enabled.DebeziumJdbcStorageOperations.createSchemaHistoryTable()creates the schema history table in ClickHouse.DebeziumJdbcStorageOperations.createViewForShowReplicaStatus()sets up the replica status monitoring view.replication.history.enable=true),ClickHouseAutoCreateTable.createHistoryDatabase()andcreateHistoryTable()ensure thebinlog_historydatabase and table exist.LinkedBlockingQueue[List[ClickHouseStruct]] recordsis instantiated with capacitymax.queue.size(default unbounded).threadPoolSize exceeds 1, aClickHouseBatchExecutoris initialized with scheduled worker threads runningClickHouseBatchRunnableat a fixed rate (buffer.flush.time).threadPoolSize == 1,singleThreadedWriter(ClickHouseBatchWriter) is initialized to execute synchronously on the CDC thread.DebeziumEngine.create(Connect.class)builds the capture engine.io.debezium.connector.mysql.MySqlConnectororio.debezium.connector.postgresql.PostgresConnector).DebeziumOffsetStorageorJdbcOffsetBackingStorestoring offsets in ClickHousereplica_source_infowithORDER BY offset_key).singleThreadDebeziumEventExecutor).3.2 Phase 2: Ingestion Loop & CDC Event Processing
Step 2.1: Monotonic Sequence Number & Version Calculation
The connector must assign each change event a strictly monotonic 64-bit
_versionfor ClickHouseReplacingMergeTree.DebeziumChangeEventCapture.nextSequenceNumber(long recordTs, SourcePosition position):recordTsis the source commit timestamp (source.ts_ms).SourcePositiontracks binary log coordinates:(filePrefix, fileSequence, position, row).lsn.position ranks above sequenceHighWaterPosition, the event is a first delivery (new commit).source.ts_msis the statement execution timestamp, not the commit timestamp. A long transaction committing after short transactions carries an oldersource.ts_ms. To maintain commit monotonicity:effectiveTsis clamped to the high-water floor:if (effectiveTs is less than sequenceMaxSourceTs) effectiveTs = sequenceMaxSourceTs.sequenceMaxSourceTsis updated toeffectiveTs.position ranks at or below sequenceHighWaterPosition, the event is an offset rewind / redelivery. The redelivery-stable assignment is preserved without advancing the high-water mark.diff = (effectiveTs - sequenceAnchorTs) / 1000.diff exceeds 1(more than 1 second advance):sequenceAnchorTs = effectiveTs,sequenceNumber = SEQUENCE_START (1,000,000,000).sequenceNumber++._version = effectiveTs * 1_000_000L + sequenceNumber.Step 2.2: DDL Interception & Pre-DDL Inversion Prevention
If an event carries a DDL schema change:
batchprior to the DDL record were captured against the pre-DDL schema.appendToRecords(batch, config)andbatch.clear(). This prevents schema inversion where pre-DDL records are executed against a post-DDL table.drainBeforeDDL()):records != null && !records.isEmpty(), sleeping 50ms per iteration up toDDL_DRAIN_TIMEOUT_MS(60,000ms). Workers continue executing.executor.pause(). The monitor lock insideClickHouseBatchExecutorsetsisPaused = true, causing workers arriving atbeforeExecute()to park.executor.awaitQuiescent(remainingTimeout). Spins untilactiveBatches.get() == 0.performDDLOperation()):checkIfDDLNeedsToBeIgnored).DDLParserFactory.getParser()selectsMySQLDDLParserServiceorPostgreSQLDDLParserService.clickHouseQuery).GENERATED ALWAYS AS ...) are mapped to ClickHouseDEFAULT (expr), NOTMATERIALIZED, ensuring the column can store source values while deriving default values when omitted.systemDbConnection.CacheInvalidationManager.getInstance().invalidateTable(tableKey)increments the table version counter.invalidateAll()incrementsglobalEpochto invalidate all tables.PostgresSchemaChangeDetector.invalidateCache(tableKey)clears the schema drift cache.DebeziumOffsetManagement.acknowledgeRecords().executor.resume()unlocks the gate and unparks worker threads.Step 2.3: PostgreSQL Schema Drift Detection
For PostgreSQL sources (which do not emit DDL statements over logical replication):
pgConfig.getSchemaChangeDetector().checkAndReconcile(sr, dmlTable, dmlDatabase)is called.SourceRecordKafka ConnectSchemafields against ClickHousesystem.columns(cached with TTL).PostgresSchemaReconciler.addMissingColumns()executesALTER TABLE db.table ADD COLUMN IF NOT EXISTS col typeon ClickHouse before row insertion proceeds.Step 2.4: Control Record & Heartbeat Offset Commit
snapshot=false) is emitted on a subsequent heartbeat record (which produces a nullClickHouseStruct).commitControlRecordOffset()checksisPipelineQuiescent():records.isEmpty(),routedRecords.isEmpty(), andDebeziumOffsetManagement.hasUnwrittenBatches() == false.DebeziumOffsetManagement.acknowledgeRecords()underOFFSET_COMMIT_LOCK. This commits the snapshot offset to ClickHousereplica_source_info, preventing snapshot loops on restart.3.3 Phase 3: Worker Batch Execution & ClickHouse Insertion
Step 3.1: Batch Dequeue & Tracking
ClickHouseBatchRunnable.runLegacyMode()poll batches fromrecords.DebeziumOffsetManagement.addToBatchTimestamps(batch)calculates the min and max timestamp of the batch and registers it ininFlightBatches.Step 3.2: Topic Partitioning & Destination Resolution
Map[String, List[ClickHouseStruct]] topicToRecordsMap.clickhouse.database.override.mapapplies destination remapping.clickhouse.common.database.prefixapplies database prefixing.clickhouse.database.schema.suffixapplies schema template formatting.Step 3.3: DbWriter Lifecycle & Schema Invalidation
getDbWriterForTable()checks the cachedDbWriterversion againstCacheInvalidationManager.getInstance().getVersion(tableKey).DbWriteris discarded.DbWriteris instantiated, querying ClickHousesystem.columnsto populatecolumnNameToDataTypeMap.TABLE_ENGINE.REPLACING_MERGE_TREE,REPLICATED_REPLACING_MERGE_TREE, orCOLLAPSING_MERGE_TREE).DBMetadata.getSortingKeyColumns().Step 3.4: Query Template Grouping & Column Enforcement
In
GroupInsertQueryWithBatchRecords.groupQueryWithRecords():refreshIfRecordHasUnknownColumn):cachedClickHouse columns.cached:CacheInvalidationManager.isColumnProvenAbsent(): if proven absent at current version, skips re-reading (prevents metadata query storms).system.columns.MATERIALIZEDcolumn):enforceSourceColumnIsWritable():default_kind. IfALIAS, ignored (query-time only).MATERIALIZED, ClickHouse derives it locally and rejects INSERTs. The connector enforces source authority by executing:ALTER TABLE db.table MODIFY COLUMN col (type) DEFAULT (expr)markColumnProvenAbsent(tableKey, col).QueryFormatter.createColumns):INSERTcolumn list so ClickHouse applies the columnDEFAULTinstead of bindingNULL.nullvalues are retained in theINSERTlist so an explicitNULLis bound, preventing ClickHouse from falling back to column defaults.c,r): Dispatchesafterimage toupdateQueryToRecordsMap.u):d): Dispatchesbeforeimage with delete flag.t): Directly generatesTRUNCATE TABLE \table``.Step 3.5: Statement Parameter Binding & Execution
In
PreparedStatementExecutor.insertBatch()andPreparedStatementFieldMapper.insertPreparedStatement():ORDER BYsorting key:updateRelocatesSortingKey()detects the mismatch betweenbeforeStructandafterStruct.insertTombstonePreparedStatement()binds the before-image values withis_deleted = 1and_version = record.getVersion() - 1.is_deleted = 0and_version = record.getVersion().ClickHouseDataTypeMapper.convert):PreparedStatementparameter positions:INT8(Byte),INT16(Short),INT32(Integer),INT64(Long). UsesNumber.intValue()to prevent class-cast failures on tinyints.Float32,Float64(converted toBigDecimal).BigDecimalmapped toDecimal(p, s).MicroTimestamp/Timestamp: Formatted with source/server timezone conversion.ZonedTimestamp: Converted using UTC/server zone mapping.Date/Date32: Converted from epoch days.MicroTime: Converted to microseconds since midnight (Int64).StringorJSON.io.debezium.data.Bitsto match big-endian ClickHouseBitrepresentations.WKBReaderto parse WKB polygons and binds to ClickHouse Geo types._version: Bound fromrecord.getVersion()(rejectUnderivableVersionthrows if-1).is_deleted: Bound to1for DELETE / tombstones,0for live rows._sign: Bound to-1for deletes,1for inserts (CollapsingMergeTree).ps.addBatch()accumulates records up tobuffer.max.records.ps.executeBatch()flushes to ClickHouse.ClickHouseErrorClassifier:FATAL(syntax errors, illegal column types): Batch is discarded and task stops.UNKNOWN/RETRIABLE(network timeouts, connection loss): Retried on next execution cycle.Step 3.6: Offset Acknowledgement & Commit Serialization
In
DebeziumOffsetManagement.checkIfBatchCanBeCommitted()andacknowledgeRecords():synchronized (OFFSET_COMMIT_LOCK).recordCommitter.markProcessed(record.getSourceRecord()).recordCommitter.markBatchFinished().completedBatchesto acknowledge any completed batches that were waiting on this batch.4. Method-by-Method Operation Mapping
The following catalog specifies the behavioral contract, synchronization boundaries, and state mutations of the critical operational methods.
4.1
DebeziumChangeEventCapturepublic void setup(Properties props, DebeziumRecordParserService debeziumRecordParserService, boolean forceStart)props(configuration properties),debeziumRecordParserService(record parser),forceStart(flag).recordsqueue; createsClickHouseBatchExecutor; registers Debezium callbacks; starts background engine thread.ClassNotFoundExceptionorIOException; retries startup up toMAX_RETRIES(10) with 10s backoff.void handleChangeEventBatch(List[ChangeEvent] list, RecordCommitter committer, Properties props, DebeziumRecordParserService parser, ClickHouseSinkConnectorConfig config)nextSequenceNumberanddrainBeforeDDL.appendToRecords; triggersdrainBeforeDDL()on DDL events; trackslastControlRecord; pushes DML batches torecordsqueue; invokescommitControlRecordOffset().InterruptedExceptionif interrupted during queue handoff.static synchronized long nextSequenceNumber(long recordTs, SourcePosition position)ReplacingMergeTree.recordTs(source.ts_ms),position(SourcePosition).longversion (effectiveTs * 1_000_000L + sequenceNumber).synchronized (DebeziumChangeEventCapture.class).sequenceAnchorTs,sequenceNumber,sequenceHighWaterPosition, andsequenceMaxSourceTs.private void drainBeforeDDL()executor.pause()(acquiresgate) andexecutor.awaitQuiescent()(acquiresgate).IllegalStateExceptionifrecordsqueue does not empty within 60s or executor fails to quiesce.boolean commitControlRecordOffset(ChangeEvent controlRecord, RecordCommitter committer, boolean handedOffRows)controlRecord,committer,handedOffRows(flag if current batch produced DML rows).boolean(true if committed).isPipelineQuiescent() == true) andhandedOffRows == false.OFFSET_COMMIT_LOCKviaDebeziumOffsetManagement.acknowledgeRecords.replica_source_info.4.2
ClickHouseBatchExecutorpublic void pause()synchronized (gate).isPaused = true; callsgate.notifyAll().public void beforeExecute(Thread t, Runnable r)t, taskr.synchronized (gate).isPausedis true, callsgate.wait(100). IncrementsactiveBatches.public void afterExecute(Runnable r, Throwable t)r, exceptiont.synchronized (gate).activeBatches; callsgate.notifyAll().public boolean awaitQuiescent(long timeoutMs)timeoutMs(maximum wait time).boolean(true ifactiveBatches == 0).synchronized (gate).gate.wait()untilactiveBatches.get() == 0or timeout elapses.public void resume()synchronized (gate).isPaused = false; callsgate.notifyAll().4.3
GroupInsertQueryWithBatchRecordspublic boolean groupQueryWithRecords(List[ClickHouseStruct] records, Map[QueryTemplate, List[ClickHouseStruct]] queryToRecordsMap, Map[TopicPartition, Long] partitionToOffsetMap, ClickHouseSinkConnectorConfig config, String tableName, String databaseName, Connection connection, Map[String, String] columnNameToDataTypeMap)boolean(success status).queryToRecordsMapandpartitionToOffsetMap; may alter table schema or update cache viarefreshIfRecordHasUnknownColumn.StaleSchemaCacheExceptionif record and ClickHouse schema conflict irreconcilably.private Map[String, String] refreshIfRecordHasUnknownColumn(ClickHouseStruct record, Map[String, String] cached, String tableName, String databaseName, Connection connection, ClickHouseSinkConnectorConfig config)CacheInvalidationManagerconcurrent maps.system.columnsqueries; executesALTER TABLEifMATERIALIZEDcolumn needs conversion toDEFAULT; marks proven absent columns.private boolean enforceSourceColumnIsWritable(String columnName, String tableName, String databaseName, String fullyQualifiedTableName, Connection connection, ClickHouseSinkConnectorConfig config)MATERIALIZEDcolumns to writableDEFAULTcolumns when the source database supplies values for them.boolean(true if column was successfully converted and is now writable).default_kindanddefault_expressionfromsystem.columns. EmitsALTER TABLE db.table MODIFY COLUMN col type DEFAULT (expr)to ClickHouse.ALIAS, expression cannot be read, or DDL execution fails.4.4
PreparedStatementFieldMapperpublic void insertPreparedStatement(Map[String, Integer] columnNameToIndexMap, PreparedStatement ps, List[Field] fields, ClickHouseStruct record, Struct struct, boolean beforeSection, ClickHouseSinkConnectorConfig config, Map[String, String] columnNameToDataTypeMap, DBMetadata.TABLE_ENGINE engine, String tableName)PreparedStatement.ps, schema fields, CDC record, data struct,beforeSectionflag, config, column types, engine type, table name.psmust be initialized with placeholders matchingcolumnNameToIndexMap.psviasetObject(),setString(),setInt(),setNull().StaleSchemaCacheExceptionif a column exists in ClickHouse and the record but has no index placeholder; throwsSQLExceptionon type binding error.public void insertTombstonePreparedStatement(Map[String, Integer] columnNameToIndexMap, PreparedStatement ps, List[Field] fields, ClickHouseStruct record, Struct struct, ClickHouseSinkConnectorConfig config, Map[String, String] columnNameToDataTypeMap, DBMetadata.TABLE_ENGINE engine, String tableName)ReplacingMergeTreedelete tombstone row for an UPDATE that changes the row sorting key.ps, before-image fields, record, before struct, config, column types, engine, table name.insertPreparedStatementwithbeforeSection = true. Overrides delete column to1(or-1) and overrides version column torecord.getVersion() - 1.4.5
DebeziumOffsetManagementpublic static synchronized boolean checkIfBatchCanBeCommitted(List[ClickHouseStruct] batch)boolean(true if batch was committed; false if buffered).synchronized (DebeziumOffsetManagement.class).inFlightBatches, moves batch tocompletedBatches. Otherwise callsacknowledgeRecords(batch)and drains qualifyingcompletedBatches.public static void acknowledgeRecord(RecordCommitter committer, SourceRecord record, boolean isLastInBatch)committer, Kafkarecord,isLastInBatchflag.synchronized (OFFSET_COMMIT_LOCK).committer.markProcessed(record). IfisLastInBatchis true, invokescommitter.markBatchFinished().RuntimeExceptiononInterruptedException.5. Architectural Critique: Incoherencies & Complexity Traps
A rigorous analysis of the 2.11.0 codebase reveals that recurring operational bugs (stuck offsets, silent data divergence, snapshot loops, metadata query storms, and DDL drain deadlocks) are not isolated defects, but direct consequences of fundamental architectural mismatches.
5.1 Impedance Mismatch: Relational In-Place Mutation vs. Columnar Append-Only Storage
ReplacingMergeTree,CollapsingMergeTree)._version,_sign,is_deleted) and insert tombstone rows to represent deletes.ORDER BYsorting key, ClickHouseReplacingMergeTreecannot replace the old row (since the sorting key differs). The connector is forced to synthesize an artificial tombstone for the old key and an insert for the new key.FINALorargMaxis explicitly queried, shifting the burden of consistency onto the query layer.5.2 The Arithmetic Version Bleed & Restart Inversion Defect
_version = ts_ms * 1_000_000L + sequenceNumber.ts_msis multiplied by 10^6, allocating 6 decimal digits forsequenceNumber.SEQUENCE_STARTis initialized to1_000_000_000(10 decimal digits), and on restartSEQUENCE_START_INITIALis set to500_000_000(9 decimal digits).SEQUENCE_START_INITIAL(500m) is chosen so that replayed records receive lower versions than their pre-restart writes (seeded at 1000m).(T + 1) * 1e6 + 500,000,000 is strictly less than T * 1e6 + 1,000,000,000
ReplacingMergeTreepermanently discards the newer write, causing silent data loss.5.3 Global Static State Anti-Pattern & Multi-Tenancy Hazards
DebeziumChangeEventCapture.sequenceNumber,sequenceAnchorTs,sequenceHighWaterPosition,sequenceMaxSourceTs.DebeziumOffsetManagement.inFlightBatches,completedBatches,outstandingBatches,OFFSET_COMMIT_LOCK.CacheInvalidationManager.INSTANCE.5.4 DDL Synchronization Fragility & Thread Pool Deadlocks
drainBeforeDDL()is inherently fragile.beforeExecute(), preventing any dequeuing and guaranteeing a 60s timeout abort that discarded in-flight batches.drainBeforeDDL()to time out and abort the DDL, triggering continuous restart loops.5.5 Metadata Cache Storms & Tri-Partite Schema Engines
MySqlDDLParserService) in lightweight mode.PostgresSchemaChangeDetector) in lightweight mode.ClickHouseAlterTablein Kafka mode.refreshIfRecordHasUnknownColumnwas previously unable to distinguish between a stale schema cache and ClickHouse-computed columns (MATERIALIZED/ALIAS). This drove an unbounded metadata loop (observed at 9.66 millionsystem.columnsqueries/hour) that exhausted ClickHouse connection pools and stalled replication.5.6 Offset Storage Semaphore Leaks & Masked Divergence
EmbeddedEngine,commitOffsets()leaks theflushInProgresssemaphore if multiple threads callmarkProcessed()andbeginFlush()concurrently without global mutual e 10000 xclusion.ClickHouseErrorClassifiercategorized the resultingConnectException: OffsetStorageWriter is already flushingas anUNKNOWNretriable error. With infinite retries, the connector continued inserting data into ClickHouse while offsets remained frozen, allowing replication to drift hours behind without failing loudly.5.7 Code Duplication & Divergence
ClickHouseBatchWriter(single-threaded mode) andClickHouseBatchRunnable(multi-threaded mode) duplicate approximately 1,000 lines of nearly identical pipeline logic (getDbWriterForTable,processRecordsByTopic,flushRecordsToClickHouse).appendToRecordsWithHashRouting) remains partially implemented but hardcoded touseHashRouting = falseinClickHouseBatchRunnable, leaving dead code with busy-spin loops (Thread.sleep(10)).6. Blueprint for Architectural Simplification
To transition the sink connector from a reactive, bug-patching posture into a robust, high-performance replication engine, the following architectural redesign is recommended at the specification level:
Conclusion
By mapping the system to the tee, this specification clarifies every critical path in the 2.11.0 release. The systemic bugs observed in production—offset stalls, version inversions, DDL deadlocks, and metadata storms—stem from shared mutable state, arithmetic version encoding, and decoupled asynchronous DDL execution. Addressing these issues via deterministic log-based versioning, barrier-based DDL synchronization, and pipeline unification will provide a stable, coherent foundation for all future development.