Kafka Consumer语言 - 在恢复方法中获取侦听器收到的参数

Question

我正在构建一个弹簧卡夫卡消费者。我已经设置了重试机制。重试次数用尽后，我想将失败的消息推送到死信主题。

侦听方法具有以下参数

public void listen(@Payload Map<String, Object> conciseMap,
        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
        Acknowledgment ack) throws JsonProcessingException {

作为恢复方法的一部分，我想获取作为输入传递给侦听器映射的简洁地图或我的主题收到的原始消息。有没有办法做到这一点？

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Map<String, Object>> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Map<String, Object>> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConcurrency(conncurrency);
    factory.setConsumerFactory(consumerFactory());
    factory.setRetryTemplate(retryTemplate());
    factory.setRecoveryCallback(new RecoveryCallback<Object>() {
        @Override
        public Object recover(RetryContext context) throws Exception {
            // TODO Auto-generated method stub
            logger.debug(" In recovery callback method !!");
            ((Acknowledgment)context.getAttribute(RetryingMessageListenerAdapter.CONTEXT_ACKNOWLEDGMENT)).acknowledge();
            return null;
        }
    });
    factory.getContainerProperties().setAckMode(AckMode.MANUAL);
    return factory;
}
        factory.getContainerProperties().setAckMode(AckMode.MANUAL);
        return factory;
    }
    /*
     * Retry template.
     */
    protected RetryPolicy retryPolicy() {
        SimpleRetryPolicy policy = new SimpleRetryPolicy(maxRetryAttempts, retryableExceptions);
        return policy;
    }
    protected BackOffPolicy backOffPolicy() {
        ExponentialBackOffPolicy policy = new ExponentialBackOffPolicy();
        policy.setInitialInterval(initialRetryInterval);
        policy.setMultiplier(retryMultiplier);
        return policy;
    }
    protected RetryTemplate retryTemplate() {
       RetryTemplate template = new RetryTemplate();
       template.setRetryPolicy(retryPolicy());
       template.setBackOffPolicy(backOffPolicy());
       return template;
    }
}

Answer 1

您无法在

RecoveryCallback的RetryContext中获取转换后的conciseMap，但您可以在转换前从主题中检索原始ConsumerRecord：

(ConsumerRecord) context.getAttribute(RetryingMessageListenerAdapter.CONTEXT_RECORD)