Spring KafkaTemplate无法处理分区编号错误问题排查

1. 提前校验PartitionId合法性 #

在发送消息前主动获取目标Topic的分区数，直接判断传入的partitionId是否在[0, 分区数-1]的合法范围内，从根源避免无效请求：

// 注入AdminClient
@Autowired
private AdminClient adminClient;

public void sendMessage(String topic, int partitionId, String key, String payload) throws ExecutionException, InterruptedException {
    // 获取Topic元数据
    DescribeTopicsResult describeResult = adminClient.describeTopics(Collections.singleton(topic));
    Map<String, TopicDescription> topicDescMap = describeResult.all().get();
    TopicDescription topicDesc = topicDescMap.get(topic);
    
    int partitionCount = topicDesc.partitions().size();
    if (partitionId < 0 || partitionId >= partitionCount) {
        throw new IllegalArgumentException("Invalid partitionId: " + partitionId + ", valid range is 0~" + (partitionCount-1));
    }
    
    // 合法则发送消息
    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, partitionId, key, payload);
    future.addCallback(new ListenableFutureCallback<>() {
        @Override
        public void onSuccess(SendResult<String, String> result) {
            // 成功处理逻辑
        }

        @Override
        public void onFailure(Throwable ex) {
            // 失败处理逻辑
        }
    });
}

2. 配置Kafka客户端超时参数 #

无效PartitionId导致阻塞的核心原因是Kafka客户端会不断重试获取元数据或尝试发送，没有超时限制。通过配置以下参数强制触发失败回调：

• request.timeout.ms：单个请求的超时时间，可缩短至5-10秒
• delivery.timeout.ms：整个投递流程（包括重试）的最大超时时间，需大于request.timeout.ms + retries * retry.backoff.ms
• metadata.max.age.ms：元数据缓存的过期时间，设置为30秒左右，确保客户端及时更新Topic分区信息

在Spring Boot的application.yml中配置示例：

spring:
  kafka:
    producer:
      properties:
        request.timeout.ms: 5000
        delivery.timeout.ms: 10000
        metadata.max.age.ms: 30000
        retries: 0  # 无效分区无需重试，直接关闭重试更高效

3. 主动监听Future超时 #

仅依赖ListenableFutureCallback可能因客户端无限等待而无法触发回调，可主动给Future设置超时时间，强制捕获异常：

ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, partitionId, key, payload);
future.addCallback(new ListenableFutureCallback<>() {
    @Override
    public void onSuccess(SendResult<String, String> result) {
        // 成功处理逻辑
    }

    @Override
    public void onFailure(Throwable ex) {
        // 失败处理逻辑
    }
});

// 单独线程监听超时，不阻塞主线程
CompletableFuture.runAsync(() -> {
    try {
        future.get(5, TimeUnit.SECONDS);
    } catch (TimeoutException e) {
        // 处理超时逻辑，比如记录日志、触发告警
        future.cancel(true);
    } catch (Exception e) {
        // 处理其他异常
    }
});

4. 调整重试策略 #

如果保留重试配置，需确保重试仅针对可恢复的异常（如网络波动），无效PartitionId属于不可恢复异常，可通过自定义重试拦截器过滤此类异常，避免无意义的重试阻塞。

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