Spring Boot单元测试使用Embedded Kafka调用getBroker()出现空指针异常

问题原因 #

你遇到的空指针是以下几个错误共同导致的：

• 调用EmbeddedKafkaCondition.getBroker()时Spring上下文还未完成初始化，Embedded Kafka实例尚未创建，因此返回null
• 你给ReactiveKafkaConsumerTemplate加了static修饰符，Spring默认不支持静态变量的依赖注入，即使加了@Autowired也不会生效
• @Before/@BeforeClass执行时机早于Embedded Kafka的初始化流程，此时无法获取到可用的Broker实例

修复方案 #

1. 测试类补充@SpringBootTest注解，确保Spring上下文正常初始化，Embedded Kafka可以正常启动
2. 直接注入EmbeddedKafkaBroker实例获取Broker信息，不要通过EmbeddedKafkaCondition间接获取，避免获取时机错误
3. 去掉消费模板的static修饰符，初始化逻辑放在@Before（JUnit4）或@BeforeEach（JUnit5）中执行，保证执行时Embedded Kafka已经完成启动

修正后的测试代码参考：

@EmbeddedKafka(topics = TestServiceTests.REACTIVE_INT_KEY_TOPIC, partitions = 2)
@ExtendWith(SpringExtension.class)
@EnableKafka
@SpringBootTest
@DirtiesContext
public class TestServiceTests {
    public static final String REACTIVE_INT_KEY_TOPIC = "reactive_int_key_topic";

    @Autowired
    private EmbeddedKafkaBroker embeddedKafkaBroker;

    private ReactiveKafkaConsumerTemplate<String, String> reactiveKafkaConsumerTemplate;

    @Autowired
    private ReactiveKafkaProducerTemplate<String, List<String>> reactiveKafkaProducerTemplate;

    @Before
    public void setUp() {
        Map<String, Object> consumerProps = KafkaTestUtils
                .consumerProps("reactive_consumer_group", "false", embeddedKafkaBroker);
        reactiveKafkaConsumerTemplate =
                new ReactiveKafkaConsumerTemplate<>(setupReceiverOptionsWithDefaultTopic(consumerProps));
    }


    private ReceiverOptions<String, String> setupReceiverOptionsWithDefaultTopic(
            Map<String, Object> consumerProps) {

        ReceiverOptions<String, String> basicReceiverOptions = ReceiverOptions.create(consumerProps);
        return basicReceiverOptions
                .consumerProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
                .addAssignListener(p -> assertThat(p.iterator().next().topicPartition().topic())
                        .isEqualTo(REACTIVE_INT_KEY_TOPIC))
                .subscription(Collections.singletonList(REACTIVE_INT_KEY_TOPIC));
    }

    @Test
    public void consumeServiceReturnsFlux() {
        TestService testService = new TestService(reactiveKafkaConsumerTemplate, reactiveKafkaProducerTemplate);
        Flux<String> actual = testService.consumeService();
        // 补充你的测试断言逻辑即可
    }
}

额外注意事项 #

• 如果使用JUnit 5，@ExtendWith(SpringExtension.class)可以直接删除，@SpringBootTest已经默认集成了该扩展
• 测试类中尽量避免使用静态变量存储业务Bean实例，防止不同测试用例执行时互相污染数据

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