DataHub:Compatibilidad con Kafka

Tipos de temas

Kafka y DataHub tienen mecanismos diferentes para escalar temas. Para garantizar la compatibilidad con el comportamiento de Kafka, debe establecer el modo de escalado en ONLY_EXTEND cuando cree un tema de DataHub. En este modo, solo puede agregar nuevos shards a un tema. Este modo no permite dividir ni fusionar shards, y aún no se admite la eliminación de shards.

Nomenclatura de temas

El nombre de un tema de Kafka se asigna a un proyecto y un tema de DataHub, separados por un punto (.). El mapeo sigue estas reglas:

• La parte anterior al primer . corresponde al proyecto de DataHub, y la parte posterior corresponde al tema de DataHub. Por ejemplo, test_project.test_topic se asigna al proyecto test_project y al tema test_topic.

• Si un nombre contiene varios caracteres ., solo el primer . actúa como separador. Todos los caracteres . y - restantes se reemplazan por _.

Particiones

Cada shard activo en DataHub corresponde a una partición en Kafka. Por ejemplo, si un tema tiene cinco shards activos, equivale a un tema de Kafka con cinco particiones. Al escribir datos, puede especificar un ID de partición en el rango [0, 4]. Si no especifica una partición, el cliente de Kafka asigna una automáticamente.

Temas Tuple

Cuando escribe datos desde Kafka hacia un tema Tuple, el esquema del tema debe tener una o dos columnas, ambas de tipo STRING. De lo contrario, la operación de escritura fallará.

• Si el esquema tiene una sola columna, solo se escribe el valor y se descarta la clave.

• Si el esquema tiene dos columnas, la primera y la segunda columna corresponden a la clave y al valor, respectivamente.

Además, no escriba datos binarios en un tema Tuple, ya que esto provoca caracteres ilegibles. Para almacenar datos binarios, utilice un tema Blob.

Temas Blob

Cuando escribe datos desde Kafka hacia un tema Blob, el valor del mensaje de Kafka se escribe en el campo Blob. Si la clave del mensaje no es NULL, se escribe como un atributo de DataHub. El nombre del atributo es __kafka_key__ y su valor es la clave del mensaje de Kafka.

Encabezados

Los encabezados de Kafka se asignan a atributos de DataHub. Si el valor de un encabezado es NULL, se ignora y no se escribe como atributo. Se recomienda no utilizar __kafka_key__ como clave de encabezado para evitar conflictos con el nombre de atributo integrado en los temas Blob.

Grupos de consumidores

En DataHub, un ID de suscripción actúa como un grupo de consumidores, pero solo puede suscribirse a un único tema. En cambio, un grupo de consumidores de Kafka puede suscribirse a varios temas simultáneamente. Para ofrecer compatibilidad con el modelo de suscripción de Kafka, DataHub ofrece una función de grupos. Puede crear un grupo dentro de un proyecto y vincularlo a varios temas, lo que le permite suscribirse a todos ellos bajo un mismo grupo.

Un grupo gestiona internamente múltiples suscripciones de DataHub en el servidor. Después de vincular un tema, el grupo crea automáticamente una suscripción, la cual aparece en la lista de suscripciones de la página de detalles del tema. No elimine esta suscripción manualmente. Hacerlo impedirá que el grupo se suscriba al tema y provocará la pérdida de todos los desplazamientos de consumo existentes.

Un solo grupo puede suscribirse a un máximo de 50 temas. Para suscribirse a más temas, envíe un ticket de soporte.

El archivo kafka_client_producer_jaas.conf

Cree un archivo llamado kafka_client_producer_jaas.conf en cualquier directorio y agregue el siguiente contenido.

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

Dependencia de Maven

La versión del cliente de Kafka debe ser la 0.10.0.0 o posterior. Se recomienda la versión 2.4.0.

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

Código de ejemplo

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();
        }
    }
}

Resultado

Una vez que el código se ejecute correctamente, puede muestrear datos para verificar el resultado.

Código de ejemplo

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());
            }
        }
    }
}

Resultado

Después de que el código se ejecute correctamente, podrá ver los datos consumidos en su 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!)

En este ejemplo, todos los registros de datos devueltos en una sola solicitud tienen el mismo LogAppendTime, que corresponde a la marca de tiempo más reciente entre todos los registros de ese lote.

Dependencia de 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>

Código de ejemplo

Este ejemplo lee datos de un tema de entrada dentro de test_project, convierte las cadenas de clave y valor a minúsculas, y escribe el resultado en un tema de salida.

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));
        }
    }
}

Resultado

Después de iniciar la tarea de Streams, la asignación de shards tarda aproximadamente un minuto. Posteriormente, podrá ver el número de tareas actuales en la consola. El número de tareas coincide con el número de shards del tema de entrada. En este ejemplo, el tema de entrada tiene tres shards.

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

Una vez asignados los shards, puede escribir datos de prueba como (AAAA,BBBB),(CCCC,DDDD),(EEEE,FFFF) en el tema de entrada. Luego, muestree datos del tema de salida para verificar que se escribieron correctamente.

Desconexión al escribir datos

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)
    ...

Las solicitudes de metadatos de Kafka y las solicitudes de escritura de datos utilizan conexiones diferentes.

El cliente establece primero una conexión para obtener metadatos. Luego utiliza la información del broker devuelta para establecer una segunda conexión destinada a la escritura de datos. Todas las solicitudes posteriores se envían a través de esta segunda conexión.

La primera conexión, ahora inactiva, se cierra automáticamente por el servidor tras un tiempo de espera. Esto puede generar un error de desconexión en los registros. Puede ignorar este error si los datos se están escribiendo correctamente.

Fallo al iniciar el cliente de 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

Agregue la siguiente propiedad a su configuración: properties.put("ssl.endpoint.identification.algorithm", "");.

DisconnectException durante el consumo

[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

El cliente de Kafka debe mantener una conexión TCP persistente con el servidor. Esta excepción suele deberse a inestabilidad en la red. El cliente cuenta con una lógica de reintento integrada, por lo que este error generalmente no afecta el consumo.