fix: eliminate the DDL/DML race that silently drops column values by minguyen9988 · Pull Request #1432 · Altinity/clickhouse-sink-connector

minguyen9988 · 2026-09-03T05:44:05Z

fix: NULL-valued columns dropped from the INSERT, so ClickHouse writes the DEFAULT (#1389 follow-up)

A column that is genuinely NULL in the source row is omitted from the
generated INSERT entirely, so ClickHouse applies the column DEFAULT
(0 / '' / 1970-01-01) instead of NULL. Row counts still match, so a
count-based checksum reports the table clean; only a value-level
comparison catches it.

Root cause

ClickHouseStruct#setAfterStruct builds the "modified fields" list by
keeping only the fields whose value is != null:

for (Field f : schemaFields) {
    if (s.get(f) != null) {          // <-- NULL columns filtered out
        this.afterModifiedFields.add(f);
    }
}

PR #1389 then made QueryFormatter#createColumns decide INSERT membership
from that list, to stop a record buffered before an ALTER TABLE ADD
COLUMN from NULL-filling the newly added column. That fix is correct for
the case it targeted, but the list it consults conflates two different
things:

a column the record does not carry at all (pre-ALTER) -> must be
omitted, so ClickHouse applies the DEFAULT the ALTER established;
a column the record carries whose value is NULL -> must be
bound, so the cell lands as NULL.

Both look identical in the value-filtered list, so the second case
inherited the first case's treatment and its value stopped being written.

The two are distinguishable: a pre-ALTER record does not carry the column
in its SCHEMA, whereas a NULL-valued column is in the schema and only
absent from the value-filtered list. The schema is therefore the correct
authority for INSERT membership; the value-filtered list never was.

Fix

getInsertQueryUsingInputFunction takes the record's unfiltered schema
fields as a new trailing argument and decides membership from them,
falling back to the previous behaviour when null. The existing 7- and
8-argument signatures are untouched, so no caller is forced to change.
GroupInsertQueryWithBatchRecords resolves the schema from whichever image
the value-filtered list came from, by identity, so an UPDATE's
before-image is never resolved against the after-image's schema.

The binder needed no change: PreparedStatementFieldMapper already calls
ps.setNull() when struct.get(colName) returns null. The value never
reached it because the column had no placeholder to bind.

Impact observed in production

On txnrepo-sink-uat after the 2.10.1 -> 2.10.3 upgrade, with the
connector logging "Column index missing for column" and writing the row
anyway:

59,929  txnrepo_uat.event                kafka_offset
 3,264  aerion_uat.aerion_break_detail   break_original_date, estimate_resolution_date
 1,951  aerion_uat.aerion_rec            jump_count, clearing_count, rec_end_datetime, fail_reason
   433  trade_uat.enriched_trade         capped_by

Verification (maven 3.9.11 / JDK 17)

QueryFormatterNullValueDropTest 4/4 pass (formatter contract)
NullValueColumnDropTest 4/4 pass (production entry point)
with the fix neutralised 5 of 8 FAIL, emitting
INSERT INTO event(id,request,_version,is_deleted)
-- kafka_offset absent from the statement
sink-connector module suite 271 run, 0 failures, 20 errors
untouched baseline 3c0759b 263 run, 0 failures, 20 errors

The delta between the two runs is exactly the 8 tests added here. The 20
errors are pre-existing and environmental: every one is testcontainers
"Could not find a valid Docker environment" on a host with no Docker
daemon, and they reproduce identically on the untouched baseline.

Both new suites include a regression guard asserting that a record whose
schema genuinely lacks the column is still omitted, so #1389 cannot be
undone by this change.

Co-authored-by: OmniWatcher dba@jumptrading.com

Added release notes for 2.10.1 release

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

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

…#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.

…rd (#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 = debeziumRecordParserServ 8000 ice.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

…ion (#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 (#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

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

) 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

…ing (#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

…ot 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 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: the offset table name arity guard rejects every valid value (#1379) (#1430) The #1379 reporter kept seeing WARN [DbWriter] - Skipping creating offset schema history table as the query was not provided in configuration and went looking for a schema-history misconfiguration. There is none: on PostgreSQL there is no schema history at all. PostgresConnectorConfig extends RelationalDatabaseConnectorConfig, not the Historized variant -- Postgres rebuilds schema from logical decoding and the live catalog on every start -- and config_postgres.yml correspondingly sets zero schema.history keys. The WARN is about neither schema history nor a missing query. It is an arity check on the offset TABLE NAME, and it is off by one. getOffsetStorageDatabaseName splits offset.storage.jdbc.table.name on a dot and rejects the value when the result has length of 2 or less. The property is a qualified database.table pair, so a correct value has exactly two parts. Measured on the value every sample config in this repo ships, altinity_sink_connector.replica_source_info: split length 2 guard "less than or equal to 2" rejects, returns null guard "less than 2" accepts, db=altinity_sink_connector So the method returned null for every correctly configured deployment, and the two lines immediately after the guard -- which read parts 0 and 1 -- were only ever reachable for a three-part name that no config in the tree uses. ensureDatabasesExist then skipped its createDestinationDatabase call for the offset database. That last part is why this is not an outage: on the lightweight path DebeziumChangeEventCapture calls DebeziumJdbcStorageOperations.createDatabaseForDebeziumStorage at startup, which creates the same database from the same property with a correct two-part parse. The DbWriter path is dead code that has been silently doing nothing. The damage is the message: it is emitted on every start, it names a mechanism it is not about, and it says a query was not provided when one was. It cost this issue a round trip. Fixed the comparison, reworded both warnings to name the actual property and the actual condition, and extracted the parse to a package-private static so it can be tested without a database. The DbWriter constructor opens a connection and burns the full retry budget, so a test that built a writer just to read a config string took minutes and drowned in connection errors. Second defect, same area, found while confirming the first: k8s/clickhouse-sink-connector-lt-cm1-configmap.yaml sets ONLY the deprecated alias offset.storage.jdbc.offset.table.name. Debezium accepts that alias, but this connector does not read it. DebeziumJdbcStorageOperations composes the lookup key by literal concatenation of Field.name(), which yields only the canonical name, and throws IllegalArgumentException when it comes back null. That configmap cannot start the connector. Switched to the canonical offset.storage.jdbc.table.name. Both YAML layers, the configmap and the config.yml string it embeds, were re-parsed after the edit. The alias form for ddl, delete and select is left alone: those are read by Debezium itself, which honours the aliases. The two other configs that mention the alias for table.name, docker/config_oracle.yml and the systemd deploy template, already set the canonical key too, so they work; not touched. Verification: - OffsetStorageDatabaseNameTest, 5 cases, no Docker required, 0.3s: the shipped two-part default resolves, a three-part name still resolves to its first segment, and an unqualified, unset or empty value is still rejected. - Failing-first verified: restoring only the old comparison fails testShippedDefaultResolves with expected altinity_sink_connector but was null; the other four still pass, so the fix cannot be satisfied by accepting everything. - sink-connector unit suite: 262 run, 0 failures, 19 errors. All 19 are testcontainers "Could not find a valid Docker environment" on this build host, none from this change. Docker-dependent integration suites were not run locally; CI exercises them. Destructive-Op-Check: sites=1; result=pass Co-authored-by: OmniWatcher <svc-dba-github@jumptrading.com> --------- Co-authored-by: OmniWatcher <svc-dba-github@jumptrading.com>

…s the DEFAULT (#1389 follow-up) A column that is genuinely NULL in the source row is omitted from the generated INSERT entirely, so ClickHouse applies the column DEFAULT (0 / '' / 1970-01-01) instead of NULL. Row counts still match, so a count-based checksum reports the table clean; only a value-level comparison catches it. Root cause ---------- ClickHouseStruct#setAfterStruct builds the "modified fields" list by keeping only the fields whose value is != null: for (Field f : schemaFields) { if (s.get(f) != null) { // <-- NULL columns filtered out this.afterModifiedFields.add(f); } } PR #1389 then made QueryFormatter#createColumns decide INSERT membership from that list, to stop a record buffered before an ALTER TABLE ADD COLUMN from NULL-filling the newly added column. That fix is correct for the case it targeted, but the list it consults conflates two different things: - a column the record does not carry at all (pre-ALTER) -> must be omitted, so ClickHouse applies the DEFAULT the ALTER established; - a column the record carries whose value is NULL -> must be bound, so the cell lands as NULL. Both look identical in the value-filtered list, so the second case inherited the first case's treatment and its value stopped being written. The two are distinguishable: a pre-ALTER record does not carry the column in its SCHEMA, whereas a NULL-valued column is in the schema and only absent from the value-filtered list. The schema is therefore the correct authority for INSERT membership; the value-filtered list never was. Fix --- getInsertQueryUsingInputFunction takes the record's unfiltered schema fields as a new trailing argument and decides membership from them, falling back to the previous behaviour when null. The existing 7- and 8-argument signatures are untouched, so no caller is forced to change. GroupInsertQueryWithBatchRecords resolves the schema from whichever image the value-filtered list came from, by identity, so an UPDATE's before-image is never resolved against the after-image's schema. The binder needed no change: PreparedStatementFieldMapper already calls ps.setNull() when struct.get(colName) returns null. The value never reached it because the column had no placeholder to bind. Impact observed in production ----------------------------- On txnrepo-sink-uat after the 2.10.1 -> 2.10.3 upgrade, with the connector logging "Column index missing for column" and writing the row anyway: 59,929 txnrepo_uat.event kafka_offset 3,264 aerion_uat.aerion_break_detail break_original_date, estimate_resolution_date 1,951 aerion_uat.aerion_rec jump_count, clearing_count, rec_end_datetime, fail_reason 433 trade_uat.enriched_trade capped_by Verification (maven 3.9.11 / JDK 17) ------------------------------------ QueryFormatterNullValueDropTest 4/4 pass (formatter contract) NullValueColumnDropTest 4/4 pass (production entry point) with the fix neutralised 5 of 8 FAIL, emitting INSERT INTO `event`(`id`,`request`,`_version`,`is_deleted`) -- kafka_offset absent from the statement sink-connector module suite 271 run, 0 failures, 20 errors untouched baseline 3c0759b 263 run, 0 failures, 20 errors The delta between the two runs is exactly the 8 tests added here. The 20 errors are pre-existing and environmental: every one is testcontainers "Could not find a valid Docker environment" on a host with no Docker daemon, and they reproduce identically on the untouched baseline. Both new suites include a regression guard asserting that a record whose schema genuinely lacks the column is still omitted, so #1389 cannot be undone by this change. Co-authored-by: OmniWatcher <dba@jumptrading.com>

… correction to this PR's diagnosis Adds NullColumnValueRoundTripIT, an end-to-end integration test (MySQL 8.0.36 ROW/FULL/GTID -> connector -> ClickHouse 24.8, real containers via testcontain 10000 ers) covering the four failure shapes observed on txnrepo-sink-uat: 1. NULL on INSERT stays NULL (txnrepo_uat.event.kafka_offset) plus the silent blob case (trade_uat.changelog.request) 2. UPDATE-to-NULL clears the value (trade_uat.enriched_trade.capped_by) 3. several NULL columns in one row (aerion_uat.aerion_rec) 4. pre-ALTER row still gets DEFAULT (#1389 regression guard) Every assertion is value-level and asserts NULL specifically, because the defect substitutes a type-dependent DEFAULT (0 / '' / 1970-01-01) while row counts stay identical -- which is exactly why the daily value-level checksum caught this and count-based checks did not. Column types are varied deliberately so a fix that handled only integers cannot pass. Each scenario carries a populated control row, so the suite cannot pass by the connector writing NULL everywhere. IMPORTANT -- this IT does NOT confirm this PR's diagnosis --------------------------------------------------------- Run against real containers: fix applied 4/4 pass fix neutralised 4/4 pass <-- should have failed The statement generated in the IT already binds every column BEFORE the fix: INSERT INTO `event`(`id`,`kafka_offset`,`request`,`recorded_at`, `_version`,`is_deleted`) VALUES (?,?,?,?,?,?) So on this path a NULL-valued column is not dropped, and the production data loss does not reproduce. The unit tests in the previous commit are genuine -- they fail without the change and pass with it -- but they reach createColumns with a value-filtered field list that the live connector evidently does not produce for a plain INSERT with explicit NULLs. Debezium appears to emit the full after-image including nulls, so setAfterStruct's `s.get(f) != null` filter does not strip the column the way those fixtures assume. The code change therefore stands as hardening -- the record's schema is the correct authority for INSERT membership, not a value-filtered list -- but it is NOT proven to be the cause of the production loss and must not be described as the fix for it until something reproduces that loss. What remains established independently: the divergence is real and value-level verified against the source MySQL host (3,348 rows on event.kafka_offset in one day, 240 on enriched_trade.capped_by, plus a fully silent blob loss on changelog), and the connector logged "Column index missing for column" 65,577 times across four tables and wrote the rows anyway. The symptom and the error branch are established; the mechanism proposed here is not. Still to eliminate: a stale columnNameToIndexMap reused across a schema change; getWithoutDefault under non.default.value yielding a different field set; a before-image list used against the after-image schema; and the bitemporal/history-mode write path these tables actually use. The IT is committed regardless of which mechanism proves responsible: it pins the four production shapes end to end, so whatever the cause turns out to be, a regression that reintroduces it has to get past these assertions. Verification (maven 3.9.11 / JDK 17, rootless podman socket) ------------------------------------------------------------ NullColumnValueRoundTripIT 4/4 pass, 414s, real MySQL + ClickHouse sink-connector suite 271 run, 0 failures, 20 errors untouched baseline 3c0759b 263 run, 0 failures, 20 errors The 20 errors are pre-existing testcontainers "Could not find a valid Docker environment" failures in the sink-connector module and reproduce identically on the untouched baseline. Co-authored-by: OmniWatcher <dba@jumptrading.com>

… comment)

minguyen9988 · 2026-09-03T06:08:01Z

Correction from the author — the end-to-end reproduction does NOT confirm this diagnosis

I am flagging this against my own PR before anyone spends review time on it.

I wrote an end-to-end integration test (NullColumnValueRoundTripIT, MySQL 8.0.36 -> connector -> ClickHouse 24.8, real containers) covering the four production failure shapes: NULL on insert, NULL blob, UPDATE-to-NULL, and several NULL columns in one row.

It passes with the fix in place — and it also passes with the fix reverted. That is a negative result on my own hypothesis, not a confirmation:

NullColumnValueRoundTripIT, fix applied     4/4 pass
NullColumnValueRoundTripIT, fix neutralised 4/4 pass   <-- should have failed

The generated statement in the IT already binds every column before the fix:

INSERT INTO `event`(`id`,`kafka_offset`,`request`,`recorded_at`,`_version`,`is_deleted`) VALUES (?,?,?,?,?,?)

So on this path a NULL-valued column is not dropped, and the unit-level failure I based the diagnosis on does not reproduce end to end.

What this means

The unit tests in this PR are genuine — QueryFormatterNullValueDropTest and NullValueColumnDropTest both fail without the change and pass with it, and the statement they produce really does omit kafka_offset. But they reach createColumns with a value-filtered field list, and the live connector evidently does not, at least for a plain MySQL INSERT ... VALUES (null, ...). Debezium appears to emit the full after-image with explicit nulls, so setAfterStruct's s.get(f) != null filter does not strip the column the way my unit fixtures assume.

Therefore: the code change here is defensible as hardening (schema is the right authority for INSERT membership, not a value-filtered list), but it is NOT proven to be the cause of the production data loss, and it should not be described as the fix for it until something reproduces that loss.

What is solid, independent of the above

The production divergence is real and value-level verified against the source MySQL host: txnrepo_uat.event.kafka_offset 3,348 rows in one day, trade_uat.enriched_trade.capped_by 240, plus a fully silent blob loss on trade_uat.changelog.
The connector's own error log records Column index missing for column 65,577 times across four tables, then writes the row anyway.

So the symptom and the error are established; the mechanism I proposed is not.

What I am doing next

Finding the input shape that actually produces the dropped column — the error log proves the connector reaches that branch in production, so something in the real record stream differs from my fixture. Candidates I have not yet eliminated:

A stale columnNameToIndexMap reused across a schema change, so the map predates a column the current table has.
getWithoutDefault under non.default.value, which yields a different field set than struct.get.
A path where the before-image's filtered list is used against the after-image's schema.
Something specific to the bitemporal/history-mode write path these tables use.

Until one of those is demonstrated, treat this PR as unverified. I would rather say that now than have it merged as a fix for a bug it may not fix.

The production loss is a race between DDL and DML, not a NULL-handling bug. A record captured under the pre-ALTER schema is bound against the post-ALTER table: the row inserts successfully, the affected column takes the ClickHouse DEFAULT (0 / 1970-01-01 / empty) instead of its real value, and row counts stay identical on both sides. Nothing fails. A daily value-level checksum was the only thing that noticed, two days later. Five defects, each of which alone reopens the window. 1. pause()/awaitQuiescent() was a check-then-act race. beforeExecute() tested isPaused and incremented activeBatches as two separate steps, so a pool thread could read isPaused==false, be overtaken by pause() plus a drain that saw zero in-flight batches and returned TRUE, and then run its batch against the just-altered table. Both operations now happen in one critical section on a private monitor, so a successful drain means no batch is running AND none can start. pause()/resume()/afterExecute() publish through the same monitor and use wait/notify rather than sleep-polling. 2. The drain applied the DDL anyway on timeout. drainBeforeDDL() logged a warning and proceeded, which is precisely the corrupting interleaving it exists to prevent. It now throws, routing into the existing DDL retry path. It also pauses BEFORE draining the queue -- draining while the pool can still pick up work is unwinnable on a busy table. 3. A failed DDL left the writer paused forever. executor.resume() sat after performDDLOperation() with no finally; with (2) now able to throw, that path would stop replication dead. The resume is unconditional. 4. Unresolvable DDL left every cache stale. Invalidation only covered table names getTableNamesFromDDL() could resolve. Unresolvable or failed cases now call CacheInvalidationManager.invalidateAll(), which bumps a global epoch folded into every per-table version -- a per-table map alone cannot express "everything is stale", since a table never individually invalidated reads version 0 before and after the sweep. 5. Dropped columns were logged and written anyway. A data column with no placeholder was logged at ERROR and skipped; a column absent from the record but present in the cached map was bound to NULL, overwriting the stored value on an UPDATE (the RENAME/DROP half of #1389). Both now throw StaleSchemaCacheException, failing the batch into the retry path. Connector-managed and history-mode columns keep the previous behaviour. Defence in depth: GroupInsertQueryWithBatchRecords re-reads table metadata when the record carries a column the cached map does not know about. QueryFormatter decides INSERT membership from the record schema rather than the value-filtered modified-fields list. Verification (maven 3.9.11 / JDK 17, rootless podman): PauseDrainAtomicityTest 2/2 pass; with fix reverted 1/2 FAIL StaleSchemaCacheIT 4/4 pass, 465s, real MySQL 8.0.36 -> ClickHouse 24.8, DDL against a LIVE connector with a cached writer PauseDrainRaceTest 3/3 pass sink-connector suite 276 run, 0 failures, 20 errors baseline 3c0759b 263 run, 0 failures, 20 errors The 20 errors are pre-existing testcontainers no-Docker failures and reproduce identically on the baseline. PauseDrainRaceTest passes against the BROKEN executor too (verified by reverting), so it is a behavioural guard, not the proof. PauseDrainAtomicityTest is the proof: it asserts atomicity directly by holding the monitor and does fail on the broken code. The earlier NullColumnValueRoundTripIT is superseded for the same reason -- it wrote NULLs to a fresh table, where no staleness exists. Destructive-Op-Check: sites=1; result=pass Co-authored-by: OmniWatcher <dba@jumptrading.com>

minguyen9988 · 2026-09-03T06:39:10Z

fix: eliminate the DDL/DML race that silently drops column values

The production loss is a race between DDL and DML, not a NULL-handling
bug. A record captured under the pre-ALTER schema is bound against the
post-ALTER table: the row inserts successfully, the affected column takes
the ClickHouse DEFAULT (0 / 1970-01-01 / empty) instead of its real
value, and row counts stay identical on both sides. Nothing fails. A
daily value-level checksum was the only thing that noticed, two days
later.

Five defects, each of which alone reopens the window.

pause()/awaitQuiescent() was a check-then-act race. beforeExecute()
tested isPaused and incremented activeBatches as two separate steps, so a
pool thread could read isPaused==false, be overtaken by pause() plus a
drain that saw zero in-flight batches and returned TRUE, and then run its
batch against the just-altered table. Both operations now happen in one
critical section on a private monitor, so a successful drain means no
batch is running AND none can start. pause()/resume()/afterExecute()
publish through the same monitor and use wait/notify rather than
sleep-polling.
The drain applied the DDL anyway on timeout. drainBeforeDDL() logged a
warning and proceeded, which is precisely the corrupting interleaving it
exists to prevent. It now throws, routing into the existing DDL retry
path. It also pauses BEFORE draining the queue -- draining while the pool
can still pick up work is unwinnable on a busy table.
A failed DDL left the writer paused forever. executor.resume() sat
after performDDLOperation() with no finally; with (2) now able to throw,
that path would stop replication dead. The resume is unconditional.
Unresolvable DDL left every cache stale. Invalidation only covered
table names getTableNamesFromDDL() could resolve. Unresolvable or failed
cases now call CacheInvalidationManager.invalidateAll(), which bumps a
global epoch folded into every per-table version -- a per-table map alone
cannot express "everything is stale", since a table never individually
invalidated reads version 0 before and after the sweep.
Dropped columns were logged and written anyway. A data column with no
placeholder was logged at ERROR and skipped; a column absent from the
record but present in the cached map was bound to NULL, overwriting the
stored value on an UPDATE (the RENAME/DROP half of fix: no-PK tables collapse to one row; pre-ALTER rows NULL-fill added columns #1389). Both now throw
StaleSchemaCacheException, failing the batch into the retry path.
Connector-managed and history-mode columns keep the previous behaviour.

Defence in depth: GroupInsertQueryWithBatchRecords re-reads table
metadata when the record carries a column the cached map does not know
about. QueryFormatter decides INSERT membership from the record schema
rather than the value-filtered modified-fields list.

Verification (maven 3.9.11 / JDK 17, rootless podman):
PauseDrainAtomicityTest 2/2 pass; with fix reverted 1/2 FAIL
StaleSchemaCacheIT 4/4 pass, 465s, real MySQL 8.0.36 ->
ClickHouse 24.8, DDL against a LIVE
connector with a cached writer
PauseDrainRaceTest 3/3 pass
sink-connector suite 276 run, 0 failures, 20 errors
baseline 3c0759b 263 run, 0 failures, 20 errors

The 20 errors are pre-existing testcontainers no-Docker failures and
reproduce identically on the baseline.

PauseDrainRaceTest passes against the BROKEN executor too (verified by
reverting), so it is a behavioural guard, not the proof.
PauseDrainAtomicityTest is the proof: it asserts atomicity directly by
holding the monitor and does fail on the broken code. The earlier
NullColumnValueRoundTripIT is superseded for the same reason -- it wrote
NULLs to a fresh table, where no staleness exists.

Destructive-Op-Check: sites=1; result=pass
Co-authored-by: OmniWatcher dba@jumptrading.com

Base moved on two commits (#1432 DDL/DML race, #1429 db_compare). One file conflicted -- DebeziumChangeEventCapture -- in two hunks, and in both the correct resolution is neither side verbatim. 1. DDL cache invalidation ------------------------- Both sides fixed something real: base (#1432) an unresolvable DDL invalidated NOTHING while having changed the schema, so writers kept binding against a shape that no longer existed. Falls back to invalidateAll(). branch a PostgreSQL schema-change event may name no table in its tableChanges array, but the topic still identifies it; also invalidates the schema-drift detector cache. Taking base loses Postgres DDL invalidation entirely (Postgres events routinely resolve to an empty table list, so every Postgres DDL would fall through to a fleet-wide sweep, and the drift detector would keep serving a stale schema until its TTL). Taking ours loses the safety net for genuinely unresolvable DDL. Merged as a ladder: resolve from the event, then from the topic, and only if BOTH fail invalidate everything. The Postgres path keeps its targeted invalidation including the drift-detector cache; the base's fail-safe still catches what neither can name. 2. The null-record early return -- load-bearing for #1379 ---------------------------------------------------------- The branch had: chStruct = ...parse(...); if (chStruct == null) { log.debug("Skipping record with null ClickHouseStruct ..."); return null; } chStruct.setSequenceNumber(sequenceNumber); This predates both #1416 and #1428 and is INCOMPATIBLE with the fix the base now carries. Dropped in favour of the base deliberately, not as a conflict-resolution convenience: - a heartbeat parses to null by contract. Under #1428 the caller (handleChangeEventBatch) must still SEE that record so commitControlRecordOffset can acknowledge its offset -- on an idle source the post-snapshot state rides exclusively on heartbeats; - returning early means the record never reaches the control-record path, replica_source_info stays pinned at snapshot=INITIAL/snapshot_completed=false forever, and every restart re-runs the whole snapshot. That is precisely the reported bug. A comment now marks the spot so the early return is not reintroduced. Both #1416 and #1428 reach this branch through this merge and not before it: they are on develop, and 2.11.0 carries them as cherry-picks rather than as ancestors, so neither was an ancestor of the PR head. Before this merge the branch still had the pre-#1416 code at the parse site and none of the #1428 machinery (handleChangeEventBatch / commitControlRecordOffset / isPipelineQuiescent). Everything else auto-merged (55 files). Verified rather than assumed: - no conflict markers anywhere in the tree - the offset-table DDL is ORDER BY offset_key in all 27 shipped configs including the runtime default and all three Postgres ones (the other half of #1428; the single remaining "ORDER BY id" is an unrelated test fixture table in sql.py, not the offset store) - handleChangeEventBatch is wired into the engine's only ChangeConsumer, so the Postgres path reaches it identically to MySQL Build: both modules, mvn clean install -DskipTests, green under JDK 17.0.2 / Maven 3.9.11. Both destructive sites are pre-existing upstream content carried in by the merge, not introduced here: a documentation string describing the disable.drop.truncate option in the k8s configmap, and an already- annotated DROP COLUMN in StaleSchemaCacheIT scoped to a throwaway testcontainer table. Destructive-Op-Check: sites=2; result=pass

… 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 _vers FCA7 ion 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 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: 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 …

subkanthi and others added 19 commits August 26, 2026 21:57

Added release notes for 2.10.1 release

1e3ffd0

Merge pull request #1412 from Altinity/2_10_1_release_notes

eeb37b9

Added release notes for 2.10.1 release

Added release notes for 2.10.2

f9606ae

Merge pull request #1422 from Altinity/2_10_2_release_notes

04682b7

Added release notes for 2.10.2

Fix datatype integration tests

549fcc9

Merge pull request #1424 from Altinity/2_10_2_fix_integration_tests

689fd32

Fix datatype integration tests

Added release notes for 2.10.3

e19703d

Merge pull request #1427 from Altinity/release_notes_2_10_3

e7da4d6

Added release notes for 2.10.3

minguyen9988 changed the title ~~WIP: DO NOT MERGE - fix: NULL-valued columns dropped from the INSERT, so ClickHouse writes the DEFAULT~~ WIP: DO NOT MERGE - NEEDS VERIFICATION - NULL-valued columns dropped from the INSERT (#1389 follow-up) Sep 3, 2026

Jenkins Continuous Build server and others added 2 commits September 3, 2026 01:07

chore: drop the transfer-only caveat file (its text is posted as a PR…

00dd763

… comment)

minguyen9988 changed the title ~~WIP: DO NOT MERGE - NEEDS VERIFICATION - NULL-valued columns dropped from the INSERT (#1389 follow-up)~~ NEEDS VERIFICATION - NULL-valued columns dropped from the INSERT (#1389 follow-up) Sep 3, 2026

minguyen9988 requested a review from subkanthi September 3, 2026 06:12

minguyen9988 changed the base branch from develop to 2.11.0 September 3, 2026 06:13

minguyen9988 changed the title ~~NEEDS VERIFICATION - NULL-valued columns dropped from the INSERT (#1389 follow-up)~~ WIP: DO NOT MERGE - fix: eliminate the DDL/DML race that silently drops column values Sep 3, 2026

minguyen9988 changed the title ~~WIP: DO NOT MERGE - fix: eliminate the DDL/DML race that silently drops column values~~ fix: eliminate the DDL/DML race that silently drops column values Sep 3, 2026

minguyen9988 merged commit 9c12939 into 2.11.0 Sep 3, 2026
1 check passed

minguyen9988 deleted the omniwatcher/fix-null-column-dropped-from-insert branch September 3, 2026 06:44

minguyen9988 mentioned this pull request Sep 3, 2026

PostgreSQL to ClickHouse CDC Support (rebase of #1245 onto 2.10.0) #1398

Closed

aadant mentioned this pull request Sep 5, 2026

ERROR - ***** Column index missing for column #1433

Open

minguyen9988 mentioned this pull request Sep 7, 2026

Port #1431 + #1398 onto the recreated 2.11.0 branch (PostgreSQL CDC + DDL/DML race fix) #1435

Merged

Navigation Menu

Repository navigation

fix: eliminate the DDL/DML race that silently drops column values - #1432

fix: eliminate the DDL/DML race that silently drops column values#1432
minguyen9988 merged 22 commits into
2.11.0from
omniwatcher/fix-null-column-dropped-from-insert

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

Root cause

Fix

Impact observed in production

Verification (maven 3.9.11 / JDK 17)

Uh oh!

Correction from the author — the end-to-end reproduction does NOT confirm this diagnosis

What this means

What is solid, independent of the above

What I am doing next

Uh oh!

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants