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長串連，一般情況是因為網路抖動造成的，用戶端有重試邏輯，因此不會對用戶端的消費造成影響。