Conversation
Snapshot of sink-connector/python from minguyen/postgres (PR #1245) for review and merge independent of the Postgres/Java connector changes. Includes Python-focused .gitignore entries for the toolkit. Made-with: Cursor
…connector into feature/ch-sink-tools-python
…e-records-are-not-persisted Fixed operation Delete records not getting persisted.
…db_time-current-time Added db_time column.
…e-deserialization-rce-via-crafted-config-file-cve-2022-1471 Upgrade snakeyaml to 2.2 to avoid CVE
Added REST API authentication and updated test.
improve version_history.md
Add ClickHouseErrorClassifier that categorizes ClickHouse exceptions as FATAL or RETRIABLE based on error codes extracted from exception messages. Fatal errors (auth failures, schema mismatches, resource limits) will never succeed on retry. Previously, the connector silently swallowed all exceptions in ClickHouseBatchRunnable.run(), retaining the failed batch in currentBatch and retrying it indefinitely. This blocked binlog position advancement, causing silent data loss across all tables in the replication pipeline. Changes: - New ClickHouseErrorClassifier utility with error code extraction, fatal/retriable classification, and cause-chain traversal - Modified ClickHouseBatchRunnable.run() catch block to classify errors: - FATAL: log error, clear currentBatch, rethrow to stop the executor - RETRIABLE/UNKNOWN: log warning, allow normal retry on next run - Comprehensive unit tests for all classified error codes Fixes #1310 Related: #1308
…t-failure-v2 Classify ClickHouse errors and stop on fatal failures instead of retrying forever
… RENAME enterTruncateTable() produced 'db.db.table' when MySQL used fully-qualified table names because it prepended databaseName without checking for dots. enterDropTable() omitted the databaseName prefix entirely. enterRenameTable() used && instead of || when checking for dots. All three now follow the enterAlterTable() pattern: check for dot, split, and prepend configured databaseName. Added unit tests for all three fixes. Fixes #1308
…base-prefix-v2 Fix double database prefix in DDL replication for TRUNCATE, DROP, and RENAME
ClickHouseSinkTask.preCommit() is a pass-through: it returns the offsets Kafka Connect delivered to put(), regardless of whether those records were actually inserted into ClickHouse. Inserts are asynchronous (ClickHouseBatchRunnable drains an in-memory queue on a schedule), so a task crash/restart after preCommit but before the durable insert silently skips the consumed-but-not- inserted records -> data loss, with no error surfaced. The existing durable gating (DebeziumOffsetManagement.acknowledgeRecords) only works in the embedded Debezium engine path, where it calls RecordCommitter.markProcessed(). In Kafka Connect mode the RecordCommitter is null, so that path is a no-op and offset commit relies entirely on preCommit(). Track, per TopicPartition, the highest offset durably inserted: - ClickHouseBatchRunnable advances a shared watermark (merge max) after each successful flush, from the partitionToOffsetMap it already computes. - ClickHouseSinkTask.put() seeds a per-partition baseline (firstOffset - 1) so preCommit holds at the resume point until a real insert advances it (rather than trusting the delivered offsets, which would still lose records if CH is down from task start). - preCommit() returns min(deliveredOffset, durable + 1) per partition. No data loss on crash/restart; at-least-once redelivery of records inserted but not yet committed, deduplicated by ReplacingMergeTree(version). Known limitation: the watermark uses max() per partition, correct in legacy single-threaded batch processing (FIFO per partition => max == contiguous). With hash-based multi-threaded routing, a higher-offset batch could complete before a lower one and over-advance the watermark; that path would need a contiguous (gap-aware) watermark. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…slight-change Updated show_replica_status view
…connector into 1305-mysql-alter-from-smallint-unsigned-to-bigint-unsigned
…-unsigned-to-bigint-unsigned Fix unsigned mapping(workaround debezium bug)
…mismatch Adding integration test to reproduce reinsert after delete.
…ert-version-mismatch Revert "Adding integration test to reproduce reinsert after delete."
…st-part-of-github-actions-for-checksum-validations Add integration tests to validate checksums using pytest.
…rced from sysbench
…connector into sysbench_automated_tests # Conflicts: # sink-connector-lightweight/tests/checksum/conftest.py
Drop ReplicationHistoryHandler and ClickHouseStruct diffs so PR #1354 only ships the sysbench checksum tests. evert:
Sysbench automated tests
…ng, per-table LOCK TABLES READ lifecycle
Python tooling only - no Java changes.
- fstr() in mysql_table_checksum.py, clickhouse_table_checksum.py, mysql_table_count.py was `eval(f"f'{template}'")`: any text reaching the template executed as Python. Replaced with literal `{partition_expression}` substitution; mysql_table_checksum_test.py (previously a placeholder that asserted True == False on every run) now pins that the eval cannot come back.
- Thread-pool result loops read future.result() BEFORE checking future.exception(), so the first exception propagated un-logged and later ones were lost; now handled per-future with the table name logged.
- top_level_table_checksum.py: per-table `LOCK TABLES ... READ` with try/finally unlock + connection close (lock released even when a checksum subprocess dies). New tests/test_table_locking.py covers the lock lifecycle.
- db/clickhouse.py: cursor closed in finally; yaml.safe_load; credentials no longer logged in debug output.
- db/mysql.py: user/password URL-quoted so special characters in credentials do not break the SQLAlchemy URL.
- tests/__main__.py: parameterized INSERT instead of %-interpolated SQL.
Verified: `python -m unittest db_compare.tests.mysql_table_checksum_test db_compare.tests.test_table_locking` - all pass.
Part of the split of #1353 into independently mergeable sub-PRs (each <= 10 files), so the 2.10.0 branch can absorb the fixes incrementally.
…run, shlex-quote shell args Python tooling only. - mysql_dumper.py / clickhouse_loader.py logged the full shell command line including `--password <secret>`. Secrets are now registered at the point they enter a command line and masked by exact value (register_secret()/redact_password()); parsing shell quoting is not reliable because shlex.quote() splits a password containing a quote into concatenated segments. - Password and config-file arguments are shlex.quote()d instead of hand-quoted. - clickhouse_loader.py: the dry_run argument was accepted but hardcoded to False at the mysqlshell call site - now passed through; timezone fallback loop returned the last iterated zone instead of UTC on no-match. Part of the split of #1353 into independently mergeable sub-PRs (each <= 10 files), so the 2.10.0 branch can absorb the fixes incrementally.
…lation
Integration-test helpers only (testflows); no shipped code.
cluster.py interpolated SQL into a double-quoted shell word (`echo -e "{sql}" | mysql`), which corrupts payloads containing quotes/backslashes and is injection-prone. SQL now travels through a temp file. Because ~90 call sites spell identifiers with backslash-escaped backticks relying on bash's double-quote unescaping, the temp-file path performs the same removal itself - exactly the four characters bash unescapes inside double quotes (backtick, dollar, double-quote, backslash), leaving \n, \N, \% untouched.
Companion tweaks in common.py and the steps modules keep the suite consistent with that path.
Part of the split of #1353 into independently mergeable sub-PRs (each <= 10 files), so the 2.10.0 branch can absorb the fixes incrementally.
…able.drop.truncate Helm chart only. - snapshot.mode becomes configurable (connector.snapshotMode), defaulting to schema_only - the chart had it pinned to initial, which re-snapshots on every fresh deploy. - disable.drop.truncate becomes configurable (connector.disableDropTruncate), defaulting to true. Uses a key-presence check rather than Sprig `default` because `default` treats boolean false as empty - `--set connector.disableDropTruncate=false` would silently render "true". Part of the split of #1353 into independently mergeable sub-PRs (each <= 10 files), so the 2.10.0 branch can absorb the fixes incrementally.
… hardening SourceRecordParserService hardening for malformed/edge-case source records (null value schema, missing fields) so a poison record fails loudly with context instead of an opaque NPE. New SourceRecordParserServiceTest pins the behavior. Part of the split of #1353 into independently mergeable sub-PRs (each <= 10 files), so the 2.10.0 branch can absorb the fixes incrementally.
[split 4/22] helm: restore snapshot.mode=schema_only default, add configurable disable.drop.truncate
…mula Bulk DELETE + re-INSERT (e.g. a nightly refresh) could permanently lose rows: after an offset rewind, Debezium re-delivers the DELETE with a fresh processing timestamp, so it received a HIGHER _version than the later re-INSERT and ReplacingMergeTree kept the row stuck is_deleted=1. The fix anchors the version sequence to the SOURCE commit timestamp (source.ts_ms), which is identical on every re-delivery, so a re-delivered DELETE keeps its original version and can never out-rank the re-INSERT. Unlike the reverted #1347 (and #1355), the emitted formula is UNCHANGED from 2.8.0: ts_ms * 1_000_000 + counter. Values stay in the same numeric domain, so the binary remains a drop-in replacement in BOTH directions: upgrade from 2.8.0 is safe, and rollback to 2.8.0/2.10.0 remains possible (no un-supersedable version domain is introduced). The intra-second counter is keyed exclusively on the source commit clock - never on the binlog file name or position - so it is preserved across binary log rotations: two commits in the same second on either side of a rotation keep incrementing the same counter and cannot collide or invert. The anchor is global (survives batch boundaries) and never moves backward, closing the duplicate-_version race where a re-delivered older timestamp re-armed the counter reset. Also fixes the snapshot marker parsing: Debezium emits an enum string ("true", "first", "last", "incremental", ...), not a boolean; Boolean.parseBoolean misclassified "first"/"last" as streaming records. Snapshot records keep the historical processing-time anchor (their source.ts_ms is a snapshot read time with different clock semantics, and snapshots are re-read, not re-delivered, so #1346 does not apply to them). SourceTsVersionAnchorTest pins all of it: the 2.8.0 domain, the #1346 delete/re-insert scenario, counter preservation across rotation, the monotonic anchor, and the no-source-ts fallback. All 6 tests fail against the previous logic and pass with the fix.
…all directives with edge-case tests Restores the historically intended (and upstream-removed in 686cfea) initial-seed semantics, per aadant's directives on PR #1378: 1. The intra-second counter is kept even if the binary log rotates. Already implemented (counter keyed exclusively on the source commit clock; file name / position play no role). Now additionally proven inside a re-publication scenario where the rotation happens between the re-published DELETE and re-INSERT (republicationPreservesOrderAcrossRotation). 2. The first time you resume, the counter is 500m. On the first batch after start/resume (sequenceAnchorTs == 0), the counter is now seeded at SEQUENCE_START_INITIAL (500,000,000) instead of continuing from SEQUENCE_START (1,000,000,000). Debezium re-publishes every event after the last committed offset on resume; those re-published events were already written pre-restart with counters in the 1000m range, so the 500m seed guarantees every re-published duplicate ranks strictly BELOW its original write and can never supersede it. The first >1s source-clock advance resets to SEQUENCE_START, returning to the normal domain (initialSeedEscapesToNormalDomainAfterOneSecond). Both counter sites (streaming loop and addVersion) are seeded. 3. Edge cases including re-publication, all pinned in SourceTsVersionAnchorTest (9 tests): - resumeSeedsCounterAtInitial: restart mid-second; re-published event gets 500m+1 and ranks below the pre-restart 1000m write - republicationPreservesOrderAcrossRotation: crash + resume replays DELETE and re-INSERT across a binlog rotation; DELETE < re-INSERT preserved, replayed versions never out-rank originals - counterKeptAcrossBinlogRotation, anchorNeverMovesBackward, counterResetsOnlyOnSourceClockAdvance, staysInTwoEightZeroDomain, redeliveredDeleteCannotOutrankReinsert, fallsBackToProcessingTimestamp Proof of test sensitivity: with the seed reverted (pr1378head code), 6 of 9 tests fail; with it applied, 9/9 pass. Full offline suite (63 classes) run on both this commit and unmodified pr1378head: failing-class sets are IDENTICAL (47 pre-existing Docker-unavailable/environment classes) - zero regressions introduced. Version280CompatibilityTest (2.8.0 domain pin), DebeziumChangeEventCaptureTest and SequenceNumberRaceTest all pass: 500m and 1000m counters both stay within the ts_ms*1e6+counter domain.
…um-safety [split 1/22] python: remove eval() from checksum fstr(), fix silent error swallowing, per-table LOCK TABLES READ lifecycle
[split 2/22] python: mask passwords in db_dump/db_load command logging, honor dry_run, shlex-quote shell args
…-datatype [split 5/22] lightweight: SourceRecordParserService null-schema and error handling hardening
Fix #1346: redelivery-stable _version anchored to source commit timestamp - keeps the 2.8.0 formula (upgrade- and rollback-safe)
…-helpers [split 3/22] testflows: pass SQL to clients via temp file instead of shell interpolation
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
6 participants
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.
2.10.0 release