> ## Documentation Index > Fetch the complete documentation index at: https://docs.automq.com/llms.txt > Use this file to discover all available pages before exploring further. # ClickHouse Sink Connector > Configure and operate the ClickHouse Sink Connector in AutoMQ Connect, including table mapping, delivery behavior, security, monitoring, and troubleshooting. ## Overview ClickHouse Sink Connector consumes records from Kafka Topics and writes record values to ClickHouse tables for real-time analytics pipelines involving behavioral events, application logs, and business transactions. Typically, each Topic corresponds to a table with the same name, but it can also be explicitly mapped to an existing business table. Fields in structured records and JSON objects are matched to target columns by name; CSV/TSV strings are written in the target table's input column order. The Connector uses the Connect data model produced by the value Converter to process structured records with schemas, schemaless JSON objects, or string records. Writes are append-based; ordinary Kafka messages are not automatically converted into key-based updates or deletes. Optional Debezium CDC processing converts change events into versioned rows that represent current state in conjunction with the target table engine. ## Prerequisites * The ClickHouse server must be version 23.3 or later. * Create the target database and tables in advance. Column names, types, and default-value rules for missing fields must match the input records. * The ClickHouse account must be able to read target table metadata and execute INSERT. * When automatic column addition is enabled, the account also needs ALTER permission on the target tables. * When enabling `exactlyOnce`, configure an available KeeperMap state store with permissions to create its table and read and write state, and confirm that the target table engine and deduplication window meet replay deduplication requirements. ## License Licensed under Apache License 2.0. ## Quick Start Prepare a Connect Cluster, Kafka Topic, ClickHouse database, and target table in advance, and confirm network connectivity and access permissions. For resource preparation and Connector management, see [Manage Connectors](../manage-connectors). The following configuration accepts schemaless JSON objects and writes over HTTPS to a table with the same name as the Topic. ```properties theme={null} connector.class=com.clickhouse.kafka.connect.ClickHouseSinkConnector topics= hostname= database= username= password= ssl=true client_version=V2 value.converter=org.apache.kafka.connect.json.JsonConverter value.converter.schemas.enable=false ``` Replace the environment placeholders. `hostname` must not include a protocol or port. The default HTTPS port is `8443`; add `port` if the server uses another port. For HTTP deployments, set both `ssl=false` and the corresponding HTTP port, such as `8123`. Changing the protocol does not automatically change the port. Provide the password through secure credential management rather than committing actual credentials to a configuration repository. For example, if the input JSON objects contain an integer `event_id` and a string `event_type`, execute the following table creation statement in the ClickHouse database above, replacing `` with exactly the same name as ``. ```sql theme={null} CREATE TABLE ( event_id Int64, event_type String ) ENGINE = MergeTree ORDER BY event_id; ``` Use ordinary JSON objects for Kafka message values, such as `{"event_id":1,"event_type":"page_view"}`, without the Connect `schema` / `payload` wrapper. This configuration does not enable persistent-state deduplication. Recovery or replay can produce duplicate rows, and ordinary `MergeTree` does not automatically deduplicate based on `ORDER BY`. ## Configuration The defaults below are the values declared in the configuration definitions. Where omitting a parameter changes its actual behavior, this is explained separately in the notes. List values generally use commas as separators; the multi-field format for `dateTimeFormats` uses semicolons. ### Connector Identity and Input Subscription #### `connector.class` Specifies the Connector implementation class. * **Type**: `string` * **Default**: None * **Importance**: High * **Required**: Yes * **Valid Values / Notes**: Use `com.clickhouse.kafka.connect.ClickHouseSinkConnector`. #### `tasks.max` Maximum number of Tasks that can be created. * **Type**: `int` * **Default**: `1` * **Importance**: High * **Valid Values / Notes**: At least `1`. Effective parallelism is constrained by the number and assignment of input partitions. Adding Tasks does not guarantee a linear increase in throughput. #### `topics` Specifies the list of Topics to consume. * **Type**: `list` * **Default**: Empty list * **Importance**: High * **Required**: Either this or `topics.regex` * **Valid Values / Notes**: Comma-separated. Exactly one of this setting and `topics.regex` must have a nonempty value. The list must not include this Connector's own DLQ Topic. #### `topics.regex` Subscribes to Topics using a regular expression. * **Type**: `string` * **Default**: Empty string * **Importance**: High * **Required**: Either this or `topics` * **Valid Values / Notes**: Use a Java regular expression that does not match this Connector's own DLQ Topic. Prepare target tables for newly matched Topics as well. ### ClickHouse Connection and Authentication #### `hostname` ClickHouse hostname. * **Type**: `string` * **Default**: None * **Importance**: High * **Required**: Yes * **Valid Values / Notes**: Nonempty, without a protocol or port. #### `port` Port of the ClickHouse HTTP or HTTPS interface. * **Type**: `int` * **Default**: `8443` * **Importance**: High * **Valid Values / Notes**: `1` to `65535`. This is not the native TCP interface port and does not change automatically with `ssl`. #### `username` ClickHouse username. * **Type**: `string` * **Default**: Empty string * **Importance**: Low * **Valid Values / Notes**: If omitted, the connection implementation falls back to `default`. Specify an account explicitly to avoid relying on the different handling of omitted and empty values. #### `password` Password for the ClickHouse account. * **Type**: `password` * **Default**: Empty string * **Importance**: Low * **Valid Values / Notes**: Leading and trailing spaces are removed when read. Do not use passwords that depend on these spaces, and do not include actual credentials in logs or public examples. #### `ssl` Selects whether to connect over HTTPS. * **Type**: `boolean` * **Default**: `true` * **Importance**: Low * **Valid Values / Notes**: `true` or `false`. If omitted, the connection implementation may fall back to HTTP. Set it explicitly and match it to `port`. #### `client_version` Selects the ClickHouse client implementation. * **Type**: `string` * **Default**: Empty string * **Importance**: Low * **Valid Values / Notes**: Explicitly specify `V1` or `V2`. Only the exact value `V1` selects the older client; other values, including an explicitly empty string, select V2. Omitting the setting falls back to V1. #### `jdbcConnectionProperties` Adds query properties to the connection URL. * **Type**: `string` * **Default**: Empty string * **Importance**: Low * **Valid Values / Notes**: Use the `key=value&key=value` format; the leading `?` is optional. The connection URL may be logged, so do not pass passwords or tokens through this setting. #### `clickhouseSettings` Specifies ClickHouse server settings for write requests. * **Type**: `list` * **Default**: Empty list * **Importance**: Low * **Valid Values / Notes**: Comma-separated `key=value` entries, each containing only one `=`. The connection implementation adds each of `input_format_skip_unknown_fields=1`, `wait_end_of_query=1`, `async_insert=0`, and `send_progress_in_http_headers=1` that the user has not specified, even when this setting already contains other entries. User values are not overridden. These are not the declared defaults for this setting. Do not override the Connector-managed `insert_deduplication_token`. Asynchronous inserts must wait for actual insertion; `wait_for_async_insert=0` does not provide persistence acknowledgment. #### `ssl_socket_sni` Overrides the SNI hostname in the TLS handshake. * **Type**: `string` * **Default**: Empty string * **Importance**: Low * **Valid Values / Notes**: Use only for special TLS routing requirements. With the V1 client, a nonempty value also disables certificate and hostname verification; this is not an SNI adjustment without security implications. ### Proxy Connection #### `proxyType` Specifies the proxy type for ClickHouse connections. * **Type**: `string` * **Default**: Empty string * **Importance**: Low * **Valid Values / Notes**: Recommended values are `IGNORE`, `DIRECT`, `HTTP`, and `SOCKS`. An empty string or unrecognized value ignores the proxy. V2 constructs an HTTP proxy for every non-`IGNORE` type and does not preserve the original SOCKS or DIRECT semantics. #### `proxyHost` Proxy server hostname. * **Type**: `string` * **Default**: Empty string * **Importance**: Low * **Valid Values / Notes**: Used only for non-`IGNORE` proxy types. Specify the actual hostname when enabling a proxy. #### `proxyPort` Proxy server port. * **Type**: `int` * **Default**: `-1` * **Importance**: Low * **Valid Values / Notes**: Explicitly specify a valid network port when enabling a proxy. The default does not represent a usable port. ### Database and Table Routing #### `database` Default target database. * **Type**: `string` * **Default**: `default` * **Importance**: Low * **Valid Values / Notes**: The database must already exist. When Topic splitting is enabled, the database portion of the Topic name can override this value. #### `topic2TableMap` Maps Topics to target tables. * **Type**: `list` * **Default**: Empty list * **Importance**: Low * **Valid Values / Notes**: Comma-separated `topic=table` entries. Leading and trailing spaces are removed from keys and values; later entries take precedence for duplicate keys. Unmapped Topics use tables with the same name. A mapping value is a table name, not cross-database `database.table` routing syntax, and does not create a table. #### `enableDbTopicSplit` Splits the Topic name into database and table-routing names. * **Type**: `boolean` * **Default**: `false` * **Importance**: Low * **Valid Values / Notes**: Explicitly set `dbTopicSplitChar` when enabling this option. The database is overridden only when the name splits into exactly two parts; `topic2TableMap` is then looked up using the resulting Topic portion. Otherwise, the original database and Topic name are retained. #### `dbTopicSplitChar` Separator between the database and Topic portions. * **Type**: `string` * **Default**: Empty string * **Importance**: Low * **Valid Values / Notes**: Used only with `enableDbTopicSplit=true`. Recommended values are `_`, `-`, or `.`; the value is treated as a literal separator, not a regular expression. #### `suppressTableExistenceException` Suppresses exceptions when a target table does not exist. * **Type**: `boolean` * **Default**: `false` * **Importance**: Low * **Valid Values / Notes**: When enabled, records for missing tables may be ignored while Offsets continue to advance. This neither creates tables nor waits for them to appear. Keep it disabled in pipelines that require complete delivery. ### Value Formats and Field Processing #### `value.converter` Deserializes Kafka message values into Connect data. * **Type**: `class` * **Default**: `null` * **Importance**: Low * **Valid Values / Notes**: `null` uses the Worker's Converter. The class must implement Converter. Struct values with schemas, schemaless Map values, and String values use different write paths. Avro or Protobuf requires the corresponding Converter and dependencies to be supplied separately. #### `customInsertFormat` Custom format option for string input. * **Type**: `boolean` * **Default**: `false` * **Importance**: Low * **Valid Values / Notes**: This setting affects recommended format values in the configuration interface, rather than independently controlling the actual write path. The actual string format is determined by `insertFormat`. #### `insertFormat` Specifies the insertion format for String values. * **Type**: `string` * **Default**: `none` * **Importance**: Low * **Valid Values / Notes**: Explicitly select `CSV`, `TSV`, or `JSON` for string input; JSON uses JSONEachRow. `NONE` cannot write strings. Do not rely on unknown values falling back to JSON. CSV/TSV field order must match the target table's input order. #### `bypassRowBinary` Switches records with schemas from binary writes to JSONEachRow. * **Type**: `boolean` * **Default**: `false` * **Importance**: Low * **Valid Values / Notes**: Does not change the Converter or automatically fill in fields. The default schema-based path uses RowBinary or RowBinaryWithDefaults. #### `dateTimeFormats` Specifies parsing patterns for date-time fields. * **Type**: `list` * **Default**: Empty list * **Importance**: Low * **Valid Values / Notes**: The actual input uses semicolons to separate multiple `field=pattern` entries, such as `created_at=yyyy-MM-dd HH:mm:ss;updated_at=yyyy-MM-dd HH:mm:ss`. Patterns follow Java DateTimeFormatter syntax. Invalid patterns cause configuration reading to fail. #### `bypassFieldCleanup` Retains top-level fields that do not belong to the target table in the JSON write path. * **Type**: `boolean` * **Default**: `false` * **Importance**: Low * **Valid Values / Notes**: When disabled, extra top-level fields are removed according to the target columns. When enabled, server-side `input_format_skip_unknown_fields` and type checks still apply. #### `debeziumCDCEnabled` Converts Debezium Envelopes into versioned target rows. * **Type**: `boolean` * **Default**: `false` * **Importance**: Medium * **Valid Values / Notes**: Requires Struct values whose schema names end in `.Envelope`. Use `ReplacingMergeTree(_version, is_deleted)` for the target table and design the sorting key around the business primary key. Delete events (`op=d`) use `before` and write `is_deleted=1`; create, snapshot read, and update events (`op=c/r/u`) use `after` and write `is_deleted=0`. Both add `_version`. Truncate events (`op=t`) cannot clear the target table. Delete markers do not execute SQL DELETE or guarantee immediate physical deletion. ### Table Structure and Automatic Column Addition #### `tableRefreshInterval` Periodically refreshes target table metadata. * **Type**: `long` * **Default**: `0` * **Importance**: Low * **Valid Values / Notes**: `0` to `600`, in seconds. `0` disables periodic refresh. Existing tables are primarily reread when the column count increases; tracking column type changes, removals, or renames is not guaranteed. #### `bypassSchemaValidation` Skips schema alignment checks between records and the target table. * **Type**: `boolean` * **Default**: `false` * **Importance**: Low * **Valid Values / Notes**: Skips only the Connector's checks, not serialization constraints or ClickHouse type validation. Do not use it to fix incompatible data. #### `auto.evolve` Adds missing columns to the target table based on the input schema. * **Type**: `boolean` * **Default**: `false` * **Importance**: Medium * **Valid Values / Notes**: Intended for structured records with actual Connect schemas. Do not rely on schemaless JSON or String values to infer business types automatically. Requires ALTER permission. Does not create business tables, change existing column types, or remove columns. Fields are checked using the last record in each write batch; keep schema evolution ordered consistently within batches. #### `auto.evolve.ddl.refresh.retries` Number of retries to wait for new column metadata to become visible after automatic column addition. * **Type**: `int` * **Default**: `3` * **Importance**: Low * **Valid Values / Notes**: At least `0`. Controls only metadata refresh after DDL, not a unified retry count for write requests or framework errors. #### `auto.evolve.struct.to.json` Infers Connect STRUCT values as ClickHouse JSON columns during automatic column addition. * **Type**: `boolean` * **Default**: `false` * **Importance**: Medium * **Valid Values / Notes**: Applies only to automatic evolution type mapping. The server must support the JSON type and related settings. For binary writes to JSON columns, also set `input_format_binary_read_json_as_string=1` or `true` in `clickhouseSettings`. #### `clusterName` Specifies the ClickHouse cluster for automatic evolution DDL. * **Type**: `string` * **Default**: Empty string * **Importance**: Medium * **Valid Values / Notes**: A nonempty value adds `ON CLUSTER`; an empty string runs local DDL only. Requires a valid cluster and DDL permissions. Specify only a trusted cluster name, not SQL fragments. Independent of `keeperOnCluster`. ### Batching and Internal Buffering #### `ignorePartitionsWhenBatching` Combines records from different partitions of the same Topic for writes. * **Type**: `boolean` * **Default**: `false` * **Importance**: Low * **Valid Values / Notes**: Used with `exactlyOnce=false`. With `exactlyOnce=true` and no buffering, this option is not used; with buffering enabled, setting it to `true` causes Task startup to fail. Does not provide global ordering across partitions. #### `bufferCount` Record-count threshold for accumulation across multiple consumption calls. * **Type**: `int` * **Default**: `0` * **Importance**: Low * **Valid Values / Notes**: At least `0`. `0` disables internal buffering. In non-exactly-once mode, reaching the total count threshold triggers a flush; the threshold is not a maximum batch size. In exactly-once mode, complete, fixed-size batches are sent per Topic/partition, requiring `bufferFlushTime=0` and `ignorePartitionsWhenBatching=false`. Remaining records below the count continue to wait. #### `bufferFlushTime` Time threshold for triggering an internal buffer flush. * **Type**: `long` * **Default**: `0` * **Importance**: Low * **Valid Values / Notes**: At least `0`, in milliseconds. `0` disables time-based triggering. Effective only with `bufferCount>0`. The time condition is checked during Task processing calls; it is not a promise of independently scheduled flushing. Must remain `0` in exactly-once buffering mode. ### State, Deduplication, and Offsets #### `exactlyOnce` Enables the KeeperMap-based persistent-state and batch-deduplication path. * **Type**: `boolean` * **Default**: `false` * **Importance**: Low * **Valid Values / Notes**: Records state before and after processing per Topic/partition. Replay deduplication depends on an available KeeperMap, target engine deduplication support and a valid window, stable data and batch boundaries, and reliable insertion acknowledgment. Enabling this option alone does not establish an unconditional exactly-once guarantee. Data writes and state writes are not a single transaction; cross-table or cross-partition atomicity is not provided. #### `zkPath` KeeperMap state storage path. * **Type**: `string` * **Default**: `/kafka-connect` * **Importance**: Low * **Valid Values / Notes**: Must not be blank and must start with `/`. Used for state storage with `exactlyOnce=true`; plan state isolation. #### `zkDatabase` KeeperMap state table name. * **Type**: `string` * **Default**: `connect_state` * **Importance**: Low * **Valid Values / Notes**: Despite the name Database, this is a state table identifier, not a separate database creation setting. Use a valid, trusted table name. #### `keeperOnCluster` Specifies the ClickHouse cluster used for the Keeper state table. * **Type**: `string` * **Default**: Empty string * **Importance**: Low * **Valid Values / Notes**: Used for persistent-state scenarios in self-hosted clusters. Not equivalent to `clusterName` for business table DDL. #### `tolerateStateMismatch` Tolerates certain mismatches between batch ranges and existing state. * **Type**: `boolean` * **Default**: `false` * **Importance**: Low * **Valid Values / Notes**: Some records behind existing state may be skipped. This is not a switch for reinserting data after an Offset rollback and does not guarantee correction of all state errors. Evaluate data completeness before enabling it. #### `reportInsertedOffsets` Makes the direct-write strategy report Offsets for processed write ranges before committing. * **Type**: `boolean` * **Default**: `false` * **Importance**: Low * **Valid Values / Notes**: The internal buffering strategy maintains flushed ranges itself and is not controlled by this setting. Non-exactly-once cross-partition batching still uses the Worker's current Offsets. Error tolerance and missing-table suppression can make "processed" differ from "actually inserted"; this setting is not proof of successful insertion for each row. ### Timeouts and Error Handling #### `timeoutSeconds` Timeout for ClickHouse driver operations. * **Type**: `int` * **Default**: `30` * **Importance**: Low * **Valid Values / Notes**: `0` to `600`, in seconds. Configured separately from the insert-response wait limit `clickhouseClientInsertTimeoutMs`. #### `retryCount` Retry count for ClickHouse client auxiliary operations. * **Type**: `int` * **Default**: `3` * **Importance**: Low * **Valid Values / Notes**: `3` to `10`. Affects auxiliary operations such as connection probes and queries, not a unified maximum retry count for all INSERT operations. #### `clickhouseClientInsertTimeoutMs` Maximum wait for a single insert response with the V2 client. * **Type**: `long` * **Default**: `240000` * **Importance**: Low * **Valid Values / Notes**: At least `100`, in milliseconds. V1 inserts do not use this separate wait limit. It should be shorter than the consumer's effective `max.poll.interval.ms`, with headroom for processing and retries. A timeout or canceled wait does not prove that the server did not accept the data. #### `consumer.override.max.poll.interval.ms` Overrides the maximum interval between consecutive polls by the Sink consumer. * **Type**: `int` * **Default**: No independent Connector-level default; the underlying client default is `300000` * **Importance**: Medium * **Valid Values / Notes**: At least `1`, in milliseconds. Worker consumer configuration can change the effective value; whether an override is allowed depends on the Worker's client override policy. Set it longer than the insert wait time, with headroom, to avoid rebalances during slow writes. #### `errors.retry.timeout` Total retry duration for retriable Kafka Connect framework errors. * **Type**: `long` * **Default**: `0` * **Importance**: Medium * **Valid Values / Notes**: In milliseconds. `0` disables retries; `-1` retries indefinitely. Applies only to the corresponding framework retriable-processing paths, not a guarantee of automatic retries for all ClickHouse write errors. Does not replace `retryCount`. #### `errors.tolerance` Controls the error tolerance policy. * **Type**: `string` * **Default**: `none` * **Importance**: Low * **Valid Values / Notes**: `none` or `all`. The framework setting with the same name has Medium importance. `all` can skip tolerated errors and continue advancing; it is not a retry switch. Plugin write failures may be reported for an entire group and must not be treated as precise, per-row isolation of bad records. Without an available DLQ, skipped data is not necessarily durably stored. #### `errors.deadletterqueue.topic.name` Specifies a dead-letter queue Topic for error reporting. * **Type**: `string` * **Default**: Empty string * **Importance**: Medium * **Valid Values / Notes**: An empty string disables DLQ recording. Requires an error tolerance policy and the Worker's error reporting mechanism; the Topic must not be subscribed to by this Connector. Configuring a Topic name does not mean reports have been persisted successfully. Check actual DLQ writes and their errors. ## Best Practices ### Writing Multiple Business Event Streams to Existing Analytics Tables Applicable business scenario: Order events and payment events use different Kafka Topics, while ClickHouse already has analytics tables with established names. You want to retain those table names without requiring the Topics and tables to share names. Configuration example: Override `topics` in the Quick Start configuration and add the following mapping. Both tables are in the database specified in Quick Start and must be created in advance to match their respective event structures. ```properties theme={null} topics=, topic2TableMap==,= ``` Key considerations: Mapping only selects the write destination. It neither merges the field structures of the two Topics nor creates tables automatically. This suits scenarios where each stream maintains its own data model. Unmapped Topics still use tables with the same name, so prepare the corresponding target tables before expanding the subscription scope. ### Accumulating Sparse Events to Reduce Small-Batch Writes Applicable business scenario: Behavioral or application events arrive continuously, but each consumption call returns only a few records. You want to accumulate them before writing to ClickHouse to reduce frequent small-batch inserts, while giving low-volume records opportunities to flush during subsequent processing calls. Configuration example: Add the following settings to the Quick Start configuration. This approach stays in non-exactly-once mode and must not be mixed with the persistent-state deduplication buffering approach. ```properties theme={null} bufferCount=1000 bufferFlushTime=1000 ``` Key considerations: Flushing occurs when the count threshold is reached or a Task processing call checks that the time threshold has elapsed. `1000` milliseconds is not a strict upper bound on insertion latency, and no independent background timer guarantees insertion at that time. A single consumption call can return more records than the count threshold; final writes are still grouped by Topic/partition. Adjust the count and time based on event size, Worker memory, and acceptable latency, and check actual visibility latency. This approach does not eliminate replay duplicates and does not guarantee exactly-once recovery. ## Monitoring ### What to Monitor Monitor Kafka Connect cluster health, Connector and Task running and failed states, consumption and write throughput, consumer backlog and end-to-end latency, Offset commit progress and failures, changes in errors and retries, and Worker JVM heap, GC, and thread signals. When error tolerance and DLQ reporting are enabled, also monitor DLQ activity and reporting failures so that a continuously running Task is not mistaken for complete insertion of all data. ### Import the Grafana Dashboard Download the shared [Kafka Connect Grafana Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json). Ensure that Kafka Connect metrics are collected into a Prometheus-compatible data source and that labels such as cluster, Worker, Connector, and Task match the dashboard queries. Import the JSON file into Grafana and select the corresponding data source. ## Limitations * The Connector does not automatically create business databases or tables; automatic evolution only adds missing columns. * Kafka tombstones do not trigger deletions in target tables, and ordinary writes do not execute key-based SQL UPDATE or DELETE. * Input values must be converted to supported Struct, Map, String, or null values. Direct writes of arbitrary Connect primitive scalars are not supported. * Global ordering across partitions or Tasks is not provided, and target query results are not guaranteed to follow insertion order. * Data writes and Keeper state writes are not a single transaction; cross-table and cross-partition atomic transactions are not provided. * Exactly-once internal buffering does not support time-triggered flushing or cross-partition batching. Remaining records below the count threshold are not forcibly written on commit, close, or stop. * Automatic column addition does not change existing column types or remove columns, and periodic metadata refresh is not an automatic adaptation mechanism for arbitrary table structure changes. ## FAQ ### A Task Cannot Connect to ClickHouse or Cannot Find a Table at Startup Check whether `hostname` incorrectly includes a protocol, whether `port` is an HTTP/HTTPS port, and whether `ssl` matches the server. Specify `ssl` and `client_version` explicitly to avoid connection effects from fallback behavior when parameters are omitted. Confirm that the server meets the version requirement, the account can read table metadata, and the target table already exists under `database`. When using mappings, check the Topic and table names. Do not enable `suppressTableExistenceException` to hide missing-table issues; create or correct the target table first. ### JSON Input Produces Format or Field Type Errors Determine whether messages are ordinary JSON objects or Connect JSON with a `schema` / `payload` wrapper. For ordinary objects, use the Quick Start JsonConverter and disable schema-wrapper parsing. Check target column names, types, and DEFAULT or Nullable definitions for missing fields. StringConverter input also requires an explicit `insertFormat`. Do not simply enable `bypassSchemaValidation`; it does not make ClickHouse accept incompatible types. ### Small Amounts of Data Remain Invisible After Buffering Is Enabled Check whether the record count has reached `bufferCount`. In non-exactly-once mode, you can set a nonzero `bufferFlushTime`, but the time condition is checked only during subsequent Task processing calls. If the business requires data to become visible promptly, disable internal buffering. Exactly-once buffering writes only complete, fixed-size batches and does not support time-based flushing. Reduce the count threshold or disable buffering rather than adding conflicting time-trigger settings. ### Duplicate Rows Appear After a Restart A failure window exists between successful insertion and Kafka Offset commit, so replay can insert data again. An ordinary MergeTree sorting key does not provide a uniqueness constraint. First check commit failures, rebalances, and timeouts, and determine whether the server has already accepted the request. If persistent-state deduplication is needed, evaluate KeeperMap, target engine deduplication conditions and windows, stable batches, and insertion acknowledgment. Do not assume duplicates are impossible merely by setting `exactlyOnce=true` or a custom deduplication token. ### State Mismatches Appear After Rolling Back Offsets Persistent state still retains processed ranges; an external Offset rollback does not automatically coordinate KeeperMap. Stop writes first, compare the required replay range with the corresponding Topic/partition state, and then plan state cleanup or a separate replay approach while evaluating duplicate risks. `tolerateStateMismatch=true` may skip older ranges; it is not a repair switch for reimporting historical data. ### A Task Keeps Running but Data Is Missing or the DLQ Has No Records Check whether `errors.tolerance=all` or missing-table exception suppression is enabled. Either can advance records as processed even when they were not inserted. Confirm that the DLQ Topic is outside the subscription scope and the Worker's error reporting mechanism is available. Check reporting permissions, send errors, and actual messages; configuring a Topic name does not guarantee successful report persistence. When complete delivery is required, keep failures visible, fix the root cause, and arrange replay using confirmed Offset ranges.