Spring Kafka-显示带有@重试的自定义错误消息

Question

我目前正在使用Spring Kafka来使用topic中的消息以及Spring的@Retry。因此，基本上，我正在尝试处理消费者消息以防出现错误。但在这样做时，我希望避免KafkaMessageListenerContainer抛出的异常消息。相反，我想显示一条自定义消息。我尝试在ConcurrentKafkAlisterContainerFactory中添加错误处理程序，但这样做时，我的重试没有被调用。是否有人可以指导我如何显示自定义异常消息以及重试方案？下面是我的代码片段：

ConcurrentKafkalistenerContainerFactorybean配置

@Bean
ConcurrentKafkaListenerContainerFactory << ? , ? > concurrentKafkaListenerContainerFactory(ConcurrentKafkaListenerContainerFactoryConfigurer configurer, ConsumerFactory < Object, Object > kafkaConsumerFactory) {
    ConcurrentKafkaListenerContainerFactory < Object, Object > kafkaListenerContainerFactory =
        new ConcurrentKafkaListenerContainerFactory < > ();
    configurer.configure(kafkaListenerContainerFactory, kafkaConsumerFactory);
    kafkaListenerContainerFactory.setConcurrency(1);

    // Set Error Handler
    /*kafkaListenerContainerFactory.setErrorHandler(((thrownException, data) -> {
        log.info("Retries exhausted);
    }));*/
    return kafkaListenerContainerFactory;
}

Kafka消费者

@KafkaListener(
    topics = "${spring.kafka.reprocess-topic}",
    groupId = "${spring.kafka.consumer.group-id}",
    containerFactory = "concurrentKafkaListenerContainerFactory"
)
@Retryable(
    include = RestClientException.class,
    maxAttemptsExpression = "${spring.kafka.consumer.max-attempts}",
    backoff = @Backoff(delayExpression = "${spring.kafka.consumer.backoff-delay}")
)
public void onMessage(ConsumerRecord < String, String > consumerRecord) throws Exception {
    // Consume the record
    log.info("Consumed Record from topic : {} ", consumerRecord.topic());


    // process the record
    messageHandler.handleMessage(consumerRecord.value());
}

Answer 1

我的事件重试模板，

@Bean(name = "eventRetryTemplate")
public RetryTemplate eventRetryTemplate() {
RetryTemplate template = new RetryTemplate();

ExceptionClassifierRetryPolicy retryPolicy = new ExceptionClassifierRetryPolicy();
Map<Class<? extends Throwable>, RetryPolicy> policyMap = new HashMap<>();
policyMap.put(NonRecoverableException.class, new NeverRetryPolicy());
policyMap.put(RecoverableException.class, new AlwaysRetryPolicy());
retryPolicy.setPolicyMap(policyMap);

ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
backOffPolicy.setInitialInterval(backoffInitialInterval);
backOffPolicy.setMaxInterval(backoffMaxInterval);
backOffPolicy.setMultiplier(backoffMultiplier);

template.setRetryPolicy(retryPolicy);
template.setBackOffPolicy(backOffPolicy);
return template;
}

我的kafka侦听器使用重试模板，

@KafkaListener(
  groupId = "${kafka.consumer.group.id}",
  topics = "${kafka.consumer.topic}",
  containerFactory = "eventContainerFactory")
  public void eventListener(ConsumerRecord<String, String> 
  events,
  Acknowledgment acknowledgment) {
  eventRetryTemplate.execute(retryContext -> {
  retryContext.setAttribute(EVENT, "my-event");
  eventConsumer.consume(events, acknowledgment);
  return null;
  });
}

我的Kafka消费者财产，

private ConcurrentKafkaListenerContainerFactory<String, String> 
getConcurrentKafkaListenerContainerFactory(
  KafkaProperties kafkaProperties) {
ConcurrentKafkaListenerContainerFactory<String, String> factory =
    new ConcurrentKafkaListenerContainerFactory<>();
factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
factory.getContainerProperties().setAckOnError(Boolean.TRUE);
kafkaErrorEventHandler.setCommitRecovered(Boolean.TRUE);
factory.setErrorHandler(kafkaErrorEventHandler);
factory.setConcurrency(1); 
factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
factory.setConsumerFactory(getEventConsumerFactory(kafkaProperties));
return factory;
}

kafka错误事件处理程序是我的自定义错误处理程序，它扩展了SeektocurInterrorHandler并实现了handle error方法，如下图所示。。。。。

@Override
public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> 
records,
  Consumer<?, ?> consumer, MessageListenerContainer container) {

log.info("Non recoverable exception. Publishing event to Database");
super.handle(thrownException, records, consumer, container);

ConsumerRecord<String, String> consumerRecord = (ConsumerRecord<String, 
String>) records.get(0);

FailedEvent event = createFailedEvent(thrownException, consumerRecord);

failedEventService.insertFailedEvent(event);

log.info("Successfully Published eventId {} to Database...", 
event.getEventId());
}

这里，failed event service是我的自定义类，它将这些失败事件放入一个可查询的关系数据库中（我选择它作为我的DLQ）。

Answer 2

您不应该使用Retryable（可重试）以及SeektocurInterrorHandler（从2.5开始，现在是默认版本；因此我假设您使用的是该版本）。

相反，请使用最大尝试次数、后退和可重试的异常配置自定义的SeekToCurrentErrorHandler。

该错误消息是正常的；由容器记录；通过在seektocurinterrorhandler上设置logLevel属性，可以将日志级别从ERROR降低到INFO或DEBUG。您还可以向其中添加自定义恢复程序，以便在重试结束后记录自定义消息。