kafka exactly once 批处理

简介

这几天，有个大兄弟问，如何实现 kafka 多线程批消费，目标：

确保 exactly once 语义
数据不丢失
支持定时同步，如15分钟/30分钟等
支持多线程（kafka 实际上不支持，但是可以通过多个 groupid+offset 控制实现）

方案

上述目标1、2、3，简单实现可以通过存储消费对应的offset来处理，也就是进程启动时通过 seek 赋值，指定 partition 从哪个 offset 开始消费，此处使用zk存储
对于目标4，实现逻辑简单来说：
a. 找到 offset begin 与 offset end
b. 切分 offset 区间段，例如切分为5个：[0,5),[5,10),[10,15),[15,20),[20,23)
c. 根据上述起5个线程，消费指定区间的数据，由于 kafka 同个groupid，并行消费的话，会发生 rebalance，类似这样的日志，具体原理就不分析了（详细看下groupid）

image.png

d. 避免这个问题，我们可以模拟多 groupid 消费，一个 groupid 消费一段【不用原生API消费的话，可以直接用 spark struct streaming，支持类该模式】

核心代码

第一步，获取此时 kafka topic 信息，如拿到 topic partition 相关信息

consumer.subscribe(Collections.singletonList(kafkaTopic));
//        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
consumer.poll(Duration.ofSeconds(5));

// 获取offset信息
List<PartitionInfo> partitionInfos = consumer.partitionsFor(kafkaTopic);
List<TopicPartition> topicPartitions = new ArrayList<>();
for (PartitionInfo partitionInfo : partitionInfos) {
    TopicPartition partition = new TopicPartition(kafkaTopic, partitionInfo.partition());
    topicPartitions.add(partition);
}

Map<TopicPartition, Long> topicPartitionMap = consumer.endOffsets(topicPartitions, Duration.ofSeconds(MAX_TIME_AWAIT_END_OFFSET, 0));
Map<Integer, Long> partitionOffsetMap = new ConcurrentHashMap<>();
topicPartitionMap.forEach((key, value) -> partitionOffsetMap.put(key.partition(), value > 0 ? value - 1 : 0L));
LOG.info("end offsets: {}", partitionOffsetMap.toString());

第二步，找到缓存在 zk 的 partition offset 信息，并设置 partition 的 metadata

if (! isConsumeUserOffset) {
    for (TopicPartition partition : topicPartitions) {
        Long persistOffset = getPersistOffset(partition.partition());
        if (persistOffset <= 0) {
            persistOffset = consumer.position(partition);
        }
        LOG.info("get persist offset, partition num {}: {}", partition.partition(), persistOffset);
        // 启动时使用上次offset，消费下一次，所以 +1
        OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(persistOffset + 1);
        consumer.seek(partition, offsetAndMetadata);
    }
} else {
    if (autoOffsetReset.equals("earliest")) {
        LOG.info("auto.offset.reset={}", autoOffsetReset);
        consumer.seekToBeginning(topicPartitions);
    }
    if (autoOffsetReset.equals("latest")) {
        LOG.info("auto.offset.reset={}", autoOffsetReset);
        consumer.seekToEnd(topicPartitions);
    }
    if (autoOffsetReset.equals("None")) {
        // topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常， 忽略不用
        LOG.info("auto.offset.reset={}, it has so many problem, do nothing!", autoOffsetReset);
    }
}

第三部，按批消费，并存储该批次每个 partition 最后一个 offset

while (! isEndToConsume) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
    LOG.info("consumer records length: {}", records.count());
    List<String> writeRecords = new ArrayList<>();
    for (ConsumerRecord<String, String> record : records) {
        // 替换非法字符
        String value = record.value().replaceAll("\n|\t|\r", "\001");
        try {
            Map<String, Object> fieldMap = JSON.parseObject(value, new TypeReference<Map<String, Object>>() {
            });

            JSONObject jsonObject = new JSONObject(fieldMap);
            writeRecords.add(jsonObject.toJSONString());
            cachePartitionOffsetMap.put(record.partition(), record.offset());
            recordCount ++;

        } catch (Exception e) {
            // 非 JSON 数据时直接跳出
            LOG.info("record parse to json exception: {}, record: {}" , e.getMessage(), value);
//                    e.printStackTrace();
            continue;
        }
    }
 }

第四步，存储 partition offset 信息

// 判断是否消费到最后的offset
Map<Integer, Boolean> cacheHasConsumedPartitionMap = new ConcurrentHashMap<>();
if (! cachePartitionOffsetMap.isEmpty()) {
    LOG.info("cache partition offset map: {}", cachePartitionOffsetMap.toString());
    for(Integer partition : partitionOffsetMap.keySet()) {
        if (cachePartitionOffsetMap.containsKey(partition)) {
            long cacheOffset = cachePartitionOffsetMap.get(partition);
            // 存储最新 offset，作为程序退出依据
            if (cacheOffset >= partitionOffsetMap.get(partition)) {
                cacheHasConsumedPartitionMap.put(partition, true);
            }
        }
    }
    // 每个批次都存起来
    cachePartitionOffsetMap.forEach((key, value) -> setPersistOffset(key, value));
}
// 退出条件
if (cacheHasConsumedPartitionMap.size() == partitionOffsetMap.size()) {
    isEndToConsume = true;
}
// 程序触发提交
consumer.commitSync();

运行

java -cp /data/shopee/logtohdfs/logtohdfs-1.0-SNAPSHOT-jar-with-dependencies.jar com.xxx.bigdata.kafka.KafkaReader xxx_json.properties xxx

image.png

每个 partition 对应的 offset 都超过图一，程序开始初获取的offset，则退出这次调度

待完善

多线程按批处理（目标4），其实也就是，基于获取全局的 topic 信息，分拆生成多个新的 consumer，暂不实现。