温馨提示×

温馨提示×

您好，登录后才能下订单哦！

密码登录×
登录注册×
获取短信验证码
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》
登 录 注册有礼
云服务器 香港服务器 高防服务器
最新更新 网站标签 地图导航
产品

Kafka2.7是如何重设消费者组位移

发布时间：2021-10-12 09:10:38 来源：亿速云 阅读：181 作者：iii 栏目：编程语言

本篇内容介绍了“Kafka2.7是如何重设消费者组位移”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！

重设消费

Earliest

首先看看重置位移前的消费进度

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --describe

Kafka2.7是如何重设消费者组位移
根据进度截图，能看到所有分区的Lag均为0，说明消息已经被消费完，现在根据Earliest策略重置消费进度，要求重置后所有的消息均可重新消费。

脚本命令方式

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic --to-earliest --execute

此时再度查看消费进度，可以看到 Kafka2.7是如何重设消费者组位移 此时消费者可以重新消费这些消息。

脚本命令方式(指定分区)

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic:1,2 --to-earliest --execute 

Java API方式

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

final String topic = "mytopic";
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Arrays.asList(topic));
    consumer.poll(0);
    Collection<TopicPartition> partitions = consumer.partitionsFor(topic).stream()
            .map(partitionInfo -> new TopicPartition(topic, partitionInfo.partition()))
            .collect(Collectors.toList());
    consumer.seekToBeginning(partitions);
    consumer.partitionsFor(topic).forEach(i -> consumer.position(new TopicPartition(topic, i.partition())));
}

需要特殊说明的是，seekToBeginning、seekToEnd等方法执行完需要执行position才会立刻生效
Kafka2.7是如何重设消费者组位移

Java API方式(指定分区)

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

final String topic = "mytopic";
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Arrays.asList(topic));
    consumer.poll(0);
    List<TopicPartition> partitions = new ArrayList<TopicPartition>();
    partitions.add(new TopicPartition(topic, 1));
    partitions.add(new TopicPartition(topic, 2));
    consumer.seekToBeginning(partitions);
    consumer.position(new TopicPartition(topic, 1));
    consumer.position(new TopicPartition(topic, 2));
} 

Latest

首先看看重置位移前的消费进度。
Kafka2.7是如何重设消费者组位移
根据上图可以看到，kafka当前没有任何消息被消费，现在根据Latest策略重置消费进度，要求重置后原消息不再消费。

脚本命令

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic --to-latest --execute

重置后
Kafka2.7是如何重设消费者组位移

脚本命令(指定分区)

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic:1,2 --to-latest --execute 

Java API

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");

final String topic = "mytopic";
try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Arrays.asList(topic));
    consumer.poll(0);
    consumer.seekToEnd(consumer.partitionsFor(topic).stream()
            .map(partitionInfo -> new TopicPartition(topic, partitionInfo.partition()))
            .collect(Collectors.toList()));
    consumer.partitionsFor(topic).forEach(i -> consumer.position(new TopicPartition(topic, i.partition())));
} 

Java API(指定分区)

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");

final String topic = "mytopic";
try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Arrays.asList(topic));
    consumer.poll(0);
    List<TopicPartition> partitions = new ArrayList<TopicPartition>();
    partitions.add(new TopicPartition(topic, 1));
    partitions.add(new TopicPartition(topic, 2));
    consumer.seekToEnd(partitions); 
    consumer.position(new TopicPartition(topic, 1));
    consumer.position(new TopicPartition(topic, 2));
} 

Current

此方法暂时联想不到相应的应用场景，粗略跳过，待以后了解后再补充。

脚本命令

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic --to-current --execute

脚本命令(指定分区)

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic:1,2 --to-current --execute 

Java API

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");

final String topic = "mytopic";
try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Arrays.asList(topic));
    consumer.poll(0);
    consumer.partitionsFor(topic).stream().map(info -> new TopicPartition(topic, info.partition())).forEach(tp -> {
        long committedOffset = consumer.committed(tp).offset();
        consumer.seek(tp, committedOffset);
    });
} 

Java API(指定分区)

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");

final String topic = "mytopic";
try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Arrays.asList(topic));
    consumer.poll(0);
    TopicPartition tp1 = new TopicPartition(topic, 1);
    TopicPartition tp2 = new TopicPartition(topic, 2);
    consumer.seek(tp1, consumer.committed(tp1).offset());
    consumer.seek(tp2, consumer.committed(tp2).offset());
}

Specified-Offset

重置前
Kafka2.7是如何重设消费者组位移

脚本命令

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic --to-offset 5 --execute
Kafka2.7是如何重设消费者组位移

脚本命令(指定分区)

通常来说，各个分区的提交位移往往是不同的，所以将所有分区的位移设置成同一个值并不显示，需要指定分区。

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic:2 --to-offset 11 --execute 

Java API

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");

final String topic = "mytopic";
try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Arrays.asList(topic));
    consumer.poll(0);
    consumer.partitionsFor(topic).stream().forEach(pi -> {
        TopicPartition tp = new TopicPartition(topic, pi.partition());
        consumer.seek(tp, 5L);
    });
} 

Java API(指定分区)

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");

final String topic = "mytopic";
try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Arrays.asList(topic));
    consumer.poll(0);
    consumer.seek(new TopicPartition(topic, 2), 10L);
} 

Shift-By-N

重置前
Kafka2.7是如何重设消费者组位移

脚本命令

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic --shift-by -1 --execute
Kafka2.7是如何重设消费者组位移

脚本命令(指定分区)

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic:2 --shift-by -2 --execute 

Java API

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");

final String topic = "mytopic";
try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Arrays.asList(topic));
    consumer.poll(0);
    for (PartitionInfo info : consumer.partitionsFor(topic)) {
        TopicPartition tp = new TopicPartition(topic, info.partition());
        consumer.seek(tp, consumer.committed(tp).offset() - 1L);
    }
} 

Java API(指定分区)

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");

final String topic = "mytopic";
try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Arrays.asList(topic));
    consumer.poll(0);
    TopicPartition tp = new TopicPartition(topic, 2);
    consumer.seek(tp, consumer.committed(tp).offset() + 2L);
}
 

DateTime

有时按照时间点来重置位移是个不错的方式，重置前:
Kafka2.7是如何重设消费者组位移

脚本命令

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic --to-datetime 2021-05-09T00:00:00.000 --execute
Kafka2.7是如何重设消费者组位移

脚本命令(指定分区)

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic:2 --to-datetime 2020-05-09T00:00:00.000 --execute

Java API

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");

final String topic = "mytopic";
try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Arrays.asList(topic));
    consumer.poll(0);
    long ts = new Date().getTime() - 24 * 60 * 60 * 1000;
    Map<TopicPartition, Long> timeToSearch = consumer.partitionsFor(topic).stream()
            .map(pi -> new TopicPartition(topic, pi.partition()))
            .collect(Collectors.toMap(Function.identity(), tp -> ts));
    for (Entry<TopicPartition, OffsetAndTimestamp> entry : consumer.offsetsForTimes(timeToSearch).entrySet()) {
        consumer.seek(entry.getKey(), entry.getValue() == null ? consumer.committed(entry.getKey()).offset() : entry.getValue().offset());
    }
} 

Java API(指定分区)

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.108:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "mytopic-consumer-group");

final String topic = "mytopic";
try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Arrays.asList(topic));
    consumer.poll(0);
    long ts = new Date().getTime() -  365 * 24 * 60 * 60 * 1000;
    Map<TopicPartition, Long> timeToSearch = new HashMap<TopicPartition, Long>(){{
        put(new TopicPartition(topic, 2), ts);
    }};
    for (Entry<TopicPartition, OffsetAndTimestamp> entry : consumer.offsetsForTimes(timeToSearch).entrySet()) {
        consumer.seek(entry.getKey(), entry.getValue() == null ? consumer.committed(entry.getKey()).offset() : entry.getValue().offset());
    }
} 

Duration

重置前
Kafka2.7是如何重设消费者组位移

脚本命令

首先需要了解Java Duration的格式PnDTnHnMnS，这里不做详细展开。

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic --by-duration P1DT0H0M0S --execute
Kafka2.7是如何重设消费者组位移

脚本命令(指定分区)

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.108:9092 --group mytopic-consumer-group --reset-offsets --topic mytopic:2 --by-duration P1DT0H0M0S --execute

Java API方式

同DateTime

“Kafka2.7是如何重设消费者组位移”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注亿速云网站，小编将为大家输出更多高质量的实用文章！

向AI问一下细节
推荐阅读：
  1. css怎么实现图片旋转、倾斜、位移以及平滑
  2. mysql怎么重设用户密码

免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。

java kafka

猜你喜欢

最新资讯
相关推荐

相关标签

java工程师 java编程思想 java程序 java并发 java入门 java反射 javascript对象 java后端 java单例模式 java 高并发 java就业 javaweb项目 javajava笔记编程学习 java算法 java.lang.nullpointerexception java锁 weixin-java-miniapp javap java虚拟主机 java编程语言
AI
助
手
产品服务
地区划分
专题活动
帮助支持
关于我们
售后咨询
7*24小时在线电话：400-100-2938
7*24小时在线 QQ：800811969
关注亿速云

亿速云公众号

手机网站二维码

Copyright © Yisu Cloud Ltd. All Rights Reserved. 2018 版权所有

广州亿速云计算有限公司粤ICP备17096448号-1 粤公网安备 44010402001142号增值电信业务经营许可证编号：B1-20181529