Spring Kafka 重试日志记录

Question

我需要从一个 kafka 主题消费，对记录做一些工作并使用 spring-kafka 2.1.7 生成另一个主题。其他要求是事务性的，仅用于一次语义、重试和错误处理。在提交失败时一条记录我应该重试 3 次，记录每条重试消息以重试主题，然后在所有重试失败时将记录发送到死信主题。我查看了https://github.com/spring-projects/spring-kafka/issues/575，它在解决问题方面有很好的细节。我正在努力解决的问题是如何记录每条重试消息以及消费者偏移量、它试图提交的主题等详细信息。有没有办法从重试回调中获取这些信息？下面的 retrylistener sn-p 注册了一个 org.springframework.kafka.listener.LoggingErrorHandler ，它被设置为 ConcurrentKafkaListenerContainerFactory 的容器属性？

         @Bean
         public RetryListener retryListener(KafkaTemplate<String,SpecificRecord> kafkaTemplate) {
             return new RetryListenerSupport() {

                public void onError(RetryContext context, RetryCallback callback, Throwable throwable) {
                    int retryCount =context.getRetryCount();
                    kafkaTemplate) .send(new ProducerRecord<String,SpecificRecord>("topic_name",record));
                }
             };
         }

Answer 1

RetryContext 在RetryingMessageListenerAdapter 中填充了一些有用的信息：

context.setAttribute(CONTEXT_RECORD, record);
switch (RetryingMessageListenerAdapter.this.delegateType) {
    case ACKNOWLEDGING_CONSUMER_AWARE:
        context.setAttribute(CONTEXT_ACKNOWLEDGMENT, acknowledgment);
        context.setAttribute(CONTEXT_CONSUMER, consumer);
        RetryingMessageListenerAdapter.this.delegate.onMessage(record, acknowledgment, consumer);
        break;
    case ACKNOWLEDGING:
        context.setAttribute(CONTEXT_ACKNOWLEDGMENT, acknowledgment);
        RetryingMessageListenerAdapter.this.delegate.onMessage(record, acknowledgment);
        break;
    case CONSUMER_AWARE:
        context.setAttribute(CONTEXT_CONSUMER, consumer);
        RetryingMessageListenerAdapter.this.delegate.onMessage(record, consumer);
        break;
    case SIMPLE:
        RetryingMessageListenerAdapter.this.delegate.onMessage(record);
}