Sink should stop on errors

[Bens-ct]

I've noticed a few times now that we have had a small inconsistency in the data synced from MySQL to ClickHouse. It looks to be missing records like they were skipped entirely.

Unfortunately tracking down when it happens and seeing logs isn't easy I'm running ~ 12 instances each one syncing ~ 3000 tables across 50 databases and running on EKS using spot instances so logs aren't currently persisted and usually by the time we noticed it, there is nothing to check.

I suspect what is happening is that clickhouse and / or keeper is restarted and during that time the sink exceeds the number of retries and skips some of the queries

I think it is probably similar to what is captured here #463 and although there are retries, it looks like after the retry count is hit it moves on, whereas that probably keeps things moving, it does so at the risk of data integrity.

For me, I'd rather have it keep retrying and fall behind requiring manual intervention rather than have data issues that could be business critical.

It may well be there is an option to halt on errors but I can't see one in the docs or indeed from a quick scan of the code.

[subkanthi]

Hi @Bens-ct ,
After the retry, it should not move on, if clickhouse restarted, the sink connector should only retry the same batch again. Are u running 2.8.0?, the batches are acknowledged(persisted to CH offset table) only after the inserts are successful.
The failures you see were those DDL failures or inserts?

[Bens-ct]

Hi, @subkanthi

I am running 2.8.0

The failures I see are inserts, effectively I end up with some missing rows in clickhouse, but due to the number of tables involved and the size of some of them, we tend not to noticed until some days/weeks later when any of logs that may have been useful have gone.

This is the the config I am using in case I am doing something daft.

name: 'Name'
database.hostname: 'hostname'
database.port: '3306'
database.user: 'debezium'
database.password: 'password'
database.include.list: "dblist"
table.include.list: "table_list"
clickhouse.database.override.map: "overridemap"
schema.history.internal.store.only.captured.tables.ddl: 'true'
schema.history.internal.store.only.captured.databases.ddl: 'true'
connector.class: 'io.debezium.connector.mysql.MySqlConnector'
database.server.id: '1759741339'
database.server.name: 'Name'
topic.prefix: 'Name'
snapshot.mode: 'no_data'
snapshot.fetch.size: '10240'
snapshot.max.threads: '1'
snapshot.locking.mode: 'minimal'
incremental.snapshot.chunk.size: '1024'
incremental.snapshot.allow.schema.changes: 'true'
offset.storage: 'io.debezium.storage.jdbc.offset.JdbcOffsetBackingStore'
offset.storage.jdbc.url: "jdbc:clickhouse://clickhouse-cdc-cluster.clickhouse.svc.cluster.local:8123/altinity_sink_connector"
offset.storage.jdbc.user: 'user'
offset.storage.jdbc.password: 'pass'
offset.storage.jdbc.table.name: 'altinity_sink_connector.offset'
offset.storage.jdbc.table.ddl: "CREATE TABLE if not exists %s ON CLUSTER 'main' (id String, offset_key String, offset_val String, record_insert_ts DateTime, record_insert_seq UInt64, _version UInt64 MATERIALIZED toUnixTimestamp64Nano(now64(9))) ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/altinity_sink_connector/offset_{table}', '{replica}', version) ORDER BY offset_key SETTINGS index_granularity = 8192"
offset.storage.jdbc.table.delete: "SELECT * FROM %s"
offset.storage.jdbc.table.select: "SELECT id, offset_key, offset_val FROM %s FINAL ORDER BY record_insert_ts, record_insert_seq"
offset.flush.interval.ms: '60000'
schema.history.internal: 'io.debezium.storage.jdbc.history.JdbcSchemaHistory'
schema.history.internal.jdbc.url: "jdbc:clickhouse://clickhouse-cdc-cluster.clickhouse.svc.cluster.local:8123/altinity_sink_connector"
schema.history.internal.jdbc.user: 'user'
schema.history.internal.jdbc.password: 'pass'
schema.history.internal.jdbc.schema.history.table.name: 'altinity_sink_connector.schema_history'
schema.history.internal.jdbc.schema.history.table.ddl: "CREATE TABLE if not exists %s ON CLUSTER 'main' (id String, history_data String, history_data_seq Int32, record_insert_ts DateTime, record_insert_seq Int32) ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/altinity_sink_connector/schema_history{table}', '{replica}', record_insert_seq) ORDER BY id SETTINGS index_granularity = 8192"
schema.history.internal.jdbc.schema.history.table.delete: "delete from %s where 1=1"
clickhouse.server.url: 'clickhouse-cdc-cluster.clickhouse.svc.cluster.local'
clickhouse.server.port: '8123'
clickhouse.server.user: 'user'
clickhouse.server.password: 'pass'
auto.create.tables: 'false'
replacingmergetree.delete.column: '_deleted'
clickhouse.datetime.timezone: 'Europe/London'
database.connectionTimeZone: 'Europe/London'
single.threaded: 'true'
metrics.port: '8087'
metrics.enable: 'true'
include.schema.changes: 'true'
enable.snapshot.ddl: 'false'
auto.create.tables.replicated: 'true'
errors.retry.timeout: '600'
clickhouse.jdbc.params: "socket_timeout=120000,connection_timeout=60000,keepalive.timeout=300000,max_buffer_size=10485760"
max.batch.size: '10000'
max.queue.size: '50000'
use.nongraceful.disconnect: 'true'
database.keep.alive.interval.ms: '30000'
database.connection.reconnect.backoff.ms: '1000'
database.connection.reconnect.backoff.max.ms: '10000'
database.ssl.mode: 'disabled'
replica.status.view: "CREATE VIEW IF NOT EXISTS %s.show_replica_status ON CLUSTER 'main' AS SELECT now() - fromUnixTimestamp(JSONExtractUInt(offset_val, 'ts_sec')) AS seconds_behind_source, toDateTime(fromUnixTimestamp(JSONExtractUInt(offset_val, 'ts_sec')), 'UTC') AS utc_time, fromUnixTimestamp(JSONExtractUInt(offset_val, 'ts_sec')) AS local_time FROM %s settings final=1"
thread.pool.size: '2'
connection.pool.max.size: '500'
connection.pool.min.idle: '15'
connection.pool.timeout: '300000'
connection.pool.max.lifetime: '1800000'
clickhouse.settings.max_partitions_per_insert_block: '1000'
schema.history.internal.skip.unparseable.ddl: 'true'
ignore.ddl.regex: '(?i)^\sALTER\s+TABLE\s+\S+\s+AUTO_INCREMENT\s=\s*\d+\s*$'

Thanks

[aadant]

@Bens-ct do you have some logs to share with us that shows the problem ?

Do you see any errors on insert in the system.query_log in ClickHouse ?

The expected behavior is to retry forever when there is a DML (insert) or DDL error.

This option in theory can be problematic but it is usually fine.

schema.history.internal.skip.unparseable.ddl: 'true'

[subkanthi]

Hi @Bens-ct

Sometimes the rows are merged because of the ReplacingMergeTree merge process, it might be better to investigate the ORDER BY columns

Hi, @subkanthi

I am running 2.8.0

The failures I see are inserts, effectively I end up with some missing rows in clickhouse, but due to the number of tables involved and the size of some of them, we tend not to noticed until some days/weeks later when any of logs that may have been useful have gone.

This is the the config I am using in case I am doing something daft.

name: 'Name'
database.hostname: 'hostname'
database.port: '3306'
database.user: 'debezium'
database.password: 'password'
database.include.list: "dblist"
table.include.list: "table_list"
clickhouse.database.override.map: "overridemap"
schema.history.internal.store.only.captured.tables.ddl: 'true'
schema.history.internal.store.only.captured.databases.ddl: 'true'
connector.class: 'io.debezium.connector.mysql.MySqlConnector'
database.server.id: '1759741339'
database.server.name: 'Name'
topic.prefix: 'Name'
snapshot.mode: 'no_data'
snapshot.fetch.size: '10240'
snapshot.max.threads: '1'
snapshot.locking.mode: 'minimal'
incremental.snapshot.chunk.size: '1024'
incremental.snapshot.allow.schema.changes: 'true'
offset.storage: 'io.debezium.storage.jdbc.offset.JdbcOffsetBackingStore'
offset.storage.jdbc.url: "jdbc:clickhouse://clickhouse-cdc-cluster.clickhouse.svc.cluster.local:8123/altinity_sink_connector"
offset.storage.jdbc.user: 'user'
offset.storage.jdbc.password: 'pass'
offset.storage.jdbc.table.name: 'altinity_sink_connector.offset'
offset.storage.jdbc.table.ddl: "CREATE TABLE if not exists %s ON CLUSTER 'main' (id String, offset_key String, offset_val String, record_insert_ts DateTime, record_insert_seq UInt64, _version UInt64 MATERIALIZED toUnixTimestamp64Nano(now64(9))) ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/altinity_sink_connector/offset_{table}', '{replica}', version) ORDER BY offset_key SETTINGS index_granularity = 8192"
offset.storage.jdbc.table.delete: "SELECT * FROM %s"
offset.storage.jdbc.table.select: "SELECT id, offset_key, offset_val FROM %s FINAL ORDER BY record_insert_ts, record_insert_seq"
offset.flush.interval.ms: '60000'
schema.history.internal: 'io.debezium.storage.jdbc.history.JdbcSchemaHistory'
schema.history.internal.jdbc.url: "jdbc:clickhouse://clickhouse-cdc-cluster.clickhouse.svc.cluster.local:8123/altinity_sink_connector"
schema.history.internal.jdbc.user: 'user'
schema.history.internal.jdbc.password: 'pass'
schema.history.internal.jdbc.schema.history.table.name: 'altinity_sink_connector.schema_history'
schema.history.internal.jdbc.schema.history.table.ddl: "CREATE TABLE if not exists %s ON CLUSTER 'main' (id String, history_data String, history_data_seq Int32, record_insert_ts DateTime, record_insert_seq Int32) ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/altinity_sink_connector/schema_history{table}', '{replica}', record_insert_seq) ORDER BY id SETTINGS index_granularity = 8192"
schema.history.internal.jdbc.schema.history.table.delete: "delete from %s where 1=1"
clickhouse.server.url: 'clickhouse-cdc-cluster.clickhouse.svc.cluster.local'
clickhouse.server.port: '8123'
clickhouse.server.user: 'user'
clickhouse.server.password: 'pass'
auto.create.tables: 'false'
replacingmergetree.delete.column: '_deleted'
clickhouse.datetime.timezone: 'Europe/London'
database.connectionTimeZone: 'Europe/London'
single.threaded: 'true'
metrics.port: '8087'
metrics.enable: 'true'
include.schema.changes: 'true'
enable.snapshot.ddl: 'false'
auto.create.tables.replicated: 'true'
errors.retry.timeout: '600'
clickhouse.jdbc.params: "socket_timeout=120000,connection_timeout=60000,keepalive.timeout=300000,max_buffer_size=10485760"
max.batch.size: '10000'
max.queue.size: '50000'
use.nongraceful.disconnect: 'true'
database.keep.alive.interval.ms: '30000'
database.connection.reconnect.backoff.ms: '1000'
database.connection.reconnect.backoff.max.ms: '10000'
database.ssl.mode: 'disabled'
replica.status.view: "CREATE VIEW IF NOT EXISTS %s.show_replica_status ON CLUSTER 'main' AS SELECT now() - fromUnixTimestamp(JSONExtractUInt(offset_val, 'ts_sec')) AS seconds_behind_source, toDateTime(fromUnixTimestamp(JSONExtractUInt(offset_val, 'ts_sec')), 'UTC') AS utc_time, fromUnixTimestamp(JSONExtractUInt(offset_val, 'ts_sec')) AS local_time FROM %s settings final=1"
thread.pool.size: '2'
connection.pool.max.size: '500'
connection.pool.min.idle: '15'
connection.pool.timeout: '300000'
connection.pool.max.lifetime: '1800000'
clickhouse.settings.max_partitions_per_insert_block: '1000'
schema.history.internal.skip.unparseable.ddl: 'true'
ignore.ddl.regex: '(?i)^\sALTER\s+TABLE\s+\S+\s+AUTO_INCREMENT\s=\s*\d+\s*$'

Thanks

[Bens-ct]

I've managed to recreate it and capture the logs. Basically clickhouse is running on spot nodes so may be interrupted (as far as I can tell this is a supported configuration, I've got keeper on on demand nodes).

To recreate I terminated the EKS node that hosts one of the pods from the clickhouse statefullset and I got missing data. I was able to find these logs.

2026-02-02 11:24:54.059 INFO  - *** INSERT QUERY for Database(rs2_real_mysql_db_name) ***: insert into `table_name`(`id`,<snipped cols>,`created_type` Nullable(Int8),`completed_type` Nullable(Int8),`created_at` Nullable(String),`updated_at` Nullable(String),`_version` UInt64,`_deleted` UInt8')
2026-02-02 11:24:54.059 ERROR - Error getting prepared statement, retry attempt (0/2) failed
java.sql.SQLException: Connection is closed
        at com.zaxxer.hikari.pool.ProxyConnection$ClosedConnection.lambda$getClosedConnection$0(ProxyConnection.java:563) ~[app.jar:?]
        at jdk.proxy3.$Proxy28.prepareStatement(Unknown Source) ~[?:?]
        at com.zaxxer.hikari.pool.ProxyConnection.prepareStatement(ProxyConnection.java:328) ~[app.jar:?]
        at com.zaxxer.hikari.pool.HikariProxyConnection.prepareStatement(HikariProxyConnection.java) ~[app.jar:?]
        at com.altinity.clickhouse.sink.connector.db.DBMetadata.getPreparedStatement(DBMetadata.java:749) ~[app.jar:?]
        at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.lambda$executePreparedStatement$0(PreparedStatementExecutor.java:202) ~[app.jar:?]
        at java.lang.Iterable.forEach(Iterable.java:75) ~[?:?]
        at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.executePreparedStatement(PreparedStatementExecutor.java:196) ~[app.jar:?]
        at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.addToPreparedStatementBatch(PreparedStatementExecutor.java:151) ~[app.jar:?]
        at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.flushRecordsToClickHouse(ClickHouseBatchWriter.java:453) ~[app.jar:?]
        at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.processRecordsByTopic(ClickHouseBatchWriter.java:404) ~[app.jar:?]
        at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.persistRecords(ClickHouseBatchWriter.java:221) ~[app.jar:?]
        at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.appendToRecords(DebeziumChangeEventCapture.java:909) ~[app.jar:?]
        at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture$1.handleBatch(DebeziumChangeEventCapture.java:209) ~[app.jar:?]
        at io.debezium.embedded.ConvertingEngineBuilder$ConvertingChangeConsumer.handleBatch(ConvertingEngineBuilder.java:75) ~[app.jar:?]
        at io.debezium.embedded.EmbeddedEngine.pollRecords(EmbeddedEngine.java:731) ~[app.jar:?]
        at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:470) ~[app.jar:?]
        at io.debezium.embedded.ConvertingEngineBuilder$1.run(ConvertingEngineBuilder.java:163) ~[app.jar:?]
        at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.lambda$setupDebeziumEventCapture$0(DebeziumChangeEventCapture.java:290) ~[app.jar:?]
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
        at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
        at java.lang.Thread.run(Thread.java:840) [?:?]
2026-02-02 11:24:54.069 ERROR - ******* ERROR inserting Batch Database(real_mysql_db_name), Table(table_name) *****************
java.sql.BatchUpdateException: Code: 60. DB::Exception: Table system.table_name does not exist. Maybe you meant different_mysql_db_name.table_name?. (UNKNOWN_TABLE) (version 25.8.2.29 (official build))

        at com.clickhouse.jdbc.SqlExceptionUtils.batchUpdateError(SqlExceptionUtils.java:107) ~[app.jar:?]
        at com.clickhouse.jdbc.internal.InputBasedPreparedStatement.executeAny(InputBasedPreparedStatement.java:154) ~[app.jar:?]
        at com.clickhouse.jdbc.internal.AbstractPreparedStatement.executeLargeBatch(AbstractPreparedStatement.java:85) ~[app.jar:?]
        at com.clickhouse.jdbc.internal.ClickHouseStatementImpl.executeBatch(ClickHouseStatementImpl.java:819) ~[app.jar:?]
        at com.zaxxer.hikari.pool.ProxyStatement.executeBatch(ProxyStatement.java:127) ~[app.jar:?]
        at com.zaxxer.hikari.pool.HikariProxyPreparedStatement.executeBatch(HikariProxyPreparedStatement.java) ~[app.jar:?]
        at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.lambda$executePreparedStatement$0(PreparedStatementExecutor.java:258) ~[app.jar:?]
        at java.lang.Iterable.forEach(Iterable.java:75) ~[?:?]
        at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.executePreparedStatement(PreparedStatementExecutor.java:196) ~[app.jar:?]
        at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.addToPreparedStatementBatch(PreparedStatementExecutor.java:151) ~[app.jar:?]
        at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.flushRecordsToClickHouse(ClickHouseBatchWriter.java:453) ~[app.jar:?]
        at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.processRecordsByTopic(ClickHouseBatchWriter.java:404) ~[app.jar:?]
        at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.persistRecords(ClickHouseBatchWriter.java:221) ~[app.jar:?]
        at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.appendToRecords(DebeziumChangeEventCapture.java:909) ~[app.jar:?]
        at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture$1.handleBatch(DebeziumChangeEventCapture.java:209) ~[app.jar:?]
        at io.debezium.embedded.ConvertingEngineBuilder$ConvertingChangeConsumer.handleBatch(ConvertingEngineBuilder.java:75) ~[app.jar:?]
        at io.debezium.embedded.EmbeddedEngine.pollRecords(EmbeddedEngine.java:731) ~[app.jar:?]
        at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:470) ~[app.jar:?]
        at io.debezium.embedded.ConvertingEngineBuilder$1.run(ConvertingEngineBuilder.java:163) ~[app.jar:?]
        at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.lambda$setupDebeziumEventCapture$0(DebeziumChangeEventCapture.java:290) ~[app.jar:?]
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
        at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
        at java.lang.Thread.run(Thread.java:840) [?:?]
2026-02-02 11:24:54.069 ERROR - Error persisting records to ClickHousejava.lang.RuntimeException: java.sql.BatchUpdateException: Code: 60. DB::Exception: Table system.table_name does not exist. Maybe you meant different_mysql_db_name.table_name?. (UNKNOWN_TABLE) (version 25.8.2.29 (official build))

It looks like some kind of connection pool issue where when it loses the connection, it "forgets" what the database name was and switches to trying to work with the system database, logs a bunch of errors and then carries on without ever writing the records to clickhouse.

[Bens-ct]

I've also found it seems to happen when there is no node restart as well so ruling that out as the cause

2026-02-03 20:02:10.172 INFO - ****** Acknowledging records ****** 2026-02-03 20:02:11.697 INFO - ****** Thread: Sink connector Debezium Event Thread Batch Size: 20 ****** 2026-02-03 20:02:11.697 INFO - *** INSERT QUERY for Database(database_name) ***: insert into table_name(id,<col list>_version UInt64,_deleted` UInt8')
2026-02-03 20:02:11.698 ERROR - Error getting prepared statement, retry attempt (0/2) failed
java.sql.SQLException: Connection is closed
at com.zaxxer.hikari.pool.ProxyConnection$ClosedConnection.lambda$getClosedConnection$0(ProxyConnection.java:563) ~[app.jar:?]
at jdk.proxy3.$Proxy28.prepareStatement(Unknown Source) ~[?:?]
at com.zaxxer.hikari.pool.ProxyConnection.prepareStatement(ProxyConnection.java:328) ~[app.jar:?]
at com.zaxxer.hikari.pool.HikariProxyConnection.prepareStatement(HikariProxyConnection.java) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.db.DBMetadata.getPreparedStatement(DBMetadata.java:749) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.lambda$executePreparedStatement$0(PreparedStatementExecutor.java:202) ~[app.jar:?]
at java.lang.Iterable.forEach(Iterable.java:75) ~[?:?]
at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.executePreparedStatement(PreparedStatementExecutor.java:196) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.addToPreparedStatementBatch(PreparedStatementExecutor.java:151) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.flushRecordsToClickHouse(ClickHouseBatchWriter.java:453) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.processRecordsByTopic(ClickHouseBatchWriter.java:404) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.persistRecords(ClickHouseBatchWriter.java:221) ~[app.jar:?]
at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.appendToRecords(DebeziumChangeEventCapture.java:909) ~[app.jar:?]
at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture$1.handleBatch(DebeziumChangeEventCapture.java:209) ~[app.jar:?]
at io.debezium.embedded.ConvertingEngineBuilder$ConvertingChangeConsumer.handleBatch(ConvertingEngineBuilder.java:75) ~[app.jar:?]
at io.debezium.embedded.EmbeddedEngine.pollRecords(EmbeddedEngine.java:731) ~[app.jar:?]
at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:470) ~[app.jar:?]
at io.debezium.embedded.ConvertingEngineBuilder$1.run(ConvertingEngineBuilder.java:163) ~[app.jar:?]
at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.lambda$setupDebeziumEventCapture$0(DebeziumChangeEventCapture.java:290) ~[app.jar:?]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
at java.lang.Thread.run(Thread.java:840) [?:?]
2026-02-03 20:02:11.705 ERROR - ******* ERROR inserting Batch Database(database), Table(table_name) *****************
java.sql.BatchUpdateException: Code: 60. DB::Exception: Table system.table_name does not exist. Maybe you meant different_database.table_name?. (UNKNOWN_TABLE) (version 25.8.2.29 (official build))

    at com.clickhouse.jdbc.SqlExceptionUtils.batchUpdateError(SqlExceptionUtils.java:107) ~[app.jar:?]
    at com.clickhouse.jdbc.internal.InputBasedPreparedStatement.executeAny(InputBasedPreparedStatement.java:154) ~[app.jar:?]
    at com.clickhouse.jdbc.internal.AbstractPreparedStatement.executeLargeBatch(AbstractPreparedStatement.java:85) ~[app.jar:?]
    at com.clickhouse.jdbc.internal.ClickHouseStatementImpl.executeBatch(ClickHouseStatementImpl.java:819) ~[app.jar:?]
    at com.zaxxer.hikari.pool.ProxyStatement.executeBatch(ProxyStatement.java:127) ~[app.jar:?]
    at com.zaxxer.hikari.pool.HikariProxyPreparedStatement.executeBatch(HikariProxyPreparedStatement.java) ~[app.jar:?]
    at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.lambda$executePreparedStatement$0(PreparedStatementExecutor.java:258) ~[app.jar:?]
    at java.lang.Iterable.forEach(Iterable.java:75) ~[?:?]
    at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.executePreparedStatement(PreparedStatementExecutor.java:196) ~[app.jar:?]
    at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.addToPreparedStatementBatch(PreparedStatementExecutor.java:151) ~[app.jar:?]
    at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.flushRecordsToClickHouse(ClickHouseBatchWriter.java:453) ~[app.jar:?]
    at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.processRecordsByTopic(ClickHouseBatchWriter.java:404) ~[app.jar:?]
    at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.persistRecords(ClickHouseBatchWriter.java:221) ~[app.jar:?]
    at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.appendToRecords(DebeziumChangeEventCapture.java:909) ~[app.jar:?]
    at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture$1.handleBatch(DebeziumChangeEventCapture.java:209) ~[app.jar:?]
    at io.debezium.embedded.ConvertingEngineBuilder$ConvertingChangeConsumer.handleBatch(ConvertingEngineBuilder.java:75) ~[app.jar:?]
    at io.debezium.embedded.EmbeddedEngine.pollRecords(EmbeddedEngine.java:731) ~[app.jar:?]
    at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:470) ~[app.jar:?]
    at io.debezium.embedded.ConvertingEngineBuilder$1.run(ConvertingEngineBuilder.java:163) ~[app.jar:?]
    at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.lambda$setupDebeziumEventCapture$0(DebeziumChangeEventCapture.java:290) ~[app.jar:?]
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
    at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
    at java.lang.Thread.run(Thread.java:840) [?:?]

2026-02-03 20:02:11.705 ERROR - Error persisting records to ClickHousejava.lang.RuntimeException: java.sql.BatchUpdateException: Code: 60. DB::Exception: Table system.table_name does not exist. Maybe you meant different_database.table_name?. (UNKNOWN_TABLE) (version 25.8.2.29 (official build))
2026-02-03 20:02:12.206 INFO - ****** Thread: Sink connector Debezium Event Thread Batch Size: 41 ******
2026-02-03 20:02:12.207 INFO - *** INSERT QUERY for Database(database_name) ***: insert into table_name(id,_version UInt64,_deleted UInt8')
2026-02-03 20:02:12.207 ERROR - Error getting prepared statement, retry attempt (0/2) failed
java.sql.SQLException: Connection is closed
at com.zaxxer.hikari.pool.ProxyConnection$ClosedConnection.lambda$getClosedConnection$0(ProxyConnection.java:563) ~[app.jar:?]
at jdk.proxy3.$Proxy28.prepareStatement(Unknown Source) ~[?:?]
at com.zaxxer.hikari.pool.ProxyConnection.prepareStatement(ProxyConnection.java:328) ~[app.jar:?]
at com.zaxxer.hikari.pool.HikariProxyConnection.prepareStatement(HikariProxyConnection.java) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.db.DBMetadata.getPreparedStatement(DBMetadata.java:749) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.lambda$executePreparedStatement$0(PreparedStatementExecutor.java:202) ~[app.jar:?]
at java.lang.Iterable.forEach(Iterable.java:75) ~[?:?]
at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.executePreparedStatement(PreparedStatementExecutor.java:196) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.addToPreparedStatementBatch(PreparedStatementExecutor.java:151) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.flushRecordsToClickHouse(ClickHouseBatchWriter.java:453) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.processRecordsByTopic(ClickHouseBatchWriter.java:404) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.persistRecords(ClickHouseBatchWriter.java:221) ~[app.jar:?]
at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.appendToRecords(DebeziumChangeEventCapture.java:909) ~[app.jar:?]
at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture$1.handleBatch(DebeziumChangeEventCapture.java:209) ~[app.jar:?]
at io.debezium.embedded.ConvertingEngineBuilder$ConvertingChangeConsumer.handleBatch(ConvertingEngineBuilder.java:75) ~[app.jar:?]
at io.debezium.embedded.EmbeddedEngine.pollRecords(EmbeddedEngine.java:731) ~[app.jar:?]
at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:470) ~[app.jar:?]
at io.debezium.embedded.ConvertingEngineBuilder$1.run(ConvertingEngineBuilder.java:163) ~[app.jar:?]
at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.lambda$setupDebeziumEventCapture$0(DebeziumChangeEventCapture.java:290) ~[app.jar:?]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
at java.lang.Thread.run(Thread.java:840) [?:?]
2026-02-03 20:02:12.214 ERROR - ******* ERROR inserting Batch Database(database), Table(table_name) *****************
java.sql.BatchUpdateException: Code: 60. DB::Exception: Table system.table_name does not exist. Maybe you meant different_database.table_name?. (UNKNOWN_TABLE) (version 25.8.2.29 (official build))

    at com.clickhouse.jdbc.SqlExceptionUtils.batchUpdateError(SqlExceptionUtils.java:107) ~[app.jar:?]
    at com.clickhouse.jdbc.internal.InputBasedPreparedStatement.executeAny(InputBasedPreparedStatement.java:154) ~[app.jar:?]
    at com.clickhouse.jdbc.internal.AbstractPreparedStatement.executeLargeBatch(AbstractPreparedStatement.java:85) ~[app.jar:?]
    at com.clickhouse.jdbc.internal.ClickHouseStatementImpl.executeBatch(ClickHouseStatementImpl.java:819) ~[app.jar:?]
    at com.zaxxer.hikari.pool.ProxyStatement.executeBatch(ProxyStatement.java:127) ~[app.jar:?]
    at com.zaxxer.hikari.pool.HikariProxyPreparedStatement.executeBatch(HikariProxyPreparedStatement.java) ~[app.jar:?]
    at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.lambda$executePreparedStatement$0(PreparedStatementExecutor.java:258) ~[app.jar:?]
    at java.lang.Iterable.forEach(Iterable.java:75) ~[?:?]
    at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.executePreparedStatement(PreparedStatementExecutor.java:196) ~[app.jar:?]
    at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.addToPreparedStatementBatch(PreparedStatementExecutor.java:151) ~[app.jar:?]
    at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.flushRecordsToClickHouse(ClickHouseBatchWriter.java:453) ~[app.jar:?]
    at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.processRecordsByTopic(ClickHouseBatchWriter.java:404) ~[app.jar:?]
    at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.persistRecords(ClickHouseBatchWriter.java:221) ~[app.jar:?]
    at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.appendToRecords(DebeziumChangeEventCapture.java:909) ~[app.jar:?]
    at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture$1.handleBatch(DebeziumChangeEventCapture.java:209) ~[app.jar:?]
    at io.debezium.embedded.ConvertingEngineBuilder$ConvertingChangeConsumer.handleBatch(ConvertingEngineBuilder.java:75) ~[app.jar:?]
    at io.debezium.embedded.EmbeddedEngine.pollRecords(EmbeddedEngine.java:731) ~[app.jar:?]
    at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:470) ~[app.jar:?]
    at io.debezium.embedded.ConvertingEngineBuilder$1.run(ConvertingEngineBuilder.java:163) ~[app.jar:?]
    at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.lambda$setupDebeziumEventCapture$0(DebeziumChangeEventCapture.java:290) ~[app.jar:?]
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
    at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
    at java.lang.Thread.run(Thread.java:840) [?:?]

2026-02-03 20:02:12.214 ERROR - Error persisting records to ClickHousejava.lang.RuntimeException: java.sql.BatchUpdateException: Code: 60. DB::Exception: Table system.table_name does not exist. Maybe you meant different_database.table_name?. (UNKNOWN_TABLE) (version 25.8.2.29 (official build))

2026-02-03 20:02:12.715 INFO - ****** Thread: Sink connector Debezium Event Thread Batch Size: 15 ******
2026-02-03 20:02:12.715 INFO - *** INSERT QUERY for Database(database_name) ***: insert into table_name(id,_version UInt64,_deleted UInt8')
2026-02-03 20:02:12.716 ERROR - Error getting prepared statement, retry attempt (0/2) failed
java.sql.SQLException: Connection is closed
at com.zaxxer.hikari.pool.ProxyConnection$ClosedConnection.lambda$getClosedConnection$0(ProxyConnection.java:563) ~[app.jar:?]
at jdk.proxy3.$Proxy28.prepareStatement(Unknown Source) ~[?:?]
at com.zaxxer.hikari.pool.ProxyConnection.prepareStatement(ProxyConnection.java:328) ~[app.jar:?]
at com.zaxxer.hikari.pool.HikariProxyConnection.prepareStatement(HikariProxyConnection.java) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.db.DBMetadata.getPreparedStatement(DBMetadata.java:749) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.lambda$executePreparedStatement$0(PreparedStatementExecutor.java:202) ~[app.jar:?]
at java.lang.Iterable.forEach(Iterable.java:75) ~[?:?]
at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.executePreparedStatement(PreparedStatementExecutor.java:196) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.addToPreparedStatementBatch(PreparedStatementExecutor.java:151) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.flushRecordsToClickHouse(ClickHouseBatchWriter.java:453) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.processRecordsByTopic(ClickHouseBatchWriter.java:404) ~[app.jar:?]
at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.persistRecords(ClickHouseBatchWriter.java:221) ~[app.jar:?]
at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.appendToRecords(DebeziumChangeEventCapture.java:909) ~[app.jar:?]
at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture$1.handleBatch(DebeziumChangeEventCapture.java:209) ~[app.jar:?]
at io.debezium.embedded.ConvertingEngineBuilder$ConvertingChangeConsumer.handleBatch(ConvertingEngineBuilder.java:75) ~[app.jar:?]
at io.debezium.embedded.EmbeddedEngine.pollRecords(EmbeddedEngine.java:731) ~[app.jar:?]
at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:470) ~[app.jar:?]
at io.debezium.embedded.ConvertingEngineBuilder$1.run(ConvertingEngineBuilder.java:163) ~[app.jar:?]
at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.lambda$setupDebeziumEventCapture$0(DebeziumChangeEventCapture.java:290) ~[app.jar:?]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
at java.lang.Thread.run(Thread.java:840) [?:?]
2026-02-03 20:02:12.723 ERROR - ******* ERROR inserting Batch Database(database), Table(table_name) *****************
java.sql.BatchUpdateException: Code: 60. DB::Exception: Table system.table_name does not exist. Maybe you meant different_database.table_name?. (UNKNOWN_TABLE) (version 25.8.2.29 (official build))

    at com.clickhouse.jdbc.SqlExceptionUtils.batchUpdateError(SqlExceptionUtils.java:107) ~[app.jar:?]
    at com.clickhouse.jdbc.internal.InputBasedPreparedStatement.executeAny(InputBasedPreparedStatement.java:154) ~[app.jar:?]
    at com.clickhouse.jdbc.internal.AbstractPreparedStatement.executeLargeBatch(AbstractPreparedStatement.java:85) ~[app.jar:?]
    at com.clickhouse.jdbc.internal.ClickHouseStatementImpl.executeBatch(ClickHouseStatementImpl.java:819) ~[app.jar:?]
    at com.zaxxer.hikari.pool.ProxyStatement.executeBatch(ProxyStatement.java:127) ~[app.jar:?]
    at com.zaxxer.hikari.pool.HikariProxyPreparedStatement.executeBatch(HikariProxyPreparedStatement.java) ~[app.jar:?]
    at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.lambda$executePreparedStatement$0(PreparedStatementExecutor.java:258) ~[app.jar:?]
    at java.lang.Iterable.forEach(Iterable.java:75) ~[?:?]
    at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.executePreparedStatement(PreparedStatementExecutor.java:196) ~[app.jar:?]
    at com.altinity.clickhouse.sink.connector.db.batch.PreparedStatementExecutor.addToPreparedStatementBatch(PreparedStatementExecutor.java:151) ~[app.jar:?]
    at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.flushRecordsToClickHouse(ClickHouseBatchWriter.java:453) ~[app.jar:?]
    at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.processRecordsByTopic(ClickHouseBatchWriter.java:404) ~[app.jar:?]
    at com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchWriter.persistRecords(ClickHouseBatchWriter.java:221) ~[app.jar:?]
    at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.appendToRecords(DebeziumChangeEventCapture.java:909) ~[app.jar:?]
    at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture$1.handleBatch(DebeziumChangeEventCapture.java:209) ~[app.jar:?]
    at io.debezium.embedded.ConvertingEngineBuilder$ConvertingChangeConsumer.handleBatch(ConvertingEngineBuilder.java:75) ~[app.jar:?]
    at io.debezium.embedded.EmbeddedEngine.pollRecords(EmbeddedEngine.java:731) ~[app.jar:?]
    at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:470) ~[app.jar:?]
    at io.debezium.embedded.ConvertingEngineBuilder$1.run(ConvertingEngineBuilder.java:163) ~[app.jar:?]
    at com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture.lambda$setupDebeziumEventCapture$0(DebeziumChangeEventCapture.java:290) ~[app.jar:?]
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
    at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
    at java.lang.Thread.run(Thread.java:840) [?:?]

2026-02-03 20:02:12.723 ERROR - Error persisting records to ClickHousejava.lang.RuntimeException: java.sql.BatchUpdateException: Code: 60. DB::Exception: Table system.table_name does not exist. Maybe you meant different_database.table_name?. (UNKNOWN_TABLE) (version 25.8.2.29 (official build))`

And then it carries on as normal just having skipped those, at this point clickhouse was reachable and working normally as evidenced by its suggestion of other databases

I wonder if there is any way to work around this, we have just switched to using ClickHouse from redshift and this is going to cause some big issues as it seems to happen more often than I'd initially thought. If it was just when a node restarted I couuld have stopped the sink connectors when that was about to happen but now it looks like it could be pretty unpredictable.

Is there potentially something I can do to mitigate it with connection pool settings? I may be wrong but given it does look conneciton pool related, I'll switch one of my connectors to connection.pool.disable: 'true' and see if that has any impact.

Thanks

[aadant]

Yes we do not use the connection pool in production. Please disable it !

[Bens-ct]

Can confirm that since disabling connection pooling everything seems a lot happier and no evidence of any further data loss.