ClickHouse Kafka Connect Sink

### Configuration options[​](#configuration-options "Direct link to Configuration options")

To connect the ClickHouse Sink to the ClickHouse server, you need to provide:

- connection details: hostname (**required**) and port (optional)
- user credentials: password (**required**) and username (optional)
- connector class: `com.clickhouse.kafka.connect.ClickHouseSinkConnector` (**required**)
- topics or topics.regex: the Kafka topics to poll - topic names must match table names (**required**)
- key and value converters: set based on the type of data on your topic. Required if not already defined in worker config.

The full table of configuration options:

|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |
| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |
| Property Name Description Default Value|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `hostname` (Required) The hostname or IP address of the server N/A|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `port` The ClickHouse port - default is 8443 (for HTTPS in the cloud), but for HTTP (the default for self-hosted) it should be 8123 `8443`| `ssl` Enable ssl connection to ClickHouse `true`| `jdbcConnectionProperties` Connection properties when connecting to Clickhouse. Must start with `?` and joined by `&` between `param=value` `""`| `username` ClickHouse database username `default`| `password` (Required) ClickHouse database password N/A|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `database` ClickHouse database name `default`| `connector.class` (Required) Connector Class(explicit set and keep as the default value) `"com.clickhouse.kafka.connect.ClickHouseSinkConnector"`| `tasks.max` The number of Connector Tasks `"1"`| `errors.retry.timeout` Kafka Connect max retry duction in milliseconds. `0` for no retries. `-1` for infinite retries. Recommended value is more then "10000" ms (10 seconds) Timeout `"0"`| `exactlyOnce` Exactly Once Enabled `"false"`| `topics` (Required) The Kafka topics to poll - topic names must match table names `""`| `key.converter` (Required\* - See Description) Set according to the types of your keys. Required here if you're passing keys (and not defined in worker config). `"org.apache.kafka.connect.storage.StringConverter"`| `value.converter` (Required\* - See Description) Set based on the type of data on your topic. Supported: - JSON, String, Avro or Protobuf formats. Required here if not defined in worker config. `"org.apache.kafka.connect.json.JsonConverter"`| `value.converter.schemas.enable` Connector Value Converter Schema Support `"false"`| `errors.tolerance` Connector Error Tolerance. Supported: none, all `"none"`| `errors.deadletterqueue.topic.name` If set (with errors.tolerance=all), a DLQ will be used for failed batches (see [Troubleshooting](#troubleshooting)) `""`| `errors.deadletterqueue.context.headers.enable` Adds additional headers for the DLQ `""`| `clickhouseSettings` Comma-separated list of ClickHouse settings (e.g. "insert\_quorum=2, etc...") `""`| `topic2TableMap` Comma-separated list that maps topic names to table names (e.g. "topic1=table1, topic2=table2, etc...") `""`| `tableRefreshInterval` Time (in seconds) to refresh the table definition cache `0`| `keeperOnCluster` Allows configuration of ON CLUSTER parameter for self-hosted instances (e.g. `ON CLUSTER clusterNameInConfigFileDefinition`) for exactly-once connect\_state table (see [Distributed DDL Queries](/docs/sql-reference/distributed-ddl) `""`| `bypassRowBinary` Allows disabling use of RowBinary and RowBinaryWithDefaults for Schema-based data (Avro, Protobuf, etc.) - should only be used when data will have missing columns, and Nullable/Default are unacceptable `"false"`| `dateTimeFormats` Date time formats for parsing DateTime64 schema fields, separated by `;` (e.g. `someDateField=yyyy-MM-dd HH:mm:ss.SSSSSSSSS;someOtherDateField=yyyy-MM-dd HH:mm:ss`). `""`| `tolerateStateMismatch` Allows the connector to drop records "earlier" than the current offset stored AFTER\_PROCESSING (e.g. if offset 5 is sent, and offset 250 was the last recorded offset). Should be used to fix ingestion after failure and should be reverted back to `"false"` once done. `"false"`| `ignorePartitionsWhenBatching` Will ignore partition when collecting messages for insert (though only if `exactlyOnce` is `false`). Performance Note: The more connector tasks, the fewer kafka partitions assigned per task - this can mean diminishing returns. `"false"`| `bufferCount` (since v1.3.6) Number of records to buffer in memory before flushing to ClickHouse. `0` disables internal buffering. Buffering is not supported with `exactlyOnce=true`. `"0"`| `bufferFlushTime` (since v1.3.6) Maximum time in milliseconds to buffer records before flush when `exactlyOnce=false`. `0` disables time-based flushing. Default value is `0`. Only required for time-base threshold. Only effective when `bufferCount > 0`. `"0"`| `reportInsertedOffsets` (since v1.3.6) Enables returning only successfully inserted offsets from `preCommit` (instead of `currentOffsets`) when `exactlyOnce=false`. This does not apply when `ignorePartitionsWhenBatching=true`, where `currentOffsets` are still returned. `"false"` | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | |

### Target tables[​](#target-tables "Direct link to Target tables")