Prerequisites

• The URL to connect to the Clickhouse Cloud database, specifying the port and database.
  Example:
1.   jdbc:clickhouse://<cloud_host>:<clickhouse_port>/<database>?ssl=true
• Connection user: 
  
• JDBC user: Clickhouse Username.

• Connection password:
  
  • JDBC password: Clickhouse Password.

6. Writes Settings:

• Insert mode: Registration insertion mode. Options: Insert.
• Batch size: Number of records grouped for SQL transactions.
• Enable Delete: Allows the deletion of records in the target database.  Options: True or False.
  

• The insert mode depend on the Primary key mode configuration (record_key or record_value) to accurately identify records.

• The Primary key mode, insert does not use keys to check for duplicates.

• The Enable deletions para meter depends on the Primary key mode set  of configuration a record_key.

• Key converter schema log URL: Confluent schema log URLs.
  Example:

• Value converter scheme record URL: URL of the Confluent Schemes Register for Values.
  Example:

1. ExtractTimestamp.type: Add a timestamp to the records.
   Example: org.apache.kafka.connect.transforms.InsertField$Value
2. transformes.ExtractTimestamp.timestamp.field: Field where the timestamp of the event will be inserted.
   Example: timestamp
3. transforms.InsertTimezone.type: Adds a time zone field to the records.
   Example: org.apache.kafka.connect.transforms.InsertField$Value
4. transforms.InsertTimezone.static.field: Static field for time zone.
   Example: timezone
5. transforms.InsertTimezone.static.value:Time zone field value.
   Example: America/Mexico_City

Limitations

1. When using the Onibex Clickhouse Sink Connector, please consider the following constraints, primarily due to ClickHouse JDBC driver behavior and schema handling:
   

2. No Support for GENERATED ALWAYS AS Columns
   Auto-created columns do not support expressions using GENERATED ALWAYS AS. For nullable fields, the default value is NULL.

3. Restricted Engine Support for Table Creation
   Automatic table creation is limited to ENGINE = ReplacingMergeTree(_version). Other table engines must be created manually.

4. Schema Registry Requirements
   Both key and value schemas must be registered in the Schema Registry to support correct insert and delete operations.

5. ClickHouse Kafka Connector Schema Requirements
   

1. The ClickHouse Kafka Sink Connector has specific schema requirements that must be understood before implementation:

   Supported Schema Types
   

2. JSON Schema with flat structure
3. Avro Schema with flat structure

Flat Structure Requirement: The connector only supports schemas with a flat, non-nested structure. Complex or nested data structures are not supported because the connector maps each field directly to a table column with primitive data types in ClickHouse.

No String Serialization Support: Topics using plain string serialization are not supported because they lack the required schema definition needed for proper field mapping.

Each Kafka message must contain:

• Key Schema: Used for partitioning or deduplication purposes
• Value Schema: Contains the structured payload that maps to ClickHouse table columns

The connector requires structured objects (Struct) for the message value. Each message value must:

1. Have a defined schema (JSON Schema or Avro)
2. Contain only primitive data types (string, integer, boolean, etc.)
3. Use a flat structure without nested objects or arrays
4. Be registered with a Schema Registry (when applicable)

• Plain string topics without schema
• Nested JSON objects
• Complex data structures (arrays of objects, nested structs)
• Schema-less message formats

// ✅ Supported - Flat structure with primitive types
{
  "user_id": 12345,
  "username": "john_doe",
  "email": "john@example.com",
  "created_at": "2025-05-28T10:30:00Z",
  "is_active": true
}

// ❌ Not Supported - Nested structure
{
  "user_id": 12345,
  "profile": {
    "username": "john_doe",
    "contact": {
      "email": "john@example.com"
    }
  }
}

This limitation ensures that each schema field maps directly to a corresponding column in the target ClickHouse table with appropriate primitive data types.

Avro Schema Example
🔑 Key Schema (Avro)
The Key Schema represents the Kafka message key and can be composed of one or more fields, typically used for partitioning, deduplication, or uniquely identifying the message.

Example with a single key (simple key):

1. 
   
1. {

     "type": "record",

     "name": "KeySchema",

     "namespace": "com.example.kafka",

     "fields": [

       { "name": "id", "type": "string" }                                                                                                                                                                           

     ]

   }

Data types supported for the connector

Support for developments in the scheme

Name	Description	Values
auto.create	Allows automatic table creation if they do not exist	true / false
auto.evolve	Allows automatic evolution of tables if the outline changes	true / false

Converters

Name	Description	Values
key.converter	Key converter in Avro format	io.confluent.connect.avro.AvroConverter
value.converter	Value converter in Avro format	io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url	Confluent Platform schema log URLs	http:/<schema_registry_ip>:<schema_registry_port>
value.converter.schema.registry.url	Confluent Platform Log URL for Values	http:/<schema_registry_ip>:<schema_registry_port>

Transformations

Name	Description	Values
transformations	Transformations applied to data	ExtractTimestamp, InsertTimezone
ExtractTimestamp.type	Type of transformation to add the timestamp to records	org.apache.kafka.connect.transforms.InsertField$Value
transforms.ExtractTimestamp.timestamp.field	Field where the event timestamp will be inserted	timestamp
transforms.InsertTimezone.type	Type of transformation to add the time zone	org.apache.kafka.connect.transforms.InsertField$Value
transforms.InsertTimezone.static.field	Static field where the time zone will be inserted	timezone
transforms.InsertTimezone.static.value	Time zone field value	America/Mexico_City

---

Examples Clickhouse On-premise

Example of setup with schema avro and password logging (Confluent Platform)

{
 "value.converter.schema.registry.url": "http://<schema_registry_ip>:<schema_registry_port>",
 "key.converter.schema.registry.url": "http://<schema_registry_ip>:<schema_registry_port>",
"name": "<connector_name>",
"connector.class": "com.onibex.connect.clickhouse.jdbc.OnibexClickhouseSinkConnector",
 "tasks.max": "1",
 "key.converter": "io.confluent.connect.avro.AvroConverter",
 "value.converter": "io.confluent.connect.avro.AvroConverter",
 "transforms": "ExtractTimestamp",
 "topics": "<topic_name>",
 "transforms.ExtractTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value", "transforms.ExtractTimestamp.timestamp.field": "timestamp",
 "connection.url": "jdbc:clickhouse://<host/ip>:<port>/<database>",
 "connection.user": "<username_clickhouse>",
 "connection.password": "<password_clickhouse>",
 "insert.mode": "INSERT",
 "delete.enabled": "true",
 "table.name.format": "{$topic}",
 "pk.mode": "record_key",
 "auto.create": "true",
 "auto.evolve": "true",
 "onibex.license": "<valid_onibex_license>"
 }

Example of setup with schema Json and password logging (Confluent Platform)

{
 "value.converter.schema.registry.url": "http://<schema_registry_ip>:<schema_registry_port>",
 "key.converter.schema.registry.url": "http://<schema_registry_ip>:<schema_registry_port>",
"name": "<connector_name>",
"connector.class": "com.onibex.connect.clickhouse.jdbc.OnibexClickhouseSinkConnector",
 "tasks.max": "1",
 "key.converter": "io.confluent.connect.json.JsonSchemaConverter",
 "value.converter": "io.confluent.connect.json.JsonSchemaConverter",
 "transforms": "ExtractTimestamp",
 "topics": "<topic_name>",
 "transforms.ExtractTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value",                                                                        
 "transforms.ExtractTimestamp.timestamp.field": "timestamp",
 "connection.url": "jdbc:clickhouse://<host/ip>:<port>/<database>",
 "connection.user": "<username_clickhouse>",
 "connection.password": "<password_clickhouse>",
 "insert.mode": "INSERT",
 "delete.enabled": "true",
 "table.name.format": "{$topic}",
 "pk.mode": "record_key",
 "auto.create": "true",
 "auto.evolve": "true",
 "onibex.license": "<valid_onibex_license>"
 }

Example of setup with schema avro  and password logging (Confluent Cloud)