Dépannage du connecteur Kafka¶

Étape 1 : Affichage de l’historique COPY de la table¶

Interrogez l’historique des activités de chargement pour la table cible. Pour plus d’informations, voir Vue COPY_HISTORY. Si la sortie COPY_HISTORY n’inclut pas un ensemble de fichiers attendus, interrogez une période antérieure. Si les fichiers étaient des doublons d’anciens fichiers, l’historique de chargement aurait peut-être enregistré l’activité lors de la tentative de chargement des fichiers d’origine. La colonne STATUS indique si un ensemble particulier de fichiers a été chargé, partiellement chargé ou non. La colonne FIRST_ERROR_MESSAGE fournit une raison lorsqu’une tentative est partiellement chargée ou échouée.

Le connecteur Kafka déplace les fichiers qu’il n’a pas pu charger vers la zone de préparation associée à la table cible. La syntaxe pour référencer une zone de préparation de table est @[namespace.]%table_name.

Répertoriez tous les fichiers situés dans la zone de préparation de la table avec LIST.

Par exemple :

LIST @mydb.public.%mytable;

Les noms de fichiers sont dans l’un des formats suivants. Les conditions qui produisent chaque format sont décrites dans le tableau :

Les sections suivantes fournissent des instructions pour résoudre les problèmes liés à chacun des types de fichiers :

Étape 2 : Analyse du fichier journal du connecteur Kafka¶

Si la vue COPY_HISTORY ne contient aucun enregistrement du chargement de données, analysez le fichier journal du connecteur Kafka. Le connecteur écrit des événements dans le fichier journal. Notez que le connecteur Snowflake Kafka partage le même fichier journal avec tous les plug-ins du connecteur Kafka. Le nom et l’emplacement de ce fichier journal doivent figurer dans votre fichier de configuration du connecteur Kafka. Pour plus d’informations, consultez la documentation fournie avec votre logiciel Apache Kafka.

Recherchez dans le fichier journal du connecteur Kafka les messages d’erreur liés à Snowflake. Beaucoup de ces messages porteront la chaîne ERROR et contiendront le nom de fichier com.snowflake.kafka.connector... pour faciliter la recherche de ces messages.

Les erreurs possibles que vous pourriez rencontrer incluent :

Erreur de configuration:

Causes possibles de l’erreur :

• Le connecteur ne dispose pas des informations appropriées pour s’abonner au sujet.

• Le connecteur ne dispose pas des informations appropriées pour écrire dans la table Snowflake (la paire de clés pour l’authentification peut être incorrecte, par ex.).

Notez que le connecteur Kafka valide ses paramètres. Le connecteur génère une erreur pour chaque paramètre de configuration incompatible. Le message d’erreur est écrit dans le fichier journal du cluster Kafka Connect. Si vous suspectez un problème de configuration, vérifiez les erreurs dans ce fichier journal.

Erreurs de lecture:

Le connecteur n’a peut-être pas été en mesure de lire Kafka pour les raisons suivantes :

• Kafka ou Kafka Connect peuvent ne pas être en cours d’exécution.

• Le message n’a peut-être pas encore été envoyé.

• Le message a peut-être été supprimé (expiré).

Erreur d’écriture (zone de préparation):

Causes possibles de l’erreur :

• Privilèges insuffisants sur la zone de préparation.

• La zone de préparation est hors cadre.

• La zone de préparation a été détruite.

• Un autre utilisateur ou processus a écrit des fichiers inattendus sur la zone de préparation.

Erreur d’écriture (table):

Causes possibles de l’erreur :

• Privilèges insuffisants sur la table.

Étape 3 : vérifier Kafka Connect¶

Si aucune erreur n’est signalée dans le fichier journal de Kafka Connect, vérifiez Kafka Connect. Pour obtenir des instructions de dépannage, consultez la documentation fournie par votre fournisseur de logiciels Apache Kafka.

Lignes en double avec la même partition de rubrique et le même décalage¶

Lors du chargement de données à l’aide de la version 1.4 du connecteur Kafka (ou supérieur), les lignes en double dans la table cible avec la même partition de sujet et le même décalage peuvent indiquer que l’opération de chargement a dépassé le délai d’exécution par défaut de 300000 millisecondes (300 secondes). Pour vérifier la cause, consultez le fichier journal de Kafka Connect pour l’erreur suivante :

org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member.

This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.

at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.sendOffsetCommitRequest(ConsumerCoordinator.java:1061)

Pour résoudre l’erreur, dans le fichier de configuration de Kafka (par exemple <kafka_dir>/config/connect-distributed.properties), modifiez l’une des propriétés suivantes :

consumer.max.poll.interval.ms

Augmentez le délai d’exécution à 900000 (900 secondes).

consumer.max.poll.records

Diminuez le nombre d’enregistrements chargés avec chaque opération à 50.

Échec de réponse de migration du décalage du canal Streaming - Code d’erreur : 5023¶

Lors de la mise à niveau vers la version du connecteur v2.1.0 (ou supérieure), un changement a été introduit dans le format du nom du canal Snowpipe Streaming. Par conséquent, la logique de détection d’informations sur les décalages précédemment validés ne trouvera aucune information sur ceux précédemment validés. Cela se manifestera par l’exception suivante :

com.snowflake.kafka.connector.internal.SnowflakeKafkaConnectorException: [SF_KAFKA_CONNECTOR] Exception: Failure in Streaming Channel Offset Migration Response Error Code: 5023

Detail: Streaming Channel Offset Migration from Source to Destination Channel has no/invalid response, please contact Snowflake Support

Message: Snowflake experienced a transient exception, please retry the migration request.

Pour résoudre cette erreur, dans le fichier de configuration Kafka (par exemple, <kafka_dir>/config/connect-distributed.properties), ajoutez la propriété de configuration suivante :

enable.streaming.channel.offset.migration

Désactivez la migration automatique des décalages en la définissant sur false.

Configuration du connecteur pour prendre en charge plusieurs rubriques¶

Nous avons rencontré un problème avec une seule instance de connecteur Kafka prenant en charge un grand nombre de rubriques, chacune ayant plusieurs partitions. La configuration du connecteur, même si elle semblait valide, a entraîné un cycle de rééquilibrage sans fin sans possibilité d’ingérer des données dans Snowflake. Le problème était propre au mode d’ingestion de Snowpipe Streaming (snowflake.ingestion.method=SNOWPIPE_STREAMING), mais les directives s’appliquent également au mode d’ingestion Snowpipe (snowflake.ingestion.method=SNOWPIPE). Le problème se manifeste dans le fichier journal par l’enregistrement à plusieurs reprises de ce message de journal :

[Worker-xyz] [timestamp] INFO [my-connector|task-id] [SF_INGEST] Channel is marked as closed

Cela peut généralement se produire lorsque vous configurez votre connecteur pour l’ingestion de rubriques via Regex. Nous vous recommandons d’appliquer l’ensemble d’options suivant au fichier de configuration Kafka (par exemple, <kafka_dir>/config/connect-distributed.properties) :

consumer.override.partition.assignment.strategy

Configurez la stratégie d’affectation de partition aux tâches en tant que org.apache.kafka.clients.consumer.CooperativeStickyAssignor - cela entraînera une distribution uniforme des canaux ingérés entre les tâches disponibles, réduisant ainsi le risque de rééquilibrage.

tasks.max

Le nombre de tâches instanciées par connecteur ne doit pas dépasser le nombre de CPU disponibles - le pilote sous-jacent implémente un mécanisme de limitation en fonction des CPU disponibles. L’augmentation du nombre de requêtes simultanées entraînera une pression accrue sur la mémoire de votre système, mais entraînera également des temps de traitement d’insertion plus longs, conduisant directement à l’absence de pulsations du connecteur.

Lorsqu’on parle de valeurs de délai d’expiration du connecteur, il existe un ensemble de propriétés de configuration qui les affectent directement :

consumer.override.heartbeat.interval.ms

Définit la fréquence à laquelle le thread du moniteur (un moniteur est associé à chaque tâche) enverra une pulsation à Kafka. La valeur par défaut est 3000 ms, mais en cas de charge système plus élevée, vous pouvez essayer de l’augmenter à 5000 ms.

consumer.override.session.timeout.ms

Définit la durée pendant laquelle le courtier attendra avant de supposer que le consommateur est dans un état non valide et de tenter un rééquilibrage. Ce paramètre doit être généralement 3 fois supérieur à l’intervalle de pulsation. Par conséquent, si vous avez configuré la pulsation sur 5000 ms, définissez ce paramètre sur 15000 ms.

consumer.override.max.poll.interval.ms

Définit l’intervalle maximal entre l’appel à poll() et le connecteur Kafka sous-jacent. Le temps passé entre les interrogations correspond essentiellement au lot de données traité par le connecteur (y compris le téléchargement vers Snowflake et la validation). Dans les scénarios où vous avez plusieurs tâches de traitement des données, la connexion Snowflake sous-jacente peut commencer à limiter les requêtes, ce qui entraîne des délais de traitement plus longs. Selon votre scénario, vous pouvez augmenter cette valeur jusqu’à 20 minutes (1200000 ms), en particulier lorsque vous démarrez le connecteur avec un grand nombre d’enregistrements initiaux à ingérer.

consumer.override.rebalance.timeout.ms

Lorsqu’un rééquilibrage se produit, dans un scénario avec un grand nombre de canaux par tâche, il existe de nombreuses logiques sous-jacentes par canal pour déterminer où reprendre le traitement. Ce code est exécuté de manière séquentielle. Par conséquent, plus il y a de canaux par tâche et plus la configuration initiale durera. Configurez cette propriété sur une valeur suffisamment grande pour permettre à chaque canal de réaliser son initialisation. Une valeur de 3 minutes (180000 ms) est un bon point de départ.

Il est également important de connaître la mémoire de tas disponible pour le connecteur. Ceci est particulièrement important dans les scénarios où plusieurs connecteurs s’exécutent simultanément ou lorsqu’un connecteur ingère des données provenant de plusieurs rubriques. La partition de chaque rubrique correspond à un seul canal et, par conséquent, nécessite de la mémoire.

Assurez-vous d’ajuster les paramètres de mémoire de votre processus de connexion Kafka via le paramètre Xmx. Une façon d’y parvenir est de définir la variable d’environnement KAFKA_OPTS en conséquence (c’est-à-dire, KAFKA_OPTS=-Xmx4G).

Purge inattendue des fichiers par le nettoyeur de fichiers¶

Lors de l’utilisation du connecteur Kafka avec SNOWPIPE, vous pourriez rencontrer un problème au moment de l’ingestion des données dans une seule table à partir de plusieurs rubriques. Si votre configuration ne dispose pas de l’entrée snowflake.topic2table.map ou s’il existe un mappage 1:1 entre la rubrique et la table, ce problème ne surviendra pas.

Le connecteur Kafka génère des fichiers avec des enregistrements à charger vers une zone de préparation. Ces fichiers sont formatés selon le modèle suivant : snowflake_kafka_connector_<connector-name>_stage_<table-name>/<connector-name>/<table-name>/<partition-id>/<low-watermark>_<high-watermark>_<timestamp>.json.gz. Le problème se situe dans le <partition-id> : si plusieurs rubriques chargent des données dans une seule table, des doublons sont susceptibles d’apparaître dans la valeur partition-id. Cela ne constitue pas un problème dans le cadre d’un fonctionnement normal du connecteur. Cependant, si le connecteur redémarre ou se rééquilibre, le processus de nettoyage peut associer de manière inexacte les fichiers chargés dans la zone de préparation (mais pas encore ingérés) à la mauvaise partition, et décider de les supprimer, ce qui peut entraîner une perte de données.

Le connecteur avec la version 2.4.x corrige ce problème en ajoutant le code de hachage de la rubrique source à l”partition-id pour garantir des noms de fichiers uniques qui correspondent exactement à la partition d’une rubrique unique. Ce correctif est activé par défaut - snowflake.snowpipe.stageFileNameExtensionEnabled - et concerne uniquement les configurations où une table cible est répertoriée plusieurs fois dans snowflake.topic2table.map.

Si votre configuration est affectée par cette fonctionnalité, vous risquez de vous retrouver avec des fichiers obsolètes chargés dans votre zone de préparation. Lorsque le connecteur démarre, il vérifie si votre zone de préparation contient de tels fichiers. Vous devez rechercher les entrées de journal commençant par NOTE: For table, suivi de la liste des fichiers détectés.

Vous pouvez également vérifier si certains fichiers sont affectés à la zone de préparation manuellement :

1. Recherchez la zone de préparation concernée :

   show stages like 'snowflake_kafka_connector%<your table name>';
   

   Copy

2. Répertoriez les fichiers en zone de préparation :

   list @<your stage name> pattern = '.+/<your-table-name>/[0-9]{1,4}/[0-9]+_[0-9]+_[0-9]+\.json\.gz$';
   

   Copy

La commande ci-dessus répertorie tous les fichiers correspondant à la zone de préparation de votre table et ayant des IDs de partition compris entre 0 et 9999. Ces fichiers ne seront plus ingérés, vous pouvez donc les télécharger ou les supprimer.