云服务器特惠 99/年起
  1. 腾讯云优惠网首页
  2. 腾讯云产品

消息队列 CKafka 版 自动重置消费进度

• 腾讯云产品

这篇说说Kafka参数auto.offset.reset，定义当无法获取消费分区位移时从何处开始消费，可选earliest、latest（默认）和none，并说明OFFSET_OUT_OF_RANGE错误原因及none模式的使用方式。 参数选项：earliest重置到最小offset，latest到最大offset，none不自动重置。 OFFSET_OUT_OF_RANGE：因日志过期、SDK异常或leader切换。

相关产品
消息队列 CKafka 版 产品页，含版本规格、价格与购买入口。

这里说明 auto.offset.reset 参数的相关概念及使用方式。

一、什么是 auto.offset.reset？

auto.offset.reset 参数定义了当无法获取消费分区的位移时从何处开始消费。例如：当 Broker 端没有 offset（如第一次消费或 offset 超过7天过期）时如何初始化 offset，当收到 OFFSET_OUT_OF_RANGE 错误时如何重置 Offset。

auto.offset.reset 参数设置有如下选项：

·earliest：表示自动重置到 partition 的最小 offset。

·latest：默认为 latest，表示自动重置到 partition 的最大 offset。

·none：不自动进行 offset 重置，抛出 OffsetOutOfRangeException 异常。

二、什么时候会出现 OFFSET_OUT_OF_RANGE？

该错误表示客户端提交的 offset 不在服务端允许的 offset 范围之内。例如：topicA 的分区1的 LogStartOffset 为100，LogEndOffset 为300，此时如果客户端提交的 offset 小于100或者大于300，服务端就会返回该错误，此时就会进行 offset 重置。

以下情况可能会导致客户端触发该错误：

·客户端设置了 offset，然后一段时间内没有消费，但 Topic 设置了消息保留时间，当过了保留时间后， offset 在服务端已经被删除了，即发生了日志滚动，此时客户端再提交删除了的 offset，则会发生该错误。

·因为 SDK Bug、网络丢包等问题，导致客户端提交了异常的 offset，则会触发该错误。

·服务端有未同步副本，此时发生了 leader 切换，触发了 follower 副本的截断，此时如果客户端提交的 offset 在截断的范围之内，则会触发该错误。

三、使用背景

不希望发生 offset 自动重置的情况，因为业务不允许发生大规模的重复消费。

注意：此时消费组在第一次消费的时候就会找不到 offset 而报错，这时就需要在 catch 里手动设置 offset。

四、使用说明

auto.offset.reset 设置为 None 以后，可以避免 offset 自动重置的问题，但是当增加分区的时候，因为关闭了自动重置机制，客户端不知道新的分区要从哪里开始消费，则会产生异常，此时需要人工去设置消费分组 offset 并消费。

五、使用方式

消费者在消费时，当 consumer 设置 auto.offset.reset=none， 捕获到 NoOffsetForPartitionException 异常，在 catch 里自己设置 offset。你可以根据自身业务情况选择以下方式中的其中一种。

·指定 offset，这里需要自己维护 offset，方便重试。

·指定从头开始消费。

·指定 offset 为最近可用的 offset。

·根据时间戳获取 offset，设置 offset。

示例代码如下：

package com.tencent.tcb.operation.ckafka.plain;import com.google.common.collect.Lists;import com.tencent.tcb.operation.ckafka.JavaKafkaConfigurer;import java.time.Instant;import java.time.temporal.ChronoUnit;import java.util.ArrayList;import java.util.Collection;import java.util.HashMap;import java.util.List;import java.util.Map;import java.util.Map.Entry;import java.util.Properties;import org.apache.kafka.clients.CommonClientConfigs;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.apache.kafka.clients.consumer.NoOffsetForPartitionException;import org.apache.kafka.clients.consumer.OffsetAndTimestamp;import org.apache.kafka.clients.producer.ProducerConfig;import org.apache.kafka.common.PartitionInfo;import org.apache.kafka.common.TopicPartition;import org.apache.kafka.common.config.SaslConfigs;public class KafkaPlainConsumerDemo { public static void main(String args[]) { //设置JAAS配置文件的路径。 JavaKafkaConfigurer.configureSaslPlain(); //加载kafka.properties。 Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties(); Properties props = new Properties(); //设置接入点，请通过控制台获取对应Topic的接入点。 props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers")); //接入协议。 props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT"); //Plain方式。 props.put(SaslConfigs.SASL_MECHANISM, "PLAIN"); //两次Poll之间的最大允许间隔。 //消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Consumer Group移除并触发Rebalance，默认30s。 props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000); //每次Poll的最大数量。 //注意该值不要改得太大，如果Poll太多数据，而不能在下次Poll之前消费完，则会触发一次负载均衡，产生卡顿。 props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30); //消息的反序列化方式。 props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); //当前消费实例所属的消费组，请在控制台申请之后填写。 //属于同一个组的消费实例，会负载消费消息。 props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id")); //消费offset的位置。注意！如果auto.offset.reset=none这样设置，消费组在第一次消费的时候 就会报错找不到offset，第一次这时候就需要在catch里手动设置offset。 props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none"); //构造消费对象，也即生成一个消费实例。 KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props); //设置消费组订阅的Topic，可以订阅多个。 //如果GROUP_ID_CONFIG是一样，则订阅的Topic也建议设置成一样。 List<String> subscribedTopics = new ArrayList<String>(); //如果需要订阅多个Topic，则在这里添加进去即可。 //每个Topic需要先在控制台进行创建。 String topicStr = kafkaProperties.getProperty("topic"); String[] topics = topicStr.split(","); for (String topic : topics) { subscribedTopics.add(topic.trim()); } consumer.subscribe(subscribedTopics); //循环消费消息。 while (true) { try { ConsumerRecords<String, String> records = consumer.poll(1000); //必须在下次Poll之前消费完这些数据, 且总耗时不得超过SESSION_TIMEOUT_MS_CONFIG。 建议开一个单独的线程池来消费消息，然后异步返回结果。 for (ConsumerRecord<String, String> record : records) { System.out.println( String.format("Consume partition:%d offset:%d", record.partition(), record.offset())); } } catch (NoOffsetForPartitionException e) { System.out.println(e.getMessage()); //当auto.offset.reset设置为 none时，需要捕获异常 自己设置offset。你可以根据自身业务情况选择以下方式中的其中一种。 //e.g 1 :指定offset, 这里需要自己维护offset，方便重试。 Map<Integer, Long> partitionBeginOffsetMap = getPartitionOffset(consumer, topicStr, true); Map<Integer, Long> partitionEndOffsetMap = getPartitionOffset(consumer, topicStr, false); consumer.seek(new TopicPartition(topicStr, 0), 0); //e.g 2:从头开始消费 consumer.seekToBeginning(Lists.newArrayList(new TopicPartition(topicStr, 0))); //e.g 3:指定offset为最近可用的offset。 consumer.seekToEnd(Lists.newArrayList(new TopicPartition(topicStr, 0))); //e.g 4: 根据时间戳获取offset，就是根据时间戳去设置offset。例如重置到10分钟前的offset Map<TopicPartition, Long> timestampsToSearch = new HashMap<>(); Long value = Instant.now().minus(300, ChronoUnit.SECONDS).toEpochMilli(); timestampsToSearch.put(new TopicPartition(topicStr, 0), value); Map<TopicPartition, OffsetAndTimestamp> topicPartitionOffsetAndTimestampMap = consumer .offsetsForTimes(timestampsToSearch); for (Entry<TopicPartition, OffsetAndTimestamp> entry : topicPartitionOffsetAndTimestampMap .entrySet()) { TopicPartition topicPartition = entry.getKey(); OffsetAndTimestamp entryValue = entry.getValue(); consumer.seek(topicPartition, entryValue.offset()); // 指定offset, 这里需要自己维护offset，方便重试。 } } } } /** * 获取topic的最早、最近的offset * @param consumer * @param topicStr * @param beginOrEnd true begin; false end * @return */ private static Map<Integer, Long> getPartitionOffset(KafkaConsumer<String, String> consumer, String topicStr, boolean beginOrEnd) { Collection<PartitionInfo> partitionInfos = consumer.partitionsFor(topicStr); List<TopicPartition> tp = new ArrayList<>(); Map<Integer, Long> map = new HashMap<>(); partitionInfos.forEach(str -> tp.add(new TopicPartition(topicStr, str.partition()))); Map<TopicPartition, Long> topicPartitionLongMap; if (beginOrEnd) { topicPartitionLongMap = consumer.beginningOffsets(tp); } else { topicPartitionLongMap = consumer.endOffsets(tp); } topicPartitionLongMap.forEach((key, beginOffset) -> { int partition = key.partition(); map.put(partition, beginOffset); }); return map; }}

相关内容：

  1. 消息队列 CKafka 版 Schema Registry 接入 CKafka 无论是使用传统的 Avro API 自定义序列化类与反序列化类，还是使用 Twitter 的 Bijection 类库实现 Avro 的序列化与反序列化，两种方法有相同的缺点：在每条 Kafka 记录里都嵌入了 Schema，从而导致记录的大小成倍增加。但是不管怎样，在读取记录时仍然需要用到整个 Schema，所以要先找到 Schema。 CKafka 提供了数据共用一个 Schema 的方法：将 Schema 中的。...
  2. 消息队列 CKafka 版 配置事务消息 Kafka 的事务功能是为了支持在分布式环境中实现原子性操作而设计的。它允许生产者在发送消息时确保消息的完整性和一致性，特别是在需要多条消息作为一个整体进行处理的场景中。以下是 Kafka 事务的主要概念和功能介绍。...
  3. 消息队列 CKafka 版 VPC 网络接入 这篇以在 VPC 网络环境下调用 Java SDK 为例介绍接入消息队列 CKafka 版并收发消息的操作过程，帮助你更好地理解消息收发的完整过程。...
  4. 消息队列 CKafka 版 使用 SDK 收发消息（推荐） 在控制台上完成实例、Topic 等资源配置后，可以以使用我们提供的 SDK Demo 连接集群，进行消息收发测试。这篇以调用 Java SDK 为例介绍通过开源 SDK 实现消息收发的操作过程，帮助你更好地理解消息收发的完整过程。...
  5. 消息队列 CKafka 版 Java SDK TDMQ CKafka 是一个分布式流处理平台，用于构建实时数据管道和流式应用程序。它提供了高吞吐量、低延迟、可伸缩性和容错性等特性。Kafka Clients：Kafka 自带的客户端，通过 Java 实现，是 Kafka 生产和消费标准协议的客户端。...
  6. 消息队列 CKafka 版 接入低版本自建 Kafka CKafka 兼容0.9及以上的生产/消费接口（目前可以直接购买的版本包括 2.4.1、2.8.1、3.2.3 版本），如果接入低版本（例如0.8版本）的自建 Kafka，需要对接口进行相应改造。这篇将从生产端和消费端对比0.8版本 Kafka 和高版本 Kafka，并提供改造方式。...
  7. 消息队列 CKafka 版 公网域名接入 这篇以 Java 客户端为例介绍在公网环境下，使用 SASL_PLAINTEXT 方式接入消息队列 CKafka 版并收发消息的过程。...
  8. 消息队列 CKafka 版 Spark Streaming 接入 CKafka Spark Streaming 是 Spark Core 的一个扩展，用于高吞吐且容错地处理持续性的数据，目前支持的外部输入有 Kafka、Flume、HDFS/S3、Kinesis、Twitter 和 TCP socket。...
赞 (0)
上一篇 4天前
下一篇 4天前