报错信息

Undefined offset with no reset policy for partitions

报错原因

这是因为设置的auto.offset.reset为none，表示如果在kafka broker中找不到当前消费者组的offset时，则抛出异常。

源码

/**
 * auto.offset.reset
 */
public static final String AUTO_OFFSET_RESET_CONFIG = "auto.offset.reset";
public static final String AUTO_OFFSET_RESET_DOC = "What to do when there is no initial offset in Kafka or if the current offset doesnot exist any more on the server (e.g. because that data has been deleted): earliest: automatically reset the offset to theearliest offsetlatest: automatically reset the offset to the latest offsetnone: throw exception to the consumer if noprevious offset is found for the consumer's groupanything else: throw exception to the consumer.";

代码配置

@Bean("kafkaConsumerProperties")
public Map kafkaConsumerProperties() {
    Map props = new HashMap<>(4);
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, consumerAutoOffsetReset);
    return props;
}

含义说明

• earliest

当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费

• latest

当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据

• none

topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常

参考资料

《kafka 权威指南》

Kafka auto.offset.reset值详解

kafka的auto.offset.reset详解

• amy list
  {:toc}