如何触发并测试Kafka生产者回调的onFailure方法

以下是4种可快速复现失败场景的方法，可根据你的测试场景选择：

1. 模拟Kafka集群不可用（最便捷的功能测试方案） #

• 直接停止当前服务连接的Kafka broker节点，或者修改项目配置文件中spring.kafka.bootstrap-servers的地址为不存在的服务地址（例如127.0.0.1:9999，确保该端口没有运行任何服务）
• 为了避免等待重试超时，可临时修改生产者配置：spring.kafka.producer.retries=0，发送消息后会立即触发失败回调

2. 构造非法Topic/非法序列化消息 #

• 先关闭Kafka broker端的自动创建Topic配置：auto.create.topics.enable=false，然后将代码中的kurlyTopicNamePgResponse替换为一个集群中不存在的Topic名称，发送消息时会因找不到Topic返回失败
• 也可以修改KafkaPgResponseDto结构，添加一个无法被当前配置的序列化器（默认是Json序列化器）处理的字段，例如加入一个没有无参构造方法的自定义类字段，触发序列化异常进入失败回调

3. 单元测试Mock方案（无需修改环境，适合自动化验证） #

直接通过Mock工具伪造失败的Future对象，不需要调整服务和Kafka配置，示例代码如下：

// 导入相关类：org.springframework.util.concurrent.SettableListenableFuture
SettableListenableFuture<SendResult<String, KafkaPgResponseDto>> failedFuture = new SettableListenableFuture<>();
// 自定义模拟的异常
failedFuture.setException(new RuntimeException("模拟Kafka发送失败异常"));
// Mock pgResponseKafkaTemplate的send方法返回构造的失败Future
Mockito.when(pgResponseKafkaTemplate.send(Mockito.anyString(), Mockito.any(KafkaPgResponseDto.class)))
        .thenReturn(failedFuture);

// 调用你的业务发送逻辑，此时会直接触发onFailure回调

4. 基于acks配置构造失败场景 #

如果你的生产者配置了spring.kafka.producer.acks=all，可以手动停掉对应Topic分区的所有副本，仅保留leader节点，同时调整min.insync.replicas配置为大于当前ISR列表的数量，此时发送消息会因ISR数量不足返回失败。

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