Repository navigation
fix: the offset table name arity guard rejects every valid value (#1379) - #1430
Merged
minguyen9988 merged 1 commit intoSep 2, 2026
Conversation
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
minguyen9988
added a commit
that referenced
this pull request
Sep 2, 2026
…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>
Collaborator
Author
|
sorry I merged wrongly |
minguyen9988
added a commit
that referenced
this pull request
Sep 3, 2026
) * 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 PASS 8BD0 ED 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 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> * 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> * test: end-to-end coverage for the txnrepo-uat NULL-column loss, and a 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 testcontainers) 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> * chore: drop the transfer-only caveat file (its text is posted as a PR comment) * 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. 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> --------- 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: Jenkins Continuous Build server <jenkins@fpif1-dbal2.w2k.jumptrading.com> Co-authored-by: OmniWat 72A8 cher <dba@jumptrading.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Follow-up to #1428, on top of that branch. Fixes the last loose end in #1379: the WARN the reporter kept asking about, plus a k8s configmap that cannot start the connector at all.
WIP: DO NOT MERGEbecause this targetsfix-1379-control-record-offset, which is itself still open as #1428 — merge that first.1.
getOffsetStorageDatabaseNamerejects every valid valueThe reporter kept seeing this and went looking for a schema-history misconfiguration:
There is nothing to find. On PostgreSQL there is no schema history at all —
PostgresConnectorConfigextendsRelationalDatabaseConnectorConfig, not the Historized variant, because Postgres rebuilds schema from logical decoding and the live catalog on every start.config_postgres.ymlcorrespondingly sets zeroschema.history.*keys, correctly.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:
The property is a qualified
database.tablepair, so a correct value has exactly two parts. Measured on the value every sample config in this repo ships:So the method returned
nullfor every correctly configured deployment — and note the two lines immediately after the guard read[0]and[1], i.e. they assume the two-part shape the guard just refused. They were only ever reachable for a three-part name that no config in the tree uses.Why this is not an outage, and what the damage actually is. On the lightweight path
DebeziumChangeEventCapturecallsDebeziumJdbcStorageOperations.createDatabaseForDebeziumStorageat startup, which creates the same database from the same property with a correct two-part parse. TheDbWriterpath is dead code that has been silently doing nothing. The real cost is the message: emitted on every start, naming a mechanism it is not about, and asserting a query was not provided when one was. It cost this issue a round trip.Fixed the comparison and reworded both warnings to name the actual property and the actual condition.
2. The parse is now testable without a database
Extracted to a package-private static.
DbWriter's constructor opens a connection, so a test that built a writer just to read a config string spent the full connection-retry budget first — the first version of this test took minutes and drowned inRetry Number: 9 of 10 Error creating Database. The new one runs in 0.3s with no Docker.3. Second defect: a k8s configmap that cannot start the connector
k8s/clickhouse-sink-connector-lt-cm1-configmap.yamlsets only the deprecated aliasoffset.storage.jdbc.offset.table.name.Debezium accepts that alias. This connector does not read it:
DebeziumJdbcStorageOperationscomposes its lookup key by literal concatenation ofField.name(), which yields only the canonical name, and throwsIllegalArgumentExceptionwhen the lookup comes back null. Probed against the pinned Debezium 3.1.3.Final:Switched to the canonical key. Both YAML layers — the configmap and the
config.ymlstring it embeds — were re-parsed after the edit. The alias form for.ddl/.delete/.selectis 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.ymland the systemd deploy template) already set the canonical key alongside it, so they work. Not touched.Verification
OffsetStorageDatabaseNameTest— 5 cases, no Docker, 0.3s. The shipped two-part default resolves, a three-part name still resolves to its first segment, and an unqualified / unset / empty value is still rejected.testShippedDefaultResolveswithexpected: <altinity_sink_connector> but was: <null>, while the other four still pass — so the fix cannot be satisfied by simply accepting everything.sink-connectorunit suite: 262 run, 0 failures, 19 errors. All 19 are testcontainersCould not find a valid Docker environmenton the build host; none are from this change.Docker-dependent integration suites were not run locally; CI exercises them.
Still open on #1379, not in this PR
The restart-survival integration test.
PostgresInitialDockerITassertsoffsetValue.contains("snapshot"), which passes on both"snapshot":"INITIAL"and any completed shape, and it never restarts the engine — so the re-snapshot symptom is outside its coverage entirely. #1428's two tests are unit-level. APostgresSnapshotResumeITthat stands up Postgres + ClickHouse, waits past one heartbeat interval, asserts exactly one offset row andsnapshot_completed, restarts, and asserts the snapshot does not re-run would close it. Worth doing separately rather than widening this PR.Created by OmniWatcher on behalf of the Jump Trading DBA team; requires human review before merging.