ClickHouse Kafka Connect Sink

​

Licence

​

Prérequis de l’environnement

​

Matrice de compatibilité des versions

​

Principales fonctionnalités

• Livré avec une sémantique exactly-once prête à l’emploi. Il repose sur une nouvelle fonctionnalité du cœur de ClickHouse appelée KeeperMap (utilisée comme magasin d’état par le connecteur) et permet une architecture minimaliste.
• Prise en charge des magasins d’état tiers : utilise actuellement par défaut le stockage en mémoire, mais peut aussi utiliser KeeperMap (Redis sera bientôt ajouté).
• Intégration au cœur : développée, maintenue et prise en charge par ClickHouse.
• Testé en continu sur ClickHouse Cloud.
• Insertions de données avec ou sans schéma déclaré.
• Prise en charge de tous les types de données de ClickHouse.

​

Instructions d’installation

​

Rassemblez vos informations de connexion

​

Instructions générales d’installation

• Téléchargez une archive ZIP contenant le fichier JAR du connecteur depuis la page Releases du dépôt ClickHouse Kafka Connect Sink.
• Extrayez le contenu du fichier ZIP et copiez-le à l’emplacement souhaité.
• Ajoutez à la configuration plugin.path de votre fichier de propriétés Connect le chemin du répertoire du plugin, afin que Confluent Platform puisse le trouver.
• Indiquez dans la config le nom du topic, le hostname de l’instance ClickHouse et le mot de passe.

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

• Redémarrez Confluent Platform.
• Si vous utilisez Confluent Platform, connectez-vous à l’UI du Confluent Control Center pour vérifier que le ClickHouse Sink figure dans la liste des connecteurs disponibles.

​

Options de configuration

• les informations de connexion : nom d’hôte (obligatoire) et port (facultatif)
• les identifiants utilisateur : mot de passe (obligatoire) et nom d’utilisateur (facultatif)
• la classe du connecteur : com.clickhouse.kafka.connect.ClickHouseSinkConnector (obligatoire)
• topics ou topics.regex : les topics Kafka à interroger ; les noms de topic doivent correspondre aux noms de table (obligatoire)
• les convertisseurs de clé et de valeur : à définir selon le type de données de votre topic. Obligatoire s’ils ne sont pas déjà définis dans la configuration du worker.

​

Tables cibles

​

Prétraitement

​

Types de données pris en charge

• (1) - JSON est pris en charge uniquement lorsque les paramètres ClickHouse incluent input_format_binary_read_json_as_string=1. Cela fonctionne uniquement avec la famille de formats RowBinary, et ce paramètre affecte toutes les colonnes de la requête d’insertion ; elles doivent donc toutes être de type chaîne. Dans ce cas, le connecteur convertit STRUCT en chaîne JSON.
• (2) - Lorsqu’une struct contient des unions comme oneof, le convertisseur doit être configuré pour NE PAS ajouter de préfixe/suffixe aux noms de champs. Il existe le paramètre generate.index.for.unions=false pour ProtobufConverter.

​

Exemples de configuration

​

Configuration de base

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

​

Configuration de base avec plusieurs topics

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

​

Configuration de base avec 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",
  }
}

​

Prise en charge du schéma Avro

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

​

Correspondance des types Avro

​

Schémas Avro non pris en charge

• type logique decimal de type fixed

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

• unions Nullable

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

• unions d’enregistrements

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

​

Prise en charge du schéma Protobuf

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

​

Correspondance des types Protobuf

​

Remarque sur la traduction des champs oneof en colonnes ClickHouse

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 = ...;

​

Schémas Protobuf non pris en charge

• unions multi-messages (avant la version 26.1 de CH, et avant la version 26.9 de CH sauf si un paramètre est activé, voir ci-dessous)

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 before 26.1
  message TypeA {
    string label = 1;
  }

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

​

Prise en charge du schéma JSON

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

​

Prise en charge de String

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "customInsertFormat": "true",
    "insertFormat": "CSV"
  }
}

​

Mise en mémoire tampon interne

• bufferCount contrôle le nombre d’enregistrements mis en mémoire tampon avant leur envoi.
• bufferFlushTime définit le temps d’attente maximal (en millisecondes) avant l’envoi des enregistrements mis en mémoire tampon.
• bufferFlushTime n’a d’effet que lorsque bufferCount > 0.
• bufferCount=0 et bufferFlushTime=0 laissent la mise en mémoire tampon désactivée (comportement par défaut).
• La mise en mémoire tampon n’est pas prise en charge lorsque exactlyOnce=true.

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

​

Journalisation

confluent local services connect log

​

Supervision

​

Métriques spécifiques à ClickHouse

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

​

Métriques des producteurs et consommateurs Kafka

• records-sent-total : Nombre total d’enregistrements envoyés au topic
• bytes-sent-total : Nombre total d’octets envoyés au topic
• record-send-rate : Taux moyen d’enregistrements envoyés par seconde
• byte-rate : Nombre moyen d’octets envoyés par seconde
• compression-rate : Ratio de compression obtenu

• records-sent-total : Nombre total d’enregistrements envoyés à la partition
• bytes-sent-total : Nombre total d’octets envoyés à la partition
• records-lag : Lag actuel de la partition
• records-lead : Avance actuelle de la partition
• replica-fetch-lag : Informations de lag des répliques

• connection-creation-total : Nombre total de connexions créées vers le nœud Kafka
• connection-close-total : Nombre total de connexions fermées
• request-total : Nombre total de requêtes envoyées au nœud
• response-total : Nombre total de réponses reçues du nœud
• request-rate : Taux moyen de requêtes par seconde
• response-rate : Taux moyen de réponses par seconde

• Débit : Suivre les débits d’ingestion des données
• Lag : Identifier les goulots d’étranglement et les retards de traitement
• Compression : Mesurer l’efficacité de la compression des données
• État des connexions : Surveiller la connectivité réseau et la stabilité

​

Métriques du Kafka Connect Framework

• task-count : Nombre total de tâches dans le connecteur
• running-task-count : Nombre de tâches actuellement en cours d’exécution
• paused-task-count : Nombre de tâches actuellement en pause
• failed-task-count : Nombre de tâches en échec
• destroyed-task-count : Nombre de tâches détruites
• unassigned-task-count : Nombre de tâches non attribuées

• deadletterqueue-produce-failures : Nombre d’écritures dans la DLQ ayant échoué
• deadletterqueue-produce-requests : Nombre total de tentatives d’écriture dans la DLQ
• last-error-timestamp : Horodatage de la dernière erreur
• records-skip-total : Nombre total d’enregistrements ignorés en raison d’erreurs
• records-retry-total : Nombre total d’enregistrements ayant fait l’objet d’une nouvelle tentative
• errors-total : Nombre total d’erreurs rencontrées

• offset-commit-failures : Nombre d’échecs de commit des offsets
• offset-commit-avg-time-ms : Temps moyen des commits d’offset
• offset-commit-max-time-ms : Temps maximal des commits d’offset
• put-batch-avg-time-ms : Temps moyen de traitement d’un batch
• put-batch-max-time-ms : Temps maximal de traitement d’un batch
• source-record-poll-total : Nombre total d’enregistrements récupérés

​

Bonnes pratiques de supervision

1. Surveillez le consumer lag : suivez records-lag par partition afin d’identifier les goulots d’étranglement du traitement
2. Suivez les taux d’erreur : surveillez errors-total et records-skip-total pour détecter les problèmes de qualité des données
3. Vérifiez l’état des tâches : surveillez les métriques d’état des tâches pour vous assurer qu’elles s’exécutent correctement
4. Mesurez le débit : utilisez records-send-rate et byte-rate pour suivre les performances d’ingestion
5. Surveillez l’état des connexions : vérifiez les métriques de connexion au niveau des nœuds afin de détecter d’éventuels problèmes réseau
6. Suivez l’efficacité de la compression : utilisez compression-rate pour optimiser le transfert de données

​

Limitations

• Les suppressions ne sont pas prises en charge.
• La taille du lot est héritée des propriétés du consommateur Kafka.
• Lorsque vous utilisez KeeperMap pour exactly-once et que l’offset est modifié ou réinitialisé à une valeur antérieure, vous devez supprimer le contenu de KeeperMap pour ce topic spécifique. (Consultez le guide de dépannage ci-dessous pour plus de détails)

​

Réglage des performances et optimisation du débit

​

Quand l’optimisation des performances est-elle nécessaire ?

• Charges de travail à haut débit : lorsque vous traitez des millions d’événements par seconde depuis des topics Kafka
• Consumer lag : lorsque votre connecteur ne parvient pas à suivre le rythme de production des données, ce qui entraîne un décalage croissant
• Contraintes de ressources : lorsque vous devez optimiser l’utilisation du CPU, de la mémoire ou du réseau
• Topics multiples : lorsque vous consommez simultanément plusieurs topics à fort volume
• Petites tailles de message : lorsque vous traitez de nombreux petits messages qui bénéficieraient d’un regroupement côté serveur par lots

• Vous traitez des volumes faibles à modérés (< 10 000 messages/seconde)
• Le consumer lag est stable et acceptable pour votre cas d’usage
• Les paramètres par défaut du connecteur répondent déjà à vos exigences de débit
• Votre cluster ClickHouse peut facilement absorber la charge entrante

​

Comprendre le flux des données

1. Kafka Connect Framework récupère en arrière-plan les messages des topics Kafka
2. Le connecteur interroge le tampon interne du framework pour récupérer les messages
3. Le connecteur regroupe en lots les messages selon la taille du poll
4. ClickHouse reçoit l’insertion par lot via HTTP/S
5. ClickHouse traite l’insertion (de manière synchrone ou asynchrone)

​

Réglage de la taille des lots dans Kafka Connect

• fetch.min.bytes : quantité minimale de données avant que le framework ne transmette les valeurs au connecteur (par défaut : 1 octet)
• fetch.max.bytes : quantité maximale de données à récupérer dans une seule requête (par défaut : 52428800 / 50 MB)
• fetch.max.wait.ms : temps d’attente maximal avant le renvoi des données si fetch.min.bytes n’est pas atteint (par défaut : 500 ms)

• max.poll.records : Nombre maximal d’enregistrements renvoyés lors d’une seule opération de poll (par défaut : 500)
• max.partition.fetch.bytes : Quantité maximale de données par partition (par défaut : 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

• Lots plus volumineux = meilleures performances d’ingestion dans ClickHouse, moins de parts, moins de surcharge
• Lots plus volumineux = utilisation mémoire plus élevée, augmentation potentielle de la latence de bout en bout
• Lots trop volumineux = risque de dépassement de délai, d’erreurs OutOfMemory ou de dépassement de max.poll.interval.ms

​

Insertions asynchrones

• Nombreux petits lots : votre connecteur envoie fréquemment de petits lots (< 1000 lignes par lot)
• Concurrence élevée : plusieurs tâches du connecteur écrivent dans la même table
• Déploiement distribué : vous exécutez de nombreuses instances du connecteur sur différents hôtes
• Surcoût lié à la création de parts : vous rencontrez des erreurs « too many parts »
• Charge de travail mixte : vous combinez une ingestion en temps réel avec des charges liées aux requêtes

• Vous envoyez déjà de grands lots (> 10,000 lignes par lot) à une fréquence maîtrisée
• Vous avez besoin d’une visibilité immédiate des données (les requêtes doivent voir les données instantanément)
• La sémantique d’exactement une fois avec wait_for_async_insert=0 est incompatible avec vos exigences
• Votre cas d’usage peut plutôt tirer parti d’améliorations du batching côté client

1. Reçoit la requête d’insertion du connecteur
2. Écrit les données dans un tampon en mémoire (au lieu de les écrire immédiatement sur le disque)
3. Renvoie une confirmation de succès au connecteur (si wait_for_async_insert=0)
4. Vide le tampon sur le disque lorsqu’une de ces conditions est remplie :
   • Le tampon atteint async_insert_max_data_size (par défaut : 100 MB)
   • async_insert_busy_timeout_ms millisecondes se sont écoulées depuis le premier insert (par défaut : 1000 ms)
   • Le nombre maximal de requêtes accumulées est atteint (async_insert_max_query_number, par défaut : 100)

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

• async_insert=1 : Active les insertions asynchrones
• wait_for_async_insert=1 (recommandé) : Le connecteur attend que les données soient écrites dans le stockage ClickHouse avant d’envoyer un accusé de réception. Cela garantit la livraison.
• wait_for_async_insert=0 : Le connecteur envoie immédiatement un accusé de réception après la mise en mémoire tampon. Les performances sont meilleures, mais les données peuvent être perdues en cas de plantage du serveur avant leur écriture sur le stockage.

"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 (par défaut : 104857600 / 100 MB) : Taille maximale du tampon avant vidage
• async_insert_busy_timeout_ms (par défaut : 1000) : Temps maximal (ms) avant vidage
• async_insert_stale_timeout_ms (par défaut : 0) : Temps (ms) écoulé depuis la dernière insertion avant vidage
• async_insert_max_query_number (par défaut : 100) : Nombre maximal de requêtes avant vidage

• Avantages : Moins de parts, meilleures performances de fusion, surcharge CPU réduite, débit amélioré en cas de forte concurrence
• Considérations : Les données ne peuvent pas être interrogées immédiatement, latence de bout en bout légèrement plus élevée
• Risques : Perte de données en cas de plantage du serveur si wait_for_async_insert=0, risque de pression sur la mémoire avec de grands tampons

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

​

Parallélisme du connecteur

"tasks.max": "4"

• Nombre maximal de tâches réellement efficaces = nombre de partitions du topic
• Chaque tâche maintient sa propre connexion à ClickHouse
• Plus de tâches = plus de surcoût et un risque accru de contention sur les ressources

"ignorePartitionsWhenBatching": "true"

​

Plusieurs topics à haut débit

​

Considérations relatives au moteur de table ClickHouse

• MergeTree : idéal pour la plupart des cas d’utilisation, offre un bon équilibre entre les performances de requête et d’insertion
• ReplicatedMergeTree : requis pour la haute disponibilité, ajoute une surcharge liée à la réplication
• *MergeTree avec un ORDER BY approprié : optimisez en fonction de vos modèles de requêtes

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"

​

Pool de connexions et dépassements de délai

"jdbcConnectionProperties": "?socket_timeout=300000,connection_timeout=30000"

• socket_timeout (par défaut : 30000 ms) : Temps maximal pour les opérations de lecture
• connection_timeout (par défaut : 10000 ms) : Temps maximal pour établir une connexion

​

Tampon réseau du client

"jdbcConnectionProperties": "?client_network_buffer_size=1000000"

"jdbcConnectionProperties": "?ssl=true&client_network_buffer_size=1000000"

​

Surveillance et dépannage des performances

1. Consumer lag : utilisez les outils de monitoring Kafka pour suivre le lag par partition
2. Métriques du connecteur : surveillez receivedRecords, recordProcessingTime, taskProcessingTime via JMX (voir Monitoring)
3. Métriques ClickHouse :
   • system.asynchronous_inserts : surveillez l’utilisation du buffer des insertions asynchrones
   • system.parts : surveillez le nombre de parts pour détecter les problèmes de fusion
   • system.merges : surveillez les fusions actives
   • system.events : suivez InsertedRows, InsertedBytes, FailedInsertQuery

​

Résumé des bonnes pratiques

1. Commencez par les valeurs par défaut, puis mesurez et ajustez en fonction des performances réelles
2. Privilégiez des lots plus volumineux : visez 10,000-100,000 lignes par insertion lorsque c’est possible
3. Utilisez l’insertion asynchrone lorsque vous envoyez de nombreux petits lots ou en cas de forte concurrence
4. Utilisez toujours wait_for_async_insert=1 avec une sémantique exactly-once
5. Effectuez une mise à l’échelle horizontale : augmentez tasks.max jusqu’au nombre de partitions
6. Un connecteur par topic à fort volume pour un débit maximal
7. Surveillez en continu : suivez le consumer lag, le nombre de parts et l’activité de merge
8. Testez de manière approfondie : testez toujours les changements de configuration sous une charge réaliste avant un déploiement en production

​

Exemple : configuration à haut débit

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

• Traite jusqu’à 10 000 enregistrements par poll
• Regroupe les données de plusieurs partitions en lots pour des inserts plus volumineux
• Utilise l’insertion asynchrone avec un buffer de 16 Mo
• Exécute 8 tasks en parallèle (à aligner sur votre nombre de partitions)
• Optimisée pour le débit plutôt que pour un ordre strict

​

Dépannage

​

“Incohérence d’état pour le 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]

​

“Quelles erreurs le connecteur retentera-t-il ?”

• ClickHouseException - Il s’agit d’une exception générique pouvant être levée par ClickHouse. Elle est généralement levée lorsque le serveur est surchargé, et les codes d’erreur suivants sont considérés comme particulièrement transitoires :
  • 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 - Cette exception est levée lorsque le délai d’attente du socket est dépassé.
• UnknownHostException - Cette exception est levée lorsque l’hôte ne peut pas être résolu.
• IOException - Cette exception est levée en cas de problème réseau.

​

“Toutes mes données sont vides/à zéro”

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

​

“Je veux utiliser mes clés Kafka dans ClickHouse”

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