使用Schema Registry时如何设置Spring Kafka消费者最大重试次数

针对你遇到的这个问题，结合你使用的Spring Kafka 1.3.2.RELEASE、Apache Avro 1.8.2和Confluent Schema Registry 3.1.2版本，可以通过以下几个方案来控制请求重试次数、减少错误日志输出：

1. 配置Schema Registry客户端的重试参数 #

Confluent的Schema Registry客户端本身支持重试配置，你可以通过设置最大重试次数和重试间隔，避免Registry宕机时无限制地发起请求。这些参数可以直接通过Spring Kafka的消费者属性传递：

在application.yml中添加配置：

spring:
  kafka:
    consumer:
      properties:
        schema.registry.url: http://你的-registry地址:8081
        # 最大重试次数，建议设为3-5次
        schema.registry.retry.max.retries: 3
        # 每次重试的间隔时间（毫秒），避免短时间内大量请求
        schema.registry.retry.backoff.ms: 1000

或者用application.properties：

spring.kafka.consumer.properties.schema.registry.url=http://你的-registry地址:8081
spring.kafka.consumer.properties.schema.registry.retry.max.retries=3
spring.kafka.consumer.properties.schema.registry.retry.backoff.ms=1000

这些参数会被底层的CachedSchemaRegistryClient读取，当请求Registry失败时，会按照配置的次数和间隔重试，超过次数后就会抛出异常，不会无限重试。

2. 启用并优化Schema本地缓存 #

默认的CachedSchemaRegistryClient已经带有本地缓存，但你可以调整缓存大小，让已获取过的Schema ID直接从本地读取，无需每次请求Registry：

添加缓存大小配置：

spring:
  kafka:
    consumer:
      properties:
        schema.registry.cache.size: 1000

这样，只要是之前处理过的Schema ID，即使Registry宕机，消费者也会直接使用本地缓存的Schema，不会发起新的请求，大大减少无效请求的数量。

3. 配置消费者错误处理器，处理重试失败的消息 #

当Schema请求重试耗尽后，需要避免消费者一直卡在这条失败的消息上。你可以自定义一个错误处理器，接管失败消息的处理逻辑（比如记录日志、发送到死信队列等）：

首先创建错误处理器类：

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.listener.ConsumerAwareErrorHandler;
import org.springframework.stereotype.Component;

import java.util.Collections;

@Component("kafkaConsumerErrorHandler")
public class KafkaConsumerErrorHandler implements ConsumerAwareErrorHandler {

    private static final Logger log = LoggerFactory.getLogger(KafkaConsumerErrorHandler.class);

    @Override
    public void handle(Exception thrownException, ConsumerRecord<?, ?> data, Consumer<?, ?> consumer) {
        // 记录详细错误日志
        log.error("处理消息失败，已达到Schema Registry请求重试上限，消息主题：{}，偏移量：{}", 
                  data.topic(), data.offset(), thrownException);
        
        // 可选：手动提交偏移量，让消费者跳过这条消息（根据业务需求调整）
        consumer.commitSync(Collections.singletonMap(
                new org.apache.kafka.common.TopicPartition(data.topic(), data.partition()),
                new org.apache.kafka.clients.consumer.OffsetAndMetadata(data.offset() + 1)
        ));
    }
}

然后在@KafkaListener注解中指定这个错误处理器：

import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaMessageListener {

    @KafkaListener(topics = "你的主题名称", errorHandler = "kafkaConsumerErrorHandler")
    public void listen(ConsumerRecord<String, GenericRecord> record) {
        // 你的消息处理逻辑
    }
}

这样，当Schema请求重试失败后，错误处理器会触发，避免消费者反复处理同一条消息并发起无效请求。

4. 调整日志级别，减少冗余错误日志 #

如果不想看到太多重复的Schema Registry请求失败日志，可以调整Confluent客户端的日志级别，比如把io.confluent.kafka.schemaregistry.client的日志级别设为WARN或ERROR：

以Logback为例，在logback.xml中添加：

<logger name="io.confluent.kafka.schemaregistry.client" level="WARN"/>

这样只有严重的错误才会被记录，减少日志量的同时也不会错过关键错误信息。

内容的提问来源于stack exchange，提问作者Azarea