ClickHouse Kafka Connect Sink

​

License

​

Requirements for the environment

​

Version compatibility matrix

​

Main features

• Shipped with out-of-the-box exactly-once semantics. It’s powered by a new ClickHouse core feature named KeeperMap (used as a state store by the connector) and allows for minimalistic architecture.
• Support for 3rd-party state stores: Currently defaults to In-memory but can use KeeperMap (Redis to be added soon).
• Core integration: Built, maintained, and supported by ClickHouse.
• Tested continuously against ClickHouse Cloud.
• Data inserts with a declared schema and schemaless.
• Support for all data types of ClickHouse.

​

Installation instructions

​

Gather your connection details

​

General installation instructions

• Download a zip archive containing the Connector JAR file from the Releases page of ClickHouse Kafka Connect Sink repository.
• Extract the ZIP file content and copy it to the desired location.
• Add a path with the plugin director to plugin.path configuration in your Connect properties file to allow Confluent Platform to find the plugin.
• Provide a topic name, ClickHouse instance hostname, and password in config.

connector.class=com.clickhouse.kafka.connect.ClickHouseSinkConnector
tasks.max=1
topics=<topic_name>
ssl=true
jdbcConnectionProperties=?sslmode=STRICT
security.protocol=SSL
hostname=<hostname>
database=<database_name>
password=<password>
ssl.truststore.location=/tmp/kafka.client.truststore.jks
port=8443
value.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
exactlyOnce=true
username=default
schemas.enable=false

• Restart the Confluent Platform.
• If you use Confluent Platform, log into Confluent Control Center UI to verify the ClickHouse Sink is available in the list of available connectors.

​

Configuration options

• 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.

​

Target tables

​

Pre-processing

​

Supported data types

• (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.

​

Configuration recipes

​

Basic configuration

{
  "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

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

​

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",
  }
}

​

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",
  }
}

​

Avro type mapping

​

Unsupported Avro schemas

• fixed decimal logical type

{"name": "decimal_18_4", "type": "fixed", "size": 8, "logicalType": "decimal", "precision": 18, "scale": 4}

• nullable unions

{"name": "mixed_union", "type": ["null", "string", "int"], "default": null}

• record unions

{
  "name": "record_union",
  "type": [
    {
      "type": "record",
      "name": "TypeA",
      "fields": [
        {
          "name": "label",
          "type": "string"
        }
      ]
    },
    {
      "type": "record",
      "name": "TypeB",
      "fields": [
        {
          "name": "count",
          "type": "int"
        }
      ]
    }
  ]
}

​

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",
  }
}

​

Protobuf type mapping

​

Note on translating oneof fields to ClickHouse columns

syntax = "proto3";

package com.clickhouse.kafka.connect.proto.test;

message StringIntUnion {
  oneof mixed {
    string mixed_string = 2;
    int32 mixed_int = 3;
  }
}

CREATE TABLE IF NOT EXISTS `StringIntUnion`
(
    mixed_string Nullable(String),
    mixed_int Nullable(Int32)
) ENGINE = ...;

​

Unsupported Protobuf schemas

• multi-message unions (before CH version 26.1)

syntax = "proto3";

package com.clickhouse.kafka.connect.proto.test;

message TwoRecords {
  oneof payload {
    TypeA type_a = 2;
    TypeB type_b = 3;
  }

  // translates to Nullable(Tuple(label String)) in ClickHouse, which is unsupported
  message TypeA {
    string label = 1;
  }

  // translates to Nullable(Tuple(count Int32)) in ClickHouse, which is unsupported
  message TypeB {
    int32 count = 1;
  }
}

​

JSON schema support

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

​

String support

{
  "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

• 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.

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

​

Logging

confluent local services connect log

​

Monitoring

​

ClickHouse-Specific Metrics

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

​

Kafka Producer/Consumer Metrics

• records-sent-total: Total number of records sent to the topic
• bytes-sent-total: Total bytes sent to the topic
• record-send-rate: Average rate of records sent per second
• byte-rate: Average bytes sent per second
• compression-rate: Compression ratio achieved

• records-sent-total: Total records sent to the partition
• bytes-sent-total: Total bytes sent to the partition
• records-lag: Current lag in the partition
• records-lead: Current lead in the partition
• replica-fetch-lag: Lag information for replicas

• connection-creation-total: Total connections created to the Kafka node
• connection-close-total: Total connections closed
• request-total: Total requests sent to the node
• response-total: Total responses received from the node
• request-rate: Average request rate per second
• response-rate: Average response rate per second

• Throughput: Track data ingestion rates
• Lag: Identify bottlenecks and processing delays
• Compression: Measure data compression efficiency
• Connection Health: Monitor network connectivity and stability

​

Kafka Connect Framework Metrics

• task-count: Total number of tasks in the connector
• running-task-count: Number of tasks currently running
• paused-task-count: Number of tasks currently paused
• failed-task-count: Number of tasks that have failed
• destroyed-task-count: Number of destroyed tasks
• unassigned-task-count: Number of unassigned tasks

• deadletterqueue-produce-failures: Number of failed DLQ writes
• deadletterqueue-produce-requests: Total DLQ write attempts
• last-error-timestamp: Timestamp of the last error
• records-skip-total: Total number of records skipped due to errors
• records-retry-total: Total number of records that were retried
• errors-total: Total number of errors encountered

• offset-commit-failures: Number of failed offset commits
• offset-commit-avg-time-ms: Average time for offset commits
• offset-commit-max-time-ms: Maximum time for offset commits
• put-batch-avg-time-ms: Average time to process a batch
• put-batch-max-time-ms: Maximum time to process a batch
• source-record-poll-total: Total records polled

​

Monitoring Best Practices

1. Monitor Consumer Lag: Track records-lag per partition to identify processing bottlenecks
2. Track Error Rates: Watch errors-total and records-skip-total to detect data quality issues
3. Observe Task Health: Monitor task status metrics to ensure tasks are running properly
4. Measure Throughput: Use records-send-rate and byte-rate to track ingestion performance
5. Monitor Connection Health: Check node-level connection metrics for network issues
6. Track Compression Efficiency: Use compression-rate to optimize data transfer

​

Limitations

• Deletes aren’t supported.
• Batch size is inherited from the Kafka Consumer properties.
• When using KeeperMap for exactly-once and the offset is changed or re-wound, you need to delete the content from KeeperMap for that specific topic. (See troubleshooting guide below for more details)

​

Performance tuning and throughput optimization

​

When is performance tuning needed?

• High-throughput workloads: When processing millions of events per second from Kafka topics
• Consumer lag: When your connector can’t keep up with the rate of data production, causing increasing lag
• Resource constraints: When you need to optimize CPU, memory, or network usage
• Multiple topics: When consuming from multiple high-volume topics simultaneously
• Small message sizes: When dealing with many small messages that would benefit from server-side batching

• You’re processing low to moderate volumes (< 10,000 messages/second)
• Consumer lag is stable and acceptable for your use case
• Default connector settings already meet your throughput requirements
• Your ClickHouse cluster can easily handle the incoming load

​

Understanding the data flow

1. Kafka Connect Framework fetches messages from Kafka topics in the background
2. Connector polls for messages from the framework’s internal buffer
3. Connector batches messages based on poll size
4. ClickHouse receives the batched insert via HTTP/S
5. ClickHouse processes the insert (synchronously or asynchronously)

​

Kafka Connect batch size tuning

• fetch.min.bytes: Minimum amount of data before the framework passes values to the connector (default: 1 byte)
• fetch.max.bytes: Maximum amount of data to fetch in a single request (default: 52428800 / 50 MB)
• fetch.max.wait.ms: Maximum time to wait before returning data if fetch.min.bytes isn’t met (default: 500 ms)

• max.poll.records: Maximum number of records returned in a single poll (default: 500)
• max.partition.fetch.bytes: Maximum amount of data per partition (default: 1048576 / 1 MB)

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

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

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

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

• 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

​

Asynchronous inserts

• 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

• 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

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: 100 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)

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

• 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.

"clickhouseSettings": "async_insert=1,wait_for_async_insert=1,async_insert_max_data_size=104857600,async_insert_busy_timeout_ms=1000"

• async_insert_max_data_size (default: 104857600 / 100 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

• 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

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

​

Connector parallelism

"tasks.max": "4"

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

"ignorePartitionsWhenBatching": "true"

​

Multiple high throughput topics

​

ClickHouse table engine considerations

• 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

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

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

​

Connection pooling and timeouts

"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

​

Monitoring and troubleshooting performance

1. Consumer lag: Use Kafka monitoring tools to track lag per partition
2. Connector metrics: Monitor receivedRecords, recordProcessingTime, taskProcessingTime via JMX (see 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

​

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

{
  "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.override.max.poll.records": "10000",
    "consumer.override.max.partition.fetch.bytes": "5242880",
    "consumer.override.fetch.min.bytes": "1048576",
    "consumer.override.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"
  }
}

• 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

​

“State mismatch for topic [someTopic] partition [0]”

-- First, identify the database used to store the data.
SELECT * FROM [database].connect_state

-- Identify the key that matches the topic and partition.
ALTER TABLE [database].connect_state DELETE WHERE key = [keyname]

​

“What errors will the connector retry?”

• 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”

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

​

“I want to use my Kafka keys in ClickHouse”

transforms=keyToValue
transforms.keyToValue.type=com.clickhouse.kafka.connect.transforms.KeyToValue
transforms.keyToValue.field=_key