ClickHouse Kafka Connect Sink

Note

On Confluent Cloud, adjustment of these settings requires opening a support case through Confluent Cloud.

##### Recommended settings for high throughput[​](#recommended-batch-settings "Direct link to Recommended settings for high throughput")

For optimal performance with ClickHouse, aim for larger batches:

```
# Increase the number of records per poll
consumer.max.poll.records=5000

# Increase the partition fetch size (5 MB)
consumer.max.partition.fetch.bytes=5242880

# Optional: Increase minimum fetch size to wait for more data (1 MB)
consumer.fetch.min.bytes=1048576

# Optional: Reduce wait time if latency is critical
consumer.fetch.max.wait.ms=300
```

**Important**: Kafka Connect fetch settings represent compressed data, while ClickHouse receives uncompressed data. Balance these settings based on your compression ratio.

**Trade-offs**:

- **Larger batches** = Better ClickHouse ingestion performance, fewer parts, lower overhead
- **Larger batches** = Higher memory usage, potential increased end-to-end latency
- **Too large batches** = Risk of timeouts, OutOfMemory errors, or exceeding `max.poll.interval.ms`

More details: [Confluent documentation](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration) | [Kafka documentation](https://kafka.apache.org/documentation/#consumerconfigs)

#### Asynchronous inserts[​](#asynchronous-inserts "Direct link to Asynchronous inserts")

Asynchronous inserts are a powerful feature when the connector sends relatively small batches or when you want to further optimize ingestion by shifting batching responsibility to ClickHouse.

##### When to use async inserts[​](#when-to-use-async-inserts "Direct link to When to use async inserts")

Consider enabling async inserts when:

- **Many small batches**: Your connector sends frequent small batches (< 1000 rows per batch)
- **High concurrency**: Multiple connector tasks are writing to the same table
- **Distributed deployment**: Running many connector instances across different hosts
- **Part creation overhead**: You're experiencing "too many parts" errors
- **Mixed workload**: Combining real-time ingestion with query workloads

Do **NOT** use async inserts when:

- You're already sending large batches (> 10,000 rows per batch) with controlled frequency
- You require immediate data visibility (queries must see data instantly)
- Exactly-once semantics with `wait_for_async_insert=0` conflicts with your requirements
- Your use case can benefit from client-side batching improvements instead

##### How async inserts work[​](#how-async-inserts-work "Direct link to How async inserts work")

With asynchronous inserts enabled, ClickHouse:

1. Receives the insert query from the connector
2. Writes data to an in-memory buffer (instead of immediately to disk)
3. Returns success to the connector (if `wait_for_async_insert=0`)
4. Flushes the buffer to disk when one of these conditions is met:
   - Buffer reaches `async_insert_max_data_size` (default: 10 MB)
   - `async_insert_busy_timeout_ms` milliseconds elapsed since first insert (default: 1000 ms)
   - Maximum number of queries accumulated (`async_insert_max_query_number`, default: 100)

This significantly reduces the number of parts created and improves overall throughput.

##### Enabling async inserts[​](#enabling-async-inserts "Direct link to Enabling async inserts")

Add async insert settings to the `clickhouseSettings` configuration parameter:

```
{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
  }
}
```

**Key settings**:

- **`async_insert=1`**: Enable asynchronous inserts
- **`wait_for_async_insert=1`** (recommended): Connector waits for data to be flushed to ClickHouse storage before acknowledging. Provides delivery guarantees.
- **`wait_for_async_insert=0`**: Connector acknowledges immediately after buffering. Better performance but data may be lost on server crash before flush.

##### Tuning async insert behavior[​](#tuning-async-inserts "Direct link to Tuning async insert behavior")

You can fine-tune the async insert flush behavior:

```
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1,async_insert_max_data_size=10485760,async_insert_busy_timeout_ms=1000"
```

Common tuning parameters:

- **`async_insert_max_data_size`** (default: 10485760 / 10 MB): Maximum buffer size before flush
- **`async_insert_busy_timeout_ms`** (default: 1000): Maximum time (ms) before flush
- **`async_insert_stale_timeout_ms`** (default: 0): Time (ms) since last insert before flush
- **`async_insert_max_query_number`** (default: 100): Maximum queries before flush

**Trade-offs**:

- **Benefits**: Fewer parts, better merge performance, lower CPU overhead, improved throughput under high concurrency
- **Considerations**: Data not immediately queryable, slightly increased end-to-end latency
- **Risks**: Data loss on server crash if `wait_for_async_insert=0`, potential memory pressure with large buffers

##### Async inserts with exactly-once semantics[​](#async-inserts-with-exactly-once "Direct link to Async inserts with exactly-once semantics")

When using `exactlyOnce=true` with async inserts:

```
{
  "config": {
    "exactlyOnce": "true",
    "clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
  }
}
```

**Important**: Always use `wait_for_async_insert=1` with exactly-once to ensure offset commits happen only after data is persisted.

For more information about async inserts, see the [ClickHouse async inserts documentation](/docs/best-practices/selecting-an-insert-strategy#asynchronous-inserts).

#### Connector parallelism[​](#connector-parallelism "Direct link to Connector parallelism")

Increase parallelism to improve throughput:

##### Tasks per connector[​](#tasks-per-connector "Direct link to Tasks per connector")

```
"tasks.max": "4"
```

Each task processes a subset of topic partitions. More tasks = more parallelism, but:

- Maximum effective tasks = number of topic partitions
- Each task maintains its own connection to ClickHouse
- More tasks = higher overhead and potential resource contention

**Recommendation**: Start with `tasks.max` equal to the number of topic partitions, then adjust based on CPU and throughput metrics.

##### Ignoring partitions when batching[​](#ignoring-partitions "Direct link to Ignoring partitions when batching")

By default, the connector batches messages per partition. For higher throughput, you can batch across partitions:

```
"ignorePartitionsWhenBatching": "true"
```

\*\* Warning\*\*: Only use when `exactlyOnce=false`. This setting can improve throughput by creating larger batches but loses per-partition ordering guarantees.

#### Multiple high throughput topics[​](#multiple-high-throughput-topics "Direct link to Multiple high throughput topics")

If your connector is configured to subscribe to multiple topics, you're using `topic2TableMap` to map topics to tables, and you're experiencing a bottleneck at insertion resulting in consumer lag, consider creating one connector per topic instead.

The main reason why this happens is that currently batches are inserted into every table [serially](https://github.com/ClickHouse/clickhouse-kafka-connect/blob/578ac07e8be1a920aaa3b26e49183595c3edd04b/src/main/java/com/clickhouse/kafka/connect/sink/ProxySinkTask.java#L95-L100).

**Recommendation**: For multiple high-volume topics, deploy one connector instance per topic to maximize parallel insert throughput.

#### ClickHouse table engine considerations[​](#table-engine-considerations "Direct link to ClickHouse table engine considerations")

Choose the appropriate ClickHouse table engine for your use case:

- **`MergeTree`**: Best for most use cases, balances query and insert performance
- **`ReplicatedMergeTree`**: Required for high availability, adds replication overhead
- **`*MergeTree` with proper `ORDER BY`**: Optimize for your query patterns

**Settings to consider**: