DataHub:Compatibilité Kafka

Types de topic

Kafka et DataHub disposent de mécanismes différents pour la mise à l'échelle des topics. Pour garantir la compatibilité avec le comportement de Kafka, définissez le mode de mise à l'échelle sur ONLY_EXTEND lors de la création d'un topic DataHub. Dans ce mode, vous pouvez uniquement ajouter de nouveaux shards à un topic. Ce mode n'autorise ni le fractionnement ni la fusion des shards, et la suppression de shards n'est pas encore prise en charge.

Nommage des topics

Un nom de topic Kafka correspond à un projet et un topic DataHub, séparés par un point (.). Le mappage suit les règles suivantes :

• La partie située avant le premier . représente le projet DataHub, et la partie suivante représente le topic DataHub. Par exemple, test_project.test_topic correspond au projet test_project et au topic test_topic.

• Si un nom contient plusieurs caractères ., seul le premier . sert de séparateur. Tous les autres caractères . et - sont remplacés par _.

Partitions

Chaque shard actif dans DataHub correspond à une partition dans Kafka. Par exemple, si un topic comporte cinq shards actifs, il équivaut à un topic Kafka doté de cinq partitions. Lors de l'écriture des données, spécifiez un ID de partition compris dans la plage [0, 4]. Si vous ne spécifiez aucune partition, le client Kafka en attribue une automatiquement.

Topic Tuple

Lorsque vous écrivez des données depuis Kafka vers un topic Tuple, le schéma du topic doit comporter une ou deux colonnes, toutes deux de type STRING. Dans le cas contraire, l'opération d'écriture échoue.

• Si le schéma comporte une seule colonne, seule la valeur est écrite et la clé est ignorée.

• Si le schéma comporte deux colonnes, la première et la deuxième colonne correspondent respectivement à la clé et à la valeur.

De plus, n'écrivez pas de données binaires dans un topic Tuple, car cela génère des caractères illisibles. Pour stocker des données binaires, utilisez un topic Blob.

Topic Blob

Lorsque vous écrivez des données depuis Kafka vers un topic Blob, la valeur du message Kafka est écrite dans le champ Blob. Si la clé du message n'est pas NULL, elle est écrite en tant qu'attribut DataHub. Le nom de l'attribut est __kafka_key__ et sa valeur correspond à la clé du message Kafka.

En-têtes

Les en-têtes Kafka correspondent aux attributs DataHub. Si la valeur d'un en-tête est NULL, elle est ignorée et n'est pas écrite en tant qu'attribut. Nous vous recommandons de ne pas utiliser __kafka_key__ comme clé d'en-tête afin d'éviter les conflits avec le nom d'attribut intégré dans les topics Blob.

Groupes de consommateurs

Dans DataHub, un ID d'abonnement fait office de groupe de consommateurs, mais il ne peut s'abonner qu'à un seul topic. En revanche, un groupe de consommateurs Kafka peut s'abonner simultanément à plusieurs topics. Pour assurer la compatibilité avec le modèle d'abonnement de Kafka, DataHub propose une fonctionnalité de groupe. Créez un groupe au sein d'un projet et liez-le à plusieurs topics, ce qui vous permet de vous abonner à tous ces topics via un seul groupe.

Un groupe gère en interne plusieurs abonnements DataHub sur le serveur. Après avoir lié un topic, le groupe crée automatiquement un abonnement, qui apparaît dans la liste des abonnements sur la page de détails du topic. Ne supprimez pas cet abonnement manuellement. Cela empêcherait le groupe de s'abonner au topic et entraînerait la perte de tous les offsets de consommation existants.

Un seul groupe peut s'abonner à un maximum de 50 topics. Pour vous abonner à davantage de topics, soumettez un ticket.

Fichier kafka_client_producer_jaas.conf

Créez un fichier nommé kafka_client_producer_jaas.conf dans n'importe quel répertoire et ajoutez le contenu suivant.

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="yourAccessKeyId"
  password="yourAccessKeySecret";
};

Dépendance Maven

La version du client Kafka doit être 0.10.0.0 ou ultérieure. Nous vous recommandons la version 2.4.0.

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.4.0</version>
</dependency>

Exemple de code

public class ProducerExample {
    static {
        System.setProperty("java.security.auth.login.config", "src/main/resources/kafka_client_producer_jaas.conf");
    }

    public static void main(String[] args) {

        Properties properties = new Properties();
        properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "PLAIN");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("compression.type", "lz4");

        String KafkaTopicName = "test_project.test_topic";
        Producer<String, String> producer = new KafkaProducer<String, String>(properties);

        try {
            List<Header> headers = new ArrayList<>();
            RecordHeader header1 = new RecordHeader("key1", "value1".getBytes());
            RecordHeader header2 = new RecordHeader("key2", "value2".getBytes());
            headers.add(header1);
            headers.add(header2);

            ProducerRecord<String, String> record = new ProducerRecord<>(KafkaTopicName, 0, "key", "Hello DataHub!", headers);

            // Sync send
            producer.send(record).get();

        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }
}

Résultat

Après l'exécution réussie du code, prélevez des données pour vérifier le résultat.

Exemple de code

package com.aliyun.datahub.kafka.demo;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
public class ConsumerExample2 {

    static {
        System.setProperty("java.security.auth.login.config", "src/main/resources/kafka_client_producer_jaas.conf");
    }

    public static void main(String[] args) {

        Properties properties = new Properties();
        properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "PLAIN");
        // Set group.id to the project.group format.
        properties.put("group.id", "test_project.test_kafka_group");
        properties.put("auto.offset.reset", "earliest");
        properties.put("session.timeout.ms", "60000");
        properties.put("heartbeat.interval.ms", "40000");
        properties.put("ssl.endpoint.identification.algorithm", "");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);

        List<String> topicList = new ArrayList<>();
        topicList.add("test_project.test_topic1");
        topicList.add("test_project.test_topic2");
        topicList.add("test_project.test_topic3");
        // By using a Kafka group, you can subscribe to multiple topics.
        kafkaConsumer.subscribe(topicList);

        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(5));

            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record.toString());
            }
        }
    }
}

package com.aliyun.datahub.kafka.demo;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class ConsumerExample {

    static {
        System.setProperty("java.security.auth.login.config", "src/main/resources/kafka_client_producer_jaas.conf");
    }

    public static void main(String[] args) {

        Properties properties = new Properties();
        properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "PLAIN");
        // Set group.id to the project.topic:subId format.
        properties.put("group.id", "test_project.test_topic:1611039998153N71KM");
        properties.put("auto.offset.reset", "earliest");
        properties.put("session.timeout.ms", "60000");
        properties.put("heartbeat.interval.ms", "40000");
        properties.put("ssl.endpoint.identification.algorithm", "");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);

        // When using the project.topic:subId format, you can only subscribe to a single topic.
        kafkaConsumer.subscribe(Collections.singletonList("test_project.test_topic"));

        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(5));

            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record.toString());
            }
        }
    }
}

Résultat

Après l'exécution réussie du code, vous pouvez voir les données consommées dans votre terminal.

ConsumerRecord(topic = test_project.test_topic, partition = 0, leaderEpoch = 0, offset = 0, LogAppendTime = 1611040892661, serialized key size = 3, serialized value size = 14, headers = RecordHeaders(headers = [RecordHeader(key = key1, value = [118, 97, 108, 117, 101, 49]), RecordHeader(key = key2, value = [118, 97, 108, 117, 101, 50])], isReadOnly = false), key = key, value = Hello DataHub!)

Dans cet exemple, tous les enregistrements de données renvoyés dans une seule requête partagent le même LogAppendTime, qui correspond à l'horodatage le plus récent parmi tous les enregistrements de ce lot.

Dépendance Maven

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.4.0</version>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>2.4.0</version>
</dependency>

Exemple de code

Cet exemple lit les données d'un topic d'entrée au sein de test_project, convertit les chaînes de clés et de valeurs en minuscules, et écrit le résultat dans un topic de sortie.

public class StreamExample {

    static {
        System.setProperty("java.security.auth.login.config", "src/main/resources/kafka_client_producer_jaas.conf");
    }

    public static void main(final String[] args) {
        final String input = "test_project.input";
        final String output = "test_project.output";
        final Properties properties = new Properties();
        properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
        properties.put("application.id", "test_project.input:1611293595417QH0WL");
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "PLAIN");
        properties.put("session.timeout.ms", "60000");
        properties.put("heartbeat.interval.ms", "40000");
        properties.put("auto.offset.reset", "earliest");

        final StreamsBuilder builder = new StreamsBuilder();
        TestMapper testMapper = new TestMapper();
        builder.stream(input, Consumed.with(Serdes.String(), Serdes.String()))
                .map(testMapper)
                .to(output, Produced.with(Serdes.String(), Serdes.String()));

        final KafkaStreams streams = new KafkaStreams(builder.build(), properties);
        final CountDownLatch latch = new CountDownLatch(1);

        Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
            @Override
            public void run() {
                streams.close();
                latch.countDown();
            }
        });

        try {
            streams.start();
            latch.await();
        } catch (final Throwable e) {
            System.exit(1);
        }
        System.exit(0);
    }

    static class TestMapper implements KeyValueMapper<String, String, KeyValue<String, String>> {

        @Override
        public KeyValue<String, String> apply(String s, String s2) {
            return new KeyValue<>(StringUtils.lowerCase(s), StringUtils.lowerCase(s2));
        }
    }
}

Résultat

Après le démarrage de la tâche Streams, l'attribution des shards prend environ une minute. Ensuite, vous pouvez voir le nombre de tâches actuelles dans la console. Le nombre de tâches correspond au nombre de shards dans le topic d'entrée. Dans cet exemple, le topic d'entrée comporte trois shards.

currently assigned active tasks: [0_0, 0_1, 0_2]
currently assigned standby tasks: []
revoked active tasks: []
  revoked standby tasks: []

Une fois les shards attribués, écrivez des données de test telles que (AAAA,BBBB),(CCCC,DDDD),(EEEE,FFFF) dans le topic d'entrée. Ensuite, prélevez des données du topic de sortie pour vérifier qu'elles ont été correctement écrites.

Déconnexion lors de l'écriture des données

Selector - [Producer clientId=producer-1] Connection with dh-cn-shenzhen.aliyuncs.com disconnected
java.io.EOFException
    at org.apache.kafka.common.network.SslTransportLayer.read(SslTransportLayer.java:573)
    ...

Les requêtes de métadonnées Kafka et les requêtes d'écriture de données utilisent des connexions différentes.

Le client établit d'abord une connexion pour récupérer les métadonnées. Il utilise ensuite les informations de broker renvoyées pour établir une seconde connexion destinée à l'écriture des données. Toutes les requêtes suivantes sont envoyées via cette seconde connexion.

La première connexion, désormais inactive, est automatiquement fermée par le serveur après un délai d'expiration. Cela peut générer une erreur de déconnexion dans les journaux. Vous pouvez ignorer cette erreur si les données sont écrites avec succès.

Échec du démarrage du client Kafka

Caused by: org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed
Caused by: javax.net.ssl.SSLHandshakeException: No subject alternative names matching IP address 100.67.134.161 found

Ajoutez la propriété suivante à votre configuration : properties.put("ssl.endpoint.identification.algorithm", "");.

DisconnectException lors de la consommation

[INFO][Consumer clientId=client-id, groupId=consumer-project.topic:subid] Error sending fetch request (sessionId=INVALID, epoch=INITIAL) to node 1: {}.
org.apache.kafka.common.errors.DisconnectException

Le client Kafka doit maintenir une connexion TCP persistante avec le serveur. Cette exception est généralement causée par des fluctuations réseau. Le client dispose d'une logique de nouvelle tentative intégrée, de sorte que cette erreur n'affecte généralement pas la consommation.