Repository navigation
fix(memory): sample the byte estimate once per table in a multi-table group, and close the connection a table writer re-acquired when its worker is discarded (Spec 01.05 section 3.4 item 7, Spec 01.01 section 3.3 step 4a) - #1508
Merged
minguyen9988 merged 2 commits intoSep 26, 2026
Merged
Conversation
…oup -- a single-threaded batch spans every table it touches, so its first row charged a narrow table's width to every row of a wide table behind it and the byte cap could not see the heap those rows pinned (Spec 01.05 section 3.4 item 7) Routed mode hands off one table per group, so nothing changes there. In single-threaded (legacy) mode registerHandoff receives the WHOLE Debezium batch as one group; RecordSizeEstimator.estimateGroup sampled group.get(0) and stamped that per-row estimate on every row. A batch that opens with a few rows of a narrow table and continues with megabyte BLOB rows of another was charged the narrow width throughout, so outstanding bytes stayed far below the truth and sink.connector.handoff.max.outstanding.bytes never engaged for it. estimateGroup now keeps a per-call map of topic -> per-row estimate, samples the first row of each table it meets, stamps every row with its own table's estimate and returns the sum. The map has at most one entry per table in the group and lives for the call only. A routed group still costs one walk. Test: RecordSizeEstimatorTest.multiTableGroupIsSampledPerTable -- a narrow row followed by two rows of a wide table; fails on the previous code with the wide rows stamped at the narrow estimate.
… writer re-acquired -- BaseDbWriter.getConnection() swaps a stale handle for a fresh pool checkout that only the writer knows, so closing the per-database map alone left one pool slot leaked per reconnected writer per restart (Spec 01.01 section 3.3 step 4a) The restart cleanup added for the worker pool closed databaseToConnectionMap and the system connection. A DbWriter is built on the per-database connection, but BaseDbWriter.getConnection() replaces a closed or evicted handle with HikariDbSource.initiateNewConnectionIfClosed -- a checkout from the pool that is stored in the writer's own field and never enters the worker's map. On the next engine restart the map's stale original was closed (a no-op) and the replacement stayed checked out for the life of the process; enough restarts after enough reconnects and the pool ran dry. closeConnections() now also walks topicToDbWriterMap, closes every held connection that is not one of the per-database connections just closed, and forgets the writers. BaseDbWriter.heldConnection() returns the current handle WITHOUT re-acquiring one: getConnection() would open a connection in order to close it, and against a pool that is already gone would log an error for nothing. Test: ClickHouseBatchRunnableCloseConnectionsTest .closeConnectionsClosesAWriterReacquiredConnection -- builds a writer on a tracked per-database connection, performs the same field assignment getConnection() does with a second tracked connection, and asserts both are closed; fails on the previous code with the re-acquired one still open.
minguyen9988
added a commit
to minguyen9988/clickhouse-sink-connector
that referenced
this pull request
Sep 26, 2026
…odes Upstream squash-merged the Domain 12 specification as 7396c3c (Altinity#1507) and then landed Altinity#1508-Altinity#1510; the ten add/add conflicts were the squash artefact (upstream copies byte-identical to the spec commit) and were resolved to the corrected specs and Lean model of this branch. The Java merges were automatic and orthogonal (ResumeReplayLogSummary hooks in DebeziumChangeEventCapture, writer-connection close in ClickHouseBatchRunnable). Both modules compile; the history/DDL/clamp unit suites and the resume-replay and close-connection suites are green; validate_specs.py --lake passes (68 specs, 748 citations). Destructive-Op-Check: sites=0; result=pass
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.
Two follow-ups to the memory bounds of #1505 and the restart cleanup of #1506, found by auditing the code that landed on 2.11.0 for structures that grow without a bound in bytes or across engine restarts.
1. The byte estimate is sampled once per TABLE, not once per group (Spec 01.05 section 3.4 item 7)
Routed mode hands off one table per group, so nothing changes there. In single-threaded (legacy) mode registerHandoff receives the WHOLE Debezium batch as one group (DebeziumChangeEventCapture: registerHandoff(convertedRecords, singletonList(convertedRecords))). RecordSizeEstimator.estimateGroup sampled group.get(0) and stamped that per-row estimate on every row: a batch that opens with a few rows of a narrow table and continues with megabyte BLOB rows of another was charged the narrow width throughout, so the outstanding byte count stayed far below the truth and sink.connector.handoff.max.outstanding.bytes never engaged for it.
estimateGroup now keeps a per-call map of topic to per-row estimate, samples the first row of each table it meets, stamps every row with its own table estimate and returns the sum. The map has at most one entry per table in the group and lives for the call only; a routed group still costs one walk.
Test: RecordSizeEstimatorTest.multiTableGroupIsSampledPerTable (a narrow row followed by two rows of a wide table). Mutation check: with the previous estimateGroup body it fails with the wide rows stamped at 1718 bytes instead of 6293270.
2. A discarded worker also closes the connection each table writer re-acquired (Spec 01.01 section 3.3 step 4a)
#1506 closes databaseToConnectionMap and the system connection when the pool terminates. A DbWriter is built on the per-database connection, but BaseDbWriter.getConnection() replaces a closed or evicted handle with HikariDbSource.initiateNewConnectionIfClosed(...), a checkout from the pool that is stored in the writer field alone and never enters the worker map. On the next engine restart the map entry (the stale original) was closed, a no-op, and the replacement stayed checked out for the life of the process: one pool slot per reconnected writer per restart, until the pool ran dry.
closeConnections() now also walks topicToDbWriterMap, closes every held connection that is not one of the per-database connections just closed, and forgets the writers. BaseDbWriter.heldConnection() returns the current handle WITHOUT re-acquiring one: getConnection() would open a connection in order to close it, and against a pool that is already gone would log an error for nothing.
Test: ClickHouseBatchRunnableCloseConnectionsTest.closeConnectionsClosesAWriterReacquiredConnection builds a writer on a tracked per-database connection, performs the same field assignment getConnection() does with a second tracked connection, and asserts both are closed. Mutation check: with the previous closeConnections body it fails with the re-acquired connection still open.
Specs
Tests run (JDK 17, offline Maven)
Audit result for the rest of the 2.11.0 delta since #1502
Checked and found bounded: RecordSizeEstimator (no cache, per-call state only), DebeziumOffsetManagement byte accounting (added once under the monitor, subtracted once in drainCompletedUnits, zeroed by reset(); a late report for a retired unit touches no counter), BatchChunker (a row wider than buffer.max.bytes lands alone in a chunk, no zero-progress loop), Metrics tags (topic, topic+partition, one binlog-file child), the legacy queue (500000 batches behind the byte cap) and the Debezium queue preflight (max(64 MiB, maxMemory/16) when unset).