kafka重置消费位点

kafka重置消费位点一般分几种情况

• 重置到最新的消费位点
• 重置到最早的消费位点
• 根据时间戳重置消费位点
• 跟据指定偏移量重置消费位点

基于kafka 2.0.0

package com.realtime.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.dolphinscheduler.realtime.db.KafkaDataSource;
import org.apache.dolphinscheduler.realtime.dto.ConsumerRecordDtoNew;
import org.apache.dolphinscheduler.realtime.dto.KafkaConsumerDtoNew;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;


@Slf4j
public class KafkaConsumer extends KafkaBase {

    private KafkaConsumerDtoNew kafkaConsumerDto;

    private Consumer<String, String> consumer;

    private String consumerGroup;

    public KafkaConsumer(KafkaDataSource kafkaDataSource, KafkaConsumerDtoNew kafkaConsumerDto) {
        super(kafkaDataSource);
        this.kafkaConsumerDto = kafkaConsumerDto;
        if (StringUtils.isNotBlank(kafkaConsumerDto.getGroupId())) {
            this.consumerGroup = kafkaConsumerDto.getGroupId();
        } else {
            this.consumerGroup = "consumer-" + kafkaConsumerDto.getLoginUserName() + "-" + ZonedDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss.SSS"));
        }

        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaDataSource.getBootstrapServers());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, kafkaDataSource.getKeyDeserializer());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, kafkaDataSource.getValueDeserializer());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, this.consumerGroup);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaConsumerDto.getOffset());//earliest,latest
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "10000");
        consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<>(props);
    }

    //跟据指定偏移量重置消费位点
    public void resetOffsetByTopicPartitionOffset(Map<TopicPartition, Long> partitionLongMap) {
        Map<TopicPartition, OffsetAndMetadata> offset = new HashMap<>();
        if (null != partitionLongMap && partitionLongMap.size() > 0) {
            for (Map.Entry<TopicPartition, Long> p : partitionLongMap.entrySet()) {
                offset.put(p.getKey(), new OffsetAndMetadata(p.getValue()));
            }
        }
        consumer.commitSync(offset);
    }
    //重置到最新的消费位点
    public void resetOffsetToEnd() {
        Map<TopicPartition, OffsetAndMetadata> offset = new HashMap<>();
        List<PartitionInfo> partitionInfos = consumer.partitionsFor(kafkaConsumerDto.getTopic());
        if (null != partitionInfos && partitionInfos.size() > 0) {
            for (PartitionInfo p : partitionInfos) {
                consumer.assign(Collections.singleton(new TopicPartition(p.topic(), p.partition())));
                //移动到最新offset
                consumer.seekToEnd(Collections.singleton(new TopicPartition(p.topic(), p.partition())));
                //移动到最早offset
                //consumer.seekToBeginning(Collections.singleton(new TopicPartition(p.topic(), p.partition())));
                //获取到该分区的last offset
                long position = consumer.position(new TopicPartition(p.topic(), p.partition()));
                offset.put(new TopicPartition(p.topic(), p.partition()), new OffsetAndMetadata(position));
            }
        }
        consumer.commitSync(offset);
    }

    //重置到最早的消费位点
    public void resetOffsetToBeginning() {
        Map<TopicPartition, OffsetAndMetadata> offset = new HashMap<>();
        List<PartitionInfo> partitionInfos = consumer.partitionsFor(kafkaConsumerDto.getTopic());
        if (null != partitionInfos && partitionInfos.size() > 0) {
            for (PartitionInfo p : partitionInfos) {
                consumer.assign(Collections.singleton(new TopicPartition(p.topic(), p.partition())));
                //移动到最新offset
//                consumer.seekToEnd(Collections.singleton(new TopicPartition(p.topic(), p.partition())));
                //移动到最早offset
                consumer.seekToBeginning(Collections.singleton(new TopicPartition(p.topic(), p.partition())));
                //获取到该分区的last offset
                long position = consumer.position(new TopicPartition(p.topic(), p.partition()));
                offset.put(new TopicPartition(p.topic(), p.partition()), new OffsetAndMetadata(position));
            }
        }
        consumer.commitSync(offset);
    }
    //根据时间戳重置消费位点
    public void resetOffsetByTimestamps(long timestampMs) {
        Map<TopicPartition, OffsetAndMetadata> offset = new HashMap<>();
        /*这两个方法需要绑定使用，否则consumer.assignment()获取的数据为空
        consumer.assign(Arrays.asList(new TopicPartition("t7", 2)));
        Set<TopicPartition> partitionInfos = consumer.assignment();*/
        List<PartitionInfo> partitionInfos = consumer.partitionsFor(kafkaConsumerDto.getTopic());
        if (null != partitionInfos && partitionInfos.size() > 0) {
            Map<TopicPartition, Long> map = new HashMap<>();
            for (PartitionInfo p : partitionInfos) {
                map.put(new TopicPartition(p.topic(), p.partition()), timestampMs);
            }
            Map<TopicPartition, OffsetAndTimestamp> offsetTimestamp = consumer.offsetsForTimes(map);
            for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : offsetTimestamp.entrySet()) {
                TopicPartition key = entry.getKey();
                OffsetAndTimestamp value = entry.getValue();
                //根据消费里的timestamp确定offset
                long position = 0;
                if (value != null) {
                    position = value.offset();
                } else {
                    //当指定时间戳大于最分区最新数据时间戳时，为null
                    consumer.assign(Collections.singleton(key));
                    consumer.seekToEnd(Collections.singleton(key));
                    position = consumer.position(key);
                }
                offset.put(key, new OffsetAndMetadata(position));
                //以下是从指定offset开始消费
                //consumer.seek(entry.getKey(), position);
            }
        }
        consumer.commitSync(offset);
    }

    @Override
    public void close() {
        consumer.close();
    }

}