Kafka：消费者api：无法手动读取和确认Kafka消费者api的偏移量

Question

我的用例是使用kafka消费者api，这样我们就可以从kafka主题中手动读取最后一次成功处理的数据的偏移量，然后手动确认Kafka的成功处理数据。（这是为了减少数据丢失）。然而，在我当前的实现中，程序向前移动并从下一个偏移读取，即使我注释掉了“ack.acknowledge（）”。我是新来的Kafka和实现我的消费者下面的方式（我们使用Spring引导）

问题是：即使我注释掉ack.acknowledge（），偏移量仍然在更新，消费者正在从下一个偏移量中读取，这是出乎意料的（到目前为止，我的理解是）

消费者配置[请注意，我将ConsumerConfig.ENABLE_AUTO_COMMIT_Config设置为false，并将factory.getContainerProperties（）.setAckMode（ContainerProperties.AckMode.MANUAL_IMMEDIATE）]：

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Autowired
    private AdapterStreamProperties appProperties;

    @Value("${spring.kafka.streams.properties.receive.buffer.bytes}")
    private String receiveBufferBytes;

    @Bean
    public ConsumerFactory<PreferredMediaMsgKey, SendEmailCmd> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, appProperties.getApplicationId());
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, appProperties.getBootstrapServers());
        props.put(
                ConsumerConfig.GROUP_ID_CONFIG,
                "adapter");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "com.ringo.notification.adapter.serializers.PreferredMediaMsgKeyDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "com.ringo.notification.adapter.serializers.SendEmailCmdDeserializer");
        props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, WallclockTimestampExtractor.class.getName());
        props.put(StreamsConfig.RECEIVE_BUFFER_CONFIG, receiveBufferBytes);

        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<PreferredMediaMsgKey, SendEmailCmd>
    kafkaListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<PreferredMediaMsgKey, SendEmailCmd> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }

然后我的消费者以这种方式消费[即使我注释掉ack.acknowledge ()' , 下次它仍然从下一个偏移读取：

  @KafkaListener(topics = Constants.INPUT_TOPIC, groupId = "adapter")
  public void listen(ConsumerRecord<PreferredMediaMsgKey, SendEmailCmd> record,
                     @Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partition,
                     @Header(KafkaHeaders.OFFSET) Long offset, Acknowledgment ack) {

    System.out.println("----------------------------------------------------------------------------");
    System.out.println("Reading from Offset: " +  offset + ", and Partition: " + partition);
    System.out.println("Record for this pertition: Key : "+ record.key() + ", Value : " +   record.value());
    System.out.println("----------------------------------------------------------------------------");
    NotificationProcessedFinal result = processor.processEmail(record.key(),record.value());

    if( StringUtils.isNotEmpty(result.getErrorCmd().getErrorMsg())) {
      kafkaErrorProducerTemplate.send(adapterMsgProperties.getErrorTopic(), record.key(), result.getErrorCmd());
    }
    else {
      kafkaOutputProducerTemplate.send(adapterMsgProperties.getOutputTopic(), record.key(), result.getNotifyEmailCmd());
    }
    ack.acknowledge();
  }

我的gradle的Kafkaapi版本。建造：

//Kafka Dependencie
implementation      'org.apache.kafka:kafka-streams:2.0.1'
implementation      'org.apache.kafka:kafka-clients:2.0.1'

任何洞察力都会有所帮助。

提前谢谢

Answer 1

ack.acknowledge()

仅表示您已提交，您的消费者已成功处理该消息。

这将更新消费者组偏移量，如果应用程序在没有ack的情况下停止，则当前偏移量将不会提交给kafka（您已处理但未确认的最后一条消息）。

具有相同用户组（和相同分配分区）的新用户将再次使用相同的消息，因为应用程序将从用户组偏移量读取起始点。

如果你ack或don't ack没有影响您的当前侦听器，它将简单地继续处理新消息（只要没有错误发生，我想重新平衡也可能是偏移重载的来源（不完全确定））