ClickHouse Kafka Connect Sink Connector

The ClickHouse Kafka Connect Sink is designed around four primary architectural requirements:

1. Independence from ClickHouse: The connector operates as an architecturally independent component from the ClickHouse database.
2. Push-based architecture: Data is pushed from Kafka to ClickHouse. ClickHouse does not need to connect to Kafka, making this suitable for environments where ClickHouse is cloud-hosted and Kafka is self-managed.
3. Full ClickHouse type support: The connector supports all ClickHouse types, including complex types like Map. (Note: Support for Tuple and JSON types is currently in development).
4. Exactly-once delivery: The connector ensures that every Kafka message is delivered to ClickHouse exactly one time, preventing both message loss and duplication.

You can run a complete local environment containing Kafka (broker, schema-registry, connect, control-center, ksqldb-server, rest-proxy) and ClickHouse using Docker Compose.

Prerequisites:

• Ensure you have a ~/kafka-full.yml file configured.
• Replace the CLUSTER_ID in the broker service with a unique base64 UUID generated via bin/kafka-storage.sh random-uuid.
• Note that CLASSPATH is required in the connect service due to CC-2422.

Execution: Run the following command to start the stack:

docker compose -f ~/kafka-full.yml up

The log-analyzer.jar tool allows you to investigate logs for specific Kafka topics, partitions, and offsets. It provides a summary of logs and highlights areas of interest, such as failed batches or specific query IDs.

To analyze all logs in a directory: java -jar log-analyzer.jar <path_to_logs>

To investigate a specific Kafka location, use the -topic, -partition, and -offset flags. This provides a summary of the logs specifically around that offset.

# Analyze all logs in a directory
java -jar log-analyzer.jar ~/Desktop/Logs

# Analyze logs for a specific topic, partition, and offset
java -jar log-analyzer.jar ~/Desktop/Logs -topic=<topicName> -partition=<partitionNumber> -offset=<offset>

Use this setup when you want to connect to existing cloud-hosted Kafka and ClickHouse instances using a local Dockerized Kafka Connect instance.

1. Ensure your .env file is configured with your cloud credentials.
2. Run the following command:

docker compose -f ~/kafka-connect.yml up

Key Configuration Details:

• The connect service uses confluentinc/cp-kafka-connect:latest.
• It maps ports 8083 (REST API), 7778 (JMX), and 8020 (Prometheus JMX).
• It requires a CLASSPATH for monitoring interceptors (see CC-2422).
• The CONNECT_PLUGIN_PATH includes /usr/share/dockershare for local plugin injection.

You can generate Avro binary files from an Interface Definition Language (IDL) file using avro-tools. This is useful for creating sample data or schema definitions for use with the ClickHouse Kafka Connect Sink.

java -jar ~/tools/avro-tools-1.12.0.jar idl event.idl > even.avro

To monitor JMX metrics via Prometheus and Grafana, add the following services to your docker-compose.yml file. This configuration assumes you have a prometheus.yml configuration file located in a local directory mapped to ~/DockerShare.

 prometheus:
   hostname: prometheus
   container_name: prometheus
   image: prom/prometheus
   volumes:
     - "~/DockerShare:/usr/share/dockershare/:ro"
   command:
     - '--config.file=/usr/share/dockershare/prometheus.yml'
   ports:
     - 9090:9090

 grafana:
   image: grafana/grafana
   hostname: grafana
   container_name: grafana
   depends_on:
     - prometheus
   volumes:
     - "~/DockerShare:/usr/share/dockershare/:ro"
   ports:
     - 3000:3000

The benchmark project uses JMH (Java Microbenchmark Harness) to test different components of the ClickHouse Kafka connector, ranging from specific type conversions to end-to-end record conversion.

Before running benchmarks, you must publish the connector to your local Maven repository from the repository root:

./gradlew publishToMavenLocal

To execute a benchmark, use the ./gradlew run command from the benchmark directory with the appropriate --args.

./gradlew run --args="-b kafka_connector.SimpleBenchmark -i 2 -t 10"

To generate Java classes from an existing Avro schema file, use the compile schema command in avro-tools. This allows you to create typed objects that correspond to your Avro data structures.

java -jar ~/tools/avro-tools-1.12.0.jar compile schema event.avro ../java/