Spring KafkaListener在暂停恢复后未拾取最早未提交的偏移量

Question

UseCase—希望在外部服务出现可重试异常时暂停Spring KafkaListener几秒钟，并希望从最早的未提交偏移量恢复。

问题我有-下面是实现。

1） 没有外寻用法-在恢复Spring之后，kafkalistener正在挑选进入主题分区的最新消息。这不符合目的（从上次提交的偏移量到最近的偏移量之间的消息丢失）

2） 使用seek-我不知道如何处理kafkaconsumer

源代码

消费者中的Lisener方法

 @KafkaListener(topics = "${kafka.consumer.topic}", containerFactory = "kafkaListenerContainerFactory")
    public void onReceiving(@Payload ConsumerRecord<String, String> consumerRecord, Acknowledgment acknowledgment) {

            try {
                Event event = translate(consumerRecord);
                someService.processEvent(event, consumerRecord);
                commitOffset(acknowledgment)
            } catch(ConsumerException e) {
                //DO NOT commit offset
            }
        }

    private void commitOffset(Acknowledgment acknowledgment) {
        acknowledgment.acknowledge();
    }

Service

public void processEvent(Event event, ConsumerRecord<String, String> consumerRecord) {

    try {
        //call an external API to get realTime event details
        //Have a retry on this client
       BusinessEntity businessEntity = externalServiceClient.get(event);
       //process the Entity 
       anotherService.process(businessEntity);
    } catch(RetryableException re) {
        //feign.RetryableException
        //we are using feign declarative clients 
        consumerErrorHandler.handle(re, consumerRecord);
    }
}

错误处理程序--

public class ConsumerErrorHandler implements ErrorHandler {

    @Autowired
    private final KafkaListenerEndpointRegistry registry;

    //org.springframework.core.task.SimpleAsyncTaskExecutor
    @Autowrired 
    private final Executor executor;

    @Autowired
    private Consumer<String, String> kafkaConsumer;

    @Override
    public void handle(Exception thrownException, ConsumerRecord<?, ?> data) {

        //Trying to delegate this to a new Async thread.

        executor.execute(() -> {
            registry.getListenerContainers().forEach(container -> {

                if ((!container.isContainerPaused() || !container.isPauseRequested())) {
                    log.info("STOPPING_CONSUMER on error");

                    Optional<TopicPartition> topicPartition = container.getAssignedPartitions().stream().filter(a -> a.partition() == data.partition()).findFirst();

                    container.pause();
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }

                    log.info("BEFORE_RESUME");
                    log.info("SEEK CONSUMER before RESUME to this offset: "+data.offset());

                    topicPartition.ifPresent(a ->
                    {
                        log.info("Seek from the current position: " + data.offset());
                        kafkaConsumer.seek(a, data.offset());
                    });

                    container.resume();

                    log.info("RESUMING_CONSUMER  after seek");

                    topicPartition.ifPresent(a -> {
                        log.info("CONSUMER is up NOW ??");
                    });
                }
            });
        });

    }
}

消费者配置

    private Map<String, Object> consumerConfigs() {
        Map<String, Object> confMap = new HashMap<>();
        confMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, pubSubServers);
        confMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        confMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        confMap.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupIdConfig);
        confMap.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "50000");
        confMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "50000");
        confMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        confMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, OffsetResetStrategy.EARLIEST.name().toLowerCase());
        if (this.securityProtocol.equalsIgnoreCase(SSL)) {
            confMap.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, this.securityProtocol);
            confMap.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,
                    this.getClass().getResource(clientTrustStoreLocation).getPath());
            confMap.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, this.sslTrustStorePassword);
            confMap.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG,
                    this.getClass().getResource(this.clientKeyStoreLocation).getPath());
            confMap.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, sslKeyStorePassword);
            confMap.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, sslKeyPassword);
            confMap.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG,null);
        }
        return confMap;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
      return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
      ConcurrentKafkaListenerContainerFactory<String, String> factory =
          new ConcurrentKafkaListenerContainerFactory<>();
      factory.setConcurrency("1");
      factory.getContainerProperties().setAckOnError(false);
      factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
      factory.getContainerProperties().setConsumerTaskExecutor(taskExecutor());
      factory.setConsumerFactory(consumerFactory());
      factory.setErrorHandler(consumerErrorHandler);
      factory.setRetryTemplate(retryTemplate());
      return factory;
    }

    @Bean
    public AsyncListenableTaskExecutor taskExecutor() {
      return createTaskExecutor("1");
    }

     private RetryTemplate retryTemplate() {
         RetryTemplate template = new RetryTemplate();
         template.setRetryPolicy(retryPolicy());
         template.setBackOffPolicy(backOffPolicy());
         return template;
    }

    private BackOffPolicy backOffPolicy() {
        ExponentialBackOffPolicy policy = new ExponentialBackOffPolicy();
        policy.setInitialInterval(1000);
        return policy;
    }

    private RetryPolicy retryPolicy() {
         SimpleRetryPolicy policy = new SimpleRetryPolicy();
         policy.setMaxAttempts("1");
         return policy;
    }

Answer 1

使用一个消费者数据库。

不能在另一个线程上执行搜索。请参阅KafkaConsumerjavadocs-它不是线程安全的。

您还必须为其他主题/分区查找任何剩余记录（除非您只有一个主题/分区）。

最后，在容器暂停之前，您不得退出错误处理程序-否则将发生竞争，消费者可能会在暂停之前执行另一次poll（）。

有关如何执行此类操作的示例，请参阅SeekToMONtErrorHandler和ContainerStoppingErrorHandler。必须在另一个线程上调用Stop（）以避免死锁，但您可以prence（）容器在消费者线程上（它只是设置一个标志，这样消费者将在下一个轮询（）之前暂停（）。

要resume（）容器，请使用ApplicationListener或@EventListener侦听暂停容器的容器空闲事件（设置idleEventVal以获取这些事件）。