Is there a configuration example using confluentinc/cp-kafka-connect docker?

[larryluogit]

Hi, can someone share an example configuring clickhouse-kafka sink connector with confluentinc/cp-kafka-connect docker container please?

[abdelhakimbendjabeur]

Here is a working example using apicurio schema registry

connect-distributed.properties

bootstrap.servers=SASL_SSL://xxx:9092
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule   required username='xxx'   password='xxx';
consumer.sasl.mechanism=PLAIN
consumer.security.protocol=SASL_SSL
consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule   required username='xxx'   password='xxx';
consumer.auto.offset.reset=earliest
consumer.max.poll.records=100000
consumer.max.partition.fetch.bytes=800000000
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=io.apicurio.registry.utils.converter.AvroConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true
value.converter.apicurio.registry.url=http://xxx:8080/apis/registry/v2
value.converter.apicurio.auth.username=xxx
value.converter.apicurio.auth.password=xxx
value.converter.apicurio.registry.converter.serializer=io.apicurio.registry.serde.avro.AvroKafkaSerializer
value.converter.apicurio.registry.converter.deserializer=io.apicurio.registry.serde.avro.AvroKafkaDeserializer
offset.storage.topic=kc.internal.clickhouse-sink.offsets
offset.storage.replication.factor=3
config.storage.topic=kc.internal.clickhouse-sink.configs
config.storage.replication.factor=3
status.storage.topic=kc.internal.clickhouse-sink.status
status.storage.replication.factor=3
offset.flush.interval.ms=10000
plugin.path=/usr/share/java,/usr/share/confluent-hub-components,/usr/share/java/kafka-connect-plugins/

sink-connector.json

{
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    "database": "xxx",
    "exactlyOnce": false,
    "hostname": "xxx",
    "maxBufferSize": 100000,
    "password": "xxx",
    "port": "8443",
    "retryCount": 3,
    "ssl": true,
    "tasks.max": 2,
    "timeoutSeconds": 30,
    "topics": "som.topic-name",
    "transforms": "ReplaceTopicName",  # This is a custom SMT
    "transforms.ReplaceTopicName.topics": "som.topic-name:table_name",
    "transforms.ReplaceTopicName.type": "com.xxx.kafka.connect.transforms.ReplaceTopicName",
    "username": "xxx"
}

[askmyhat]

Dockerfile:

FROM confluentinc/cp-kafka-connect:7.0.9
RUN curl -k -SL "https://github.com/ClickHouse/clickhouse-kafka-connect/releases/download/v1.0.1/clickhouse-kafka-connect-v1.0.1.zip"  --output /tmp/clickhouse-kafka-connect-v1.0.1.jar
RUN confluent-hub install --no-prompt /tmp/clickhouse-kafka-connect-v1.0.1.jar
COPY jmx_prometheus_exporter.yaml /jmx_prometheus_exporter.yaml
COPY properties.json /properties.json

Environment variables for container:

KAFKA_OPTS=-javaagent:/usr/share/java/cp-base-new/jmx_prometheus_javaagent-0.17.2.jar=4999:/jmx_prometheus_exporter.yaml
CONNECT_REST_PORT=8083
CONNECT_REST_ADVERTISED_HOST_NAME=kafka-connect
CONNECT_PLUGIN_PATH=/usr/share/java,/usr/share/confluent-hub-components,/data/connect-jars
CONNECT_BOOTSTRAP_SERVERS=kafka:9092
CONNECT_GROUP_ID=kafka-connect
CONNECT_KEY_CONVERTER=org.apache.kafka.connect.storage.StringConverter
CONNECT_VALUE_CONVERTER=org.apache.kafka.connect.json.JsonConverter
CONNECT_CONFIG_STORAGE_TOPIC=connect-configs
CONNECT_OFFSET_STORAGE_TOPIC=connect-offsets
CONNECT_STATUS_STORAGE_TOPIC=connect-status
CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR=3
CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR=3
CONNECT_STATUS_STORAGE_REPLICATION_FACTOR=3

Upload connector configuration in a running container:

docker exec -it <kafka-connect-container-name> curl -XPUT \
-H "Content-Type: application/json" \
--data "@/properties.json" \
http://localhost:8083/connectors/clickhouse-connector/config

[Paultagoras]

This is the docker compose I use while testing locally (I also use a .env file to avoid duplication):

---
version: '2'
name: 'confluent-connect'
services:
  connect:
    image: confluentinc/cp-kafka-connect:latest
    hostname: kafka-connect
    container_name: kafka-connect
    volumes:
      - "~/DockerShare:/usr/share/dockershare/:ro"
    ports:
      - "8083:8083"
    environment:
      CONNECT_BOOTSTRAP_SERVERS: ${BOOTSTRAP_SERVERS}
      CONNECT_SECURITY_PROTOCOL: ${SECURITY_PROTOCOL}
      CONNECT_CONSUMER_SECURITY_PROTOCOL: ${SECURITY_PROTOCOL}
      CONNECT_PRODUCER_SECURITY_PROTOCOL: ${SECURITY_PROTOCOL}
      CONNECT_SASL_MECHANISM: ${SASL_MECHANISM}
      CONNECT_CONSUMER_SASL_MECHANISM: ${SASL_MECHANISM}
      CONNECT_PRODUCER_SASL_MECHANISM: ${SASL_MECHANISM}
      CONNECT_SASL_JAAS_CONFIG: ${SASL_JAAS_CONFIG}
      CONNECT_CONSUMER_SASL_JAAS_CONFIG: ${SASL_JAAS_CONFIG}
      CONNECT_PRODUCER_SASL_JAAS_CONFIG: ${SASL_JAAS_CONFIG}
      CONNECT_CONSUMER_OVERRIDE_MAX_POLL_RECORDS: 1500
      CONNECT_CONSUMER_MAX_POLL_RECORDS: 5000
      CONNECT_REST_ADVERTISED_HOST_NAME: connect
      CONNECT_GROUP_ID: local-connect-group
      CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000
      CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter
      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CLASSPATH: /usr/share/java/monitoring-interceptors/monitoring-interceptors-7.3.0.jar
      CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components,/usr/share/dockershare"
      CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=ERROR,org.I0Itec.zkclient=ERROR,org.reflections=ERROR,com.clickhouse=ERROR

This is a sample configuration for the connector that I use locally as well:

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    "tasks.max": "1",
    "database": "default",
    "errors.retry.timeout": "60",
    "exactlyOnce": "false",
    "hostname": "REDACTED",
    "password": "RECACTED",
    "port": "8443",
    "ssl": "true",
    "topics": "testing",
    "username": "default",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "clickhouseSettings": "",
    "topic2TableMap": ""
  }
}

[Paultagoras]

Check out https://clickhouse.com/docs/en/integrations/kafka#clickhouse-kafka-connect-sink for additional examples and documentation :)