ClickHouse Kafka Connect Sink

**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**:

```
CREATE TABLE my_table (...)
ENGINE = MergeTree()
ORDER BY (timestamp, id)
SETTINGS 
    -- Increase max insert threads for parallel part writing
    max_insert_threads = 4,
    -- Allow inserts with quorum for reliability (ReplicatedMergeTree)
    insert_quorum = 2
```

For connector-level insert settings:

```
"clickhouseSettings": "insert_quorum=2,insert_quorum_timeout=60000"
```

#### Connection pooling and timeouts[​](#connection-pooling "Direct link to Connection pooling and timeouts")

The connector maintains HTTP connections to ClickHouse. Adjust timeouts for high-latency networks:

```
"clickhouseSettings": "socket_timeout=300000,connection_timeout=30000"
```

- **`socket_timeout`** (default: 30000 ms): Maximum time for read operations
- **`connection_timeout`** (default: 10000 ms): Maximum time to establish connection

Increase these values if you experience timeout errors with large batches.

#### Monitoring and troubleshooting performance[​](#monitoring-performance "Direct link to Monitoring and troubleshooting performance")

Monitor these key metrics:

1. **Consumer lag**: Use Kafka monitoring tools to track lag per partition
2. **Connector metrics**: Monitor `receivedRecords`, `recordProcessingTime`, `taskProcessingTime` via JMX (see [Monitoring](#monitoring))
3. **ClickHouse metrics**:
   - `system.asynchronous_inserts`: Monitor async insert buffer usage
   - `system.parts`: Monitor part count to detect merge issues
   - `system.merges`: Monitor active merges
   - `system.events`: Track `InsertedRows`, `InsertedBytes`, `FailedInsertQuery`

**Common performance issues**:

|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |
| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |
| Symptom Possible Cause Solution|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | High consumer lag Batches too small Increase `max.poll.records`, enable async inserts| "Too many parts" errors Small frequent inserts Enable async inserts, increase batch size|  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | Timeout errors Large batch size, slow network Reduce batch size, increase `socket_timeout`, check network| High CPU usage Too many small parts Enable async inserts, increase merge settings|  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | | OutOfMemory errors Batch size too large Reduce `max.poll.records`, `max.partition.fetch.bytes`| Uneven task load Uneven partition distribution Rebalance partitions or adjust `tasks.max` | | | | | | | | | | | | | | | | | | | | |

#### Best practices summary[​](#performance-best-practices "Direct link to Best practices summary")

1. **Start with defaults**, then measure and tune based on actual performance
2. **Prefer larger batches**: Aim for 10,000-100,000 rows per insert when possible
3. **Use async inserts** when sending many small batches or under high concurrency
4. **Always use `wait_for_async_insert=1`** with exactly-once semantics
5. **Scale horizontally**: Increase `tasks.max` up to the number of partitions
6. **One connector per high-volume topic** for maximum throughput
7. **Monitor continuously**: Track consumer lag, part count, and merge activity
8. **Test thoroughly**: Always test configuration changes under realistic load before production deployment

#### Example: High-throughput configuration[​](#example-high-throughput "Direct link to Example: High-throughput configuration")

Here's a complete example optimized for high throughput:

```
{
  "name": "clickhouse-high-throughput",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    "tasks.max": "8",
    
    "topics": "high_volume_topic",
    "hostname": "my-clickhouse-host.cloud",
    "port": "8443",
    "database": "default",
    "username": "default",
    "password": "<PASSWORD>",
    "ssl": "true",
    
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    
    "exactlyOnce": "false",
    "ignorePartitionsWhenBatching": "true",
    
    "consumer.max.poll.records": "10000",
    "consumer.max.partition.fetch.bytes": "5242880",
    "consumer.fetch.min.bytes": "1048576",
    "consumer.fetch.max.wait.ms": "500",
    
    "clickhouseSettings": "async_insert=1,wait_for_async_insert=1,async_insert_max_data_size=16777216,async_insert_busy_timeout_ms=1000,socket_timeout=300000"
  }
}
```

**This configuration**:

- Processes up to 10,000 records per poll
- Batches across partitions for larger inserts
- Uses async inserts with 16 MB buffer
- Runs 8 parallel tasks (match your partition count)
- Optimized for throughput over strict ordering

### Troubleshooting[​](#troubleshooting "Direct link to Troubleshooting")

#### "State mismatch for topic `[someTopic]` partition `[0]`"[​](#state-mismatch-for-topic-sometopic-partition-0 "Direct link to state-mismatch-for-topic-sometopic-partition-0")

This happens when the offset stored in KeeperMap is different from the offset stored in Kafka, usually when a topic has been deleted
or the offset has been manually adjusted.
To fix this, you would need to delete the old values stored for that given topic + partition.

**NOTE: This adjustment may have exactly-once implications.**

#### "What errors will the connector retry?"[​](#what-errors-will-the-connector-retry "Direct link to \"What errors will the connector retry?\"")

Right now the focus is on identifying errors that are transient and can be retried, including:

- `ClickHouseException` - This is a generic exception that can be thrown by ClickHouse.
  It is usually thrown when the server is overloaded and the following error codes are considered particularly transient:
  - 3 - UNEXPECTED\_END\_OF\_FILE
  - 107 - FILE\_DOESNT\_EXIST
  - 159 - TIMEOUT\_EXCEEDED
  - 164 - READONLY
  - 202 - TOO\_MANY\_SIMULTANEOUS\_QUERIES
  - 203 - NO\_FREE\_CONNECTION
  - 209 - SOCKET\_TIMEOUT
  - 210 - NETWORK\_ERROR
  - 241 - MEMORY\_LIMIT\_EXCEEDED
  - 242 - TABLE\_IS\_READ\_ONLY
  - 252 - TOO\_MANY\_PARTS
  - 285 - TOO\_FEW\_LIVE\_REPLICAS
  - 319 - UNKNOWN\_STATUS\_OF\_INSERT
  - 425 - SYSTEM\_ERROR
  - 999 - KEEPER\_EXCEPTION
- `SocketTimeoutException` - This is thrown when the socket times out.
- `UnknownHostException` - This is thrown when the host can't be resolved.
- `IOException` - This is thrown when there is a problem with the network.

#### "All my data is blank/zeroes"[​](#all-my-data-is-blankzeroes "Direct link to \"All my data is blank/zeroes\"")

Likely the fields in your data don't match the fields in the table - this is especially common with CDC (and the Debezium format).
One common solution is to add the flatten transformation to your connector configuration:

```
transforms=flatten
transforms.flatten.type=org.apache.kafka.connect.transforms.Flatten$Value
transforms.flatten.delimiter=_
```

This will transform your data from a nested JSON to a flattened JSON (using `_` as a delimiter). Fields in the table would then follow the "field1\_field2\_field3" format (i.e. "before\_id", "after\_id", etc.).

#### "I want to use my Kafka keys in ClickHouse"[​](#i-want-to-use-my-kafka-keys-in-clickhouse "Direct link to \"I want to use my Kafka keys in ClickHouse\"")