Spring Boot测试中EmbeddedKafka结合GraalVM超时问题求助

1. 补全GraalVM反射与资源元数据 #

EmbeddedKafka依赖的Kafka大量类通过反射加载，GraalVM原生编译默认不会生成这些类的元数据，导致程序卡住。可以通过两种方式配置：

• 手动添加JSON配置：在src/main/resources/META-INF/native-image下创建reflect-config.json和resource-config.json：
  reflect-config.json示例：

  [
    {
      "name": "org.apache.kafka.common.serialization.StringSerializer",
      "allDeclaredConstructors": true,
      "allPublicConstructors": true,
      "allDeclaredMethods": true,
      "allPublicMethods": true
    },
    {
      "name": "org.apache.kafka.common.serialization.StringDeserializer",
      "allDeclaredConstructors": true,
      "allPublicConstructors": true,
      "allDeclaredMethods": true,
      "allPublicMethods": true
    },
    {
      "name": "org.apache.kafka.clients.consumer.KafkaConsumer",
      "allDeclaredConstructors": true,
      "allPublicConstructors": true
    }
  ]
  

  resource-config.json示例：

  {
    "resources": [
      {"pattern": "\\Qkafka/server.properties\\E"},
      {"pattern": "\\Qkafka/client.properties\\E"}
    ]
  }
  

• 使用@NativeHint注解：在测试类上直接声明所需配置，无需手动写JSON：

  @NativeHint(
      types = {
          @TypeHint(typeNames = {"org.apache.kafka.common.serialization.StringSerializer", "org.apache.kafka.common.serialization.StringDeserializer"}, access = AccessBits.ALL),
          @TypeHint(typeNames = {"org.apache.kafka.clients.consumer.KafkaConsumer", "org.apache.kafka.clients.producer.KafkaProducer"}, access = AccessBits.ALL)
      },
      resources = @ResourceHint(patterns = {"kafka/server.properties", "kafka/client.properties"})
  )
  @SpringBootTest
  @EmbeddedKafka
  public class KafkaNativeTest {
      // 测试逻辑
  }
  

2. 调整EmbeddedKafka启动参数 #

GraalVM环境下，Kafka默认的线程模型和JMX配置可能引发阻塞，修改以下参数：

• 禁用JMX避免线程阻塞：

  @EmbeddedKafka(
      brokerProperties = {
          "jmx.port=-1",
          "listeners=PLAINTEXT://localhost:9092",
          "advertised.listeners=PLAINTEXT://localhost:9092"
      }
  )
  

• 降低线程池大小适配GraalVM资源限制：

  @EmbeddedKafka(
      brokerProperties = {
          "num.network.threads=1",
          "num.io.threads=1",
          "num.partitions=1"
      }
  )
  

3. 利用Spring Kafka官方原生支持 #

Spring Kafka 2.8+及Spring Boot 2.6+版本已提供GraalVM适配，确保依赖类被正确处理：

• 升级到对应版本的Spring Boot和Spring Kafka
• 测试类中开启controlledShutdown，避免资源泄漏导致的卡住：

  @EmbeddedKafka(controlledShutdown = true)
  

4. 原生测试中手动初始化Kafka组件 #

在GraalVM原生测试环境中，自动配置可能出现异常，手动初始化生产者和消费者：

@SpringBootTest
@EmbeddedKafka(topics = "test-topic")
public class KafkaNativeTest {

    @Autowired
    private EmbeddedKafkaBroker broker;

    private KafkaTemplate<String, String> kafkaTemplate;
    private Consumer<String, String> consumer;

    @BeforeEach
    void setUp() {
        // 初始化生产者
        Map<String, Object> producerProps = KafkaTestUtils.producerProps(broker);
        kafkaTemplate = new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(producerProps));

        // 初始化消费者
        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("test-group", "true", broker);
        consumer = new DefaultKafkaConsumerFactory<>(consumerProps).createConsumer();
        consumer.subscribe(Collections.singletonList("test-topic"));
    }

    @AfterEach
    void tearDown() {
        consumer.close(Duration.ofSeconds(1));
        kafkaTemplate.destroy();
    }

    // 测试用例
    @Test
    void testKafkaMessage() {
        kafkaTemplate.send("test-topic", "test-message");
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(2));
        assertThat(records.count()).isEqualTo(1);
    }
}

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