DataHub：兼容Kafka

Topic类型

Kafka 与 DataHub 的 Topic 扩缩容机制不同。创建 DataHub Topic 时，需将扩容模式设置为扩展模式以适配 Kafka 的行为。扩展模式下，Topic 不支持 Shard 的分裂与合并操作，仅支持增加 Shard，暂不支持减少 Shard。

Topic命名

Kafka Topic 映射为 DataHub 的 Project 和 Topic，两者以英文句号（.）分隔。映射规则如下：

• 以首个.为界，左侧为 DataHub Project，右侧为 DataHub Topic。例如 test_project.test_topic 映射为 Project test_project、Topic test_topic。

• 若名称中包含多个 .，仅首个 . 作为分隔符，其余 . 和 - 均替换为 _。

Partition

DataHub 中每个处于 Active 状态的 Shard 对应 Kafka 的 1 个 Partition。例如，当前有 5 个 Active Shard，则等效于 Kafka 有 5 个 Partition，写入时可指定 Partition 范围为 [0, 4]。未指定时，由 Kafka 客户端自动分配 Partition。

Tuple Topic

从 Kafka 向 Tuple Topic 写入数据时，Topic Schema 必须为 1 列或 2 列，且类型均为 STRING，否则写入失败。

• 列数为 1 时仅写入 Value，Key 数据将被丢弃；

• 列数为 2 时，第 1 列和第 2 列分别对应 Key 和 Value。

此外，Tuple Topic 不支持写入二进制数据，否则会出现乱码，二进制数据请写入 Blob Topic。

Blob Topic

从 Kafka 向 Blob Topic 写入数据时，Kafka 消息的 Value 写入 Blob 字段。如果消息的 Key 不为 NULL，则将其写入 DataHub 的 Attribute，属性名固定为 __kafka_key__，属性值为 Kafka 消息的 Key。

Header

Kafka 的 Header 对应 DataHub 的 Attribute。如果 Header 的 Value 为 NULL，则跳过该 Header，不写入 Attribute。建议不要使用 __kafka_key__ 作为 Header 的 Key，以避免与 Blob Topic 中的内置属性名冲突。

Consumer Group

DataHub 的消费组即订阅 ID，仅支持订阅单个 Topic；Kafka 的 Consumer Group 可同时订阅多个 Topic。为兼容 Kafka 的订阅方式，DataHub 提供了 Group 功能：用户可在 Project 下创建 Group 并绑定多个 Topic，通过该 Group 统一订阅同一 Project 下的多个 Topic。

Group 在服务端封装了多个 DataHub 订阅。绑定 Topic 后，Topic 页面的订阅列表中会出现由 Group 自动创建的订阅记录。请勿手动删除该订阅，否则 Group 将无法继续订阅该 Topic，且已有的消费点位会丢失。

单个 Group 最多支持订阅 50 个 Topic，如需订阅更多，请提交工单申请。

生成kafka_client_producer_jaas.conf文件

创建文件kafka_client_producer_jaas.conf，保存到任意路径，文件内容如下。

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

maven依赖

Kafka-client版本至少大于等于0.10.0.0，推荐2.4.0。

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

示例代码

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

运行结果

运行成功之后，抽样确认结果是否正确。

示例代码

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");
        // group.id填project.group
        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");
        // 使用kafka group可以同时订阅多个topic
        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");
        // group.id填project.topic.subId
        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);

        // 使用project.topic.subId的方式只能订阅单个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());
            }
        }
    }
}

运行结果

运行成功之后，便可以在终端看到读取到的数据。

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

本示例中同一个请求返回的数据的LogAppendTime是相同的，是该请求返回所有的数据的写入DataHub时间的最大值。

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>

代码示例

这里读取test_project下input的数据，将key和value的字符串转为小写重新写入output。

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

运行结果

启动Streams任务之后，分配shard大概需要1分钟左右，1分钟之后就可以在控制台看到当前的task数量，task数量和输入topic的shard数量保持一致，示例输入topic为3个shard。

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

shard分配成功之后，可以向input中写入一组测试数据 (AAAA,BBBB),(CCCC,DDDD),(EEEE,FFFF)，再output抽样查看数据是否正确写入。

写入数据时连接断开

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

Kafka 的元数据（Metadata）请求与数据写入请求使用不同的连接。

客户端首先建立连接获取元数据，再根据返回的 Broker 信息建立第二个连接用于写入数据，后续所有请求均通过第二个连接发送。

第一个连接因此处于闲置状态，服务端会在连接闲置超时后主动关闭，此时可能产生连接断开的错误日志。如果数据写入正常，可忽略该错误。

启动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

添加配置properties.put("ssl.endpoint.identification.algorithm", "");。

Consumer消费过程中出现DisconnectException

[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

Kafka的客户端需要与服务端保持TCP长连接，一般情况是因为网络抖动造成的，客户端有重试逻辑，因此不会对客户端的消费造成影响。