kafka重置消费位点(史上最全)
kafka重置消费位点一般分几种情况重置到最新的消费位点重置到最早的消费位点根据时间戳重置消费位点跟据指定偏移量重置消费位点基于kafka 2.0.0package org.apache.dolphinscheduler.realtime.kafka;import lombok.extern.slf4j.Slf4j;import org.apache.commons.lang3.StringUti
·
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.*;
/**
* @author ljj
* @date 2021/4/25
*/
@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();
}
}
更多推荐
所有评论(0)