ClickHouse Kafka Connect Sink

- (1) - JSON is supported only when ClickHouse settings has `input_format_binary_read_json_as_string=1`. This works only for RowBinary format family and the setting affects all columns in the insert request so they all should be a string. Connector will convert STRUCT to a JSON string in this case.
- (2) - When struct has unions like `oneof` then converter should be configured to NOT add prefix/suffix to a field names. There is `generate.index.for.unions=false` [setting for `ProtobufConverter`](https://docs.confluent.io/platform/current/schema-registry/connect.html#protobuf).

**Without a schema declared:**

A record is converted into JSON and sent to ClickHouse as a value in [JSONEachRow](/docs/interfaces/formats/JSONEachRow) format.

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

These are some common configuration recipes to get you started quickly.

#### Basic configuration[​](#basic-configuration "Direct link to Basic configuration")

The most basic configuration to get you started - it assumes you're running Kafka Connect in distributed mode and have a ClickHouse server running on `localhost:8443` with SSL enabled, data is in schemaless JSON.

```
{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    "tasks.max": "1",
    "consumer.override.max.poll.records": "5000",
    "consumer.override.max.partition.fetch.bytes": "5242880",
    "database": "default",
    "errors.retry.timeout": "60000",
    "exactlyOnce": "false",
    "hostname": "localhost",
    "port": "8443",
    "ssl": "true",
    "jdbcConnectionProperties": "?ssl=true&sslmode=strict",
    "username": "default",
    "password": "<PASSWORD>",
    "topics": "<TOPIC_NAME>",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "clickhouseSettings": ""
  }
}
```

#### Basic configuration with multiple topics[​](#basic-configuration-with-multiple-topics "Direct link to Basic configuration with multiple topics")

The connector can consume data from multiple topics

```
{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "topics": "SAMPLE_TOPIC, ANOTHER_TOPIC, YET_ANOTHER_TOPIC",
    ...
  }
}
```

#### Basic configuration with DLQ[​](#basic-configuration-with-dlq "Direct link to Basic configuration with DLQ")

```
{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "<DLQ_TOPIC>",
    "errors.deadletterqueue.context.headers.enable": "true",
  }
}
```

#### Using with different data formats[​](#using-with-different-data-formats "Direct link to Using with different data formats")

##### Avro schema support[​](#avro-schema-support "Direct link to Avro schema support")

```
{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "<SCHEMA_REGISTRY_HOST>:<PORT>",
    "value.converter.schemas.enable": "true",
  }
}
```

##### Protobuf schema support[​](#protobuf-schema-support "Direct link to Protobuf schema support")

```
{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "io.confluent.connect.protobuf.ProtobufConverter",
    "value.converter.schema.registry.url": "<SCHEMA_REGISTRY_HOST>:<PORT>",
    "value.converter.schemas.enable": "true",
  }
}
```

Please note: if you encounter issues with missing classes, not every environment comes with the protobuf converter and you may need an alternate release of the jar bundled with dependencies.

##### JSON schema support[​](#json-schema-support "Direct link to JSON schema support")

```
{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  }
}
```

##### String support[​](#string-support "Direct link to String support")

The connector supports the String Converter in different ClickHouse formats: [JSON](/docs/interfaces/formats/JSONEachRow), [CSV](/docs/interfaces/formats/CSV), and [TSV](/docs/interfaces/formats/TabSeparated).

```
{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "customInsertFormat": "true",
    "insertFormat": "CSV"
  }
}
```

### Internal buffering[​](#internal-buffering "Direct link to Internal buffering")

Internal buffering allows the sink task to accumulate records from multiple `poll()` calls and flush them to ClickHouse as larger batches. This can improve throughput in workloads where each poll produces many small per-partition batches.

Key behavior:

- `bufferCount` controls how many records are buffered before flushing.
- `bufferFlushTime` sets a maximum wait time (in milliseconds) before flushing buffered records.
- `bufferFlushTime` is only effective when `bufferCount > 0`.
- `bufferCount=0` and `bufferFlushTime=0` keep buffering disabled (default behavior).
- Buffering is not supported when `exactlyOnce=true`.

Why buffering is incompatible with exactly-once mode:
Buffering changes batch boundaries, which breaks ClickHouse block deduplication and the connector offset state machine.  
To resolve this, either disable exactly-once mode with `exactlyOnce=false` in your connector config, or disable buffering with `bufferCount=0`.

Example:

```
{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "exactlyOnce": "false",
    "bufferCount": "5000",
    "bufferFlushTime": "2000"
  }
}
```

### Logging[​](#logging "Direct link to Logging")

Logging is automatically provided by Kafka Connect Platform.
The logging destination and format might be configured via Kafka connect [configuration file](https://docs.confluent.io/platform/current/connect/logging.html#log4j-properties-file).

If using the Confluent Platform, the logs can be seen by running a CLI command:

```
confluent local services connect log
```

For additional details check out the official [tutorial](https://docs.confluent.io/platform/current/connect/logging.html).

### Monitoring[​](#monitoring "Direct link to Monitoring")

ClickHouse Kafka Connect reports runtime metrics via [Java Management Extensions (JMX)](https://www.oracle.com/technical-resources/articles/javase/jmx.html). JMX is enabled in Kafka Connector by default.

#### ClickHouse-Specific Metrics[​](#clickhouse-specific-metrics "Direct link to ClickHouse-Specific Metrics")

The connector exposes custom metrics via the following MBean name:

```
com.clickhouse:type=ClickHouseKafkaConnector,name=SinkTask{id}
```

|  |  |  |  |  |  |  |  |  |  |  |  |
| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |
| Metric Name Type Description|  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `receivedRecords` long The total number of records received.|  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | | `recordProcessingTime` long Total time in nanoseconds spent grouping and converting records to a unified structure.|  |  |  | | --- | --- | --- | | `taskProcessingTime` long Total time in nanoseconds spent processing and inserting data into ClickHouse. | | | | | | | | | | | |

#### Kafka Producer/Consumer Metrics[​](#kafka-producer-consumer-metrics "Direct link to Kafka Producer/Consumer Metrics")

The connector exposes standard Kafka producer and consumer metrics that provide insights into data flow, throughput, and performance.

**Topic-Level Metrics:**