升级Spring Boot与Kafka后@EmbeddedKafka测试Broker创建超时问题

将Spring Boot从2.4版本升级至2.7.18、Kafka从2.8.2升级至3.6.1后，运行带有@EmbeddedKafka注解的单元测试时，创建EmbeddedKafka Broker触发TimeoutException，导致ApplicationContext加载失败。

测试代码 #

@RunWith(SpringRunner.class)
@SpringBootTest
@WebAppConfiguration
@DirtiesContext(classMode = DirtiesContext.ClassMode.BEFORE_EACH_TEST_METHOD)
@ActiveProfiles("inmemory")
@EmbeddedKafka(topics={"KafkaDataHelperNonSpyUnitTest.topic"},
               adminTimeout = 30)
@TestPropertySource(value = "classpath:hsql-test.properties",
                    properties = "spring.cache.type=NONE")
public class KafkaDataHelperNonSpyUnitTest
{
  /*
   * Setup Beans Here
   */  

   @Test
   public void testGetConsumer()
   {
      Map<String, Object> props = new HashMap<>();
      props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedBrokers);
      props.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class);
      props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class);
      props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 5);

      kafkaDataHelper.getConsumer("KafkaDataHelperNonSpyUnitTest.topic", props);
   }
}

报错堆栈 #

Failed to load ApplicationContext
java.lang.IllegalStateException: Failed to load ApplicationContext
    at org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate.loadContext(DefaultCacheAwareContextLoaderDelegate.java:98)
    at org.springframework.test.context.support.DefaultTestContext.getApplicationContext(DefaultTestContext.java:124)
    at org.springframework.test.context.web.ServletTestExecutionListener.setUpRequestContextIfNecessary(ServletTestExecutionListener.java:190)
    at org.springframework.test.context.web.ServletTestExecutionListener.prepareTestInstance(ServletTestExecutionListener.java:132)
    at org.springframework.test.context.TestContextManager.prepareTestInstance(TestContextManager.java:248)
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.createTest(SpringJUnit4ClassRunner.java:227)
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner$1.runReflectiveCall(SpringJUnit4ClassRunner.java:289)
    at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.methodBlock(SpringJUnit4ClassRunner.java:291)
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:246)
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:97)
    at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331)
    at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
    at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329)
    at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
    at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293)
    at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61)
    at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70)
    at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
    at org.junit.runners.ParentRunner.run(ParentRunner.java:413)
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:190)
    at org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.runTestClass(JUnitTestClassExecutor.java:110)
    at org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:58)
    at org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:38)
    at org.gradle.api.internal.tasks.testing.junit.AbstractJUnitTestClassProcessor.processTestClass(AbstractJUnitTestClassProcessor.java:62)
    at org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.processTestClass(SuiteTestClassProcessor.java:51)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36)
    at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
    at org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:33)
    at org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:94)
    at com.sun.proxy.$Proxy2.processTestClass(Unknown Source)
    at org.gradle.api.internal.tasks.testing.worker.TestWorker$2.run(TestWorker.java:176)
    at org.gradle.api.internal.tasks.testing.worker.TestWorker.executeAndMaintainThreadName(TestWorker.java:129)
    at org.gradle.api.internal.tasks.testing.worker.TestWorker.execute(TestWorker.java:100)
    at org.gradle.api.internal.tasks.testing.worker.TestWorker.execute(TestWorker.java:60)
    at org.gradle.process.internal.worker.child.ActionExecutionWorker.execute(ActionExecutionWorker.java:56)
    at org.gradle.process.internal.worker.child.SystemApplicationClassLoaderWorker.call(SystemApplicationClassLoaderWorker.java:133)
    at org.gradle.process.internal.worker.child.SystemApplicationClassLoaderWorker.call(SystemApplicationClassLoaderWorker.java:71)
    at worker.org.gradle.process.internal.worker.GradleWorkerMain.run(GradleWorkerMain.java:69)
    at worker.org.gradle.process.internal.worker.GradleWorkerMain.main(GradleWorkerMain.java:74)
Caused by: org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'embeddedKafka': Invocation of init method failed; nested exception is org.apache.kafka.common.KafkaException: java.util.concurrent.TimeoutException
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1804)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:431)
    at org.springframework.kafka.test.context.EmbeddedKafkaContextCustomizer.customizeContext(EmbeddedKafkaContextCustomizer.java:116)
    at org.springframework.boot.test.context.SpringBootContextLoader$ContextCustomizerAdapter.initialize(SpringBootContextLoader.java:333)
    at org.springframework.boot.SpringApplication.applyInitializers(SpringApplication.java:605)
    at org.springframework.boot.SpringApplication.prepareContext(SpringApplication.java:374)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:307)
    at org.springframework.boot.test.context.SpringBootContextLoader.loadContext(SpringBootContextLoader.java:136)
    at org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate.loadContextInternal(DefaultCacheAwareContextLoaderDelegate.java:141)
    at org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate.loadContext(DefaultCacheAwareContextLoaderDelegate.java:90)
    ... 43 more
Caused by: org.apache.kafka.common.KafkaException: java.util.concurrent.TimeoutException
    at org.springframework.kafka.test.EmbeddedKafkaBroker.createTopics(EmbeddedKafkaBroker.java:490)
    at org.springframework.kafka.test.EmbeddedKafkaBroker.lambda$createKafkaTopics$8(EmbeddedKafkaBroker.java:477)
    at org.springframework.kafka.test.EmbeddedKafkaBroker.doWithAdmin(EmbeddedKafkaBroker.java:573)
    at org.springframework.kafka.test.EmbeddedKafkaBroker.createKafkaTopics(EmbeddedKafkaBroker.java:476)
    at org.springframework.kafka.test.EmbeddedKafkaBroker.afterPropertiesSet(EmbeddedKafkaBroker.java:373)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.invokeInitMethods(AbstractAutowireCapableBeanFactory.java:1863)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1800)
    ... 52 more
Caused by: java.util.concurrent.TimeoutException
    at java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1784)
    at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1928)
    at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:180)
    at org.springframework.kafka.test.EmbeddedKafkaBroker.createTopics(EmbeddedKafkaBroker.java:487)
    ... 58 more

相关build.gradle依赖片段 #

buildscript
{
   ext { springBootVersion = '2.7.18' }

   dependencies
   {
      classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
   }
}

apply plugin: "org.springframework.boot"
apply plugin: 'io.spring.dependency-management'

dependencies
{
    // Spring Framework (Core)
    def springFrameworkVersion = 5.3.27
    implementation ('org.springframework:spring-core')
    { version { strictly springFrameworkVersion } }
    implementation ('org.springframework:spring-aop')
    { version { strictly springFrameworkVersion } }
    implementation ('org.springframework:spring-context')
    { version { strictly springFrameworkVersion } }
    implementation('org.springframework:spring-context-support')
    { version { strictly springFrameworkVersion } }
    implementation ('org.springframework:spring-beans')
    { version { strictly springFrameworkVersion } }
    implementation ('org.springframework:spring-messaging')
    { version { strictly springFrameworkVersion } }
    implementation ('org.springframework:spring-expression')
    { version { strictly springFrameworkVersion } }

    // Spring Framework (Data)
    implementation('org.springframework:spring-jdbc')
    { version { strictly springFrameworkVersion } }
    implementation('org.springframework:spring-tx')
    { version { strictly springFrameworkVersion } }

    // Spring Framework (Web)
    implementation ('org.springframework:spring-web')
    { version { strictly springFrameworkVersion } }
    implementation ('org.springframework:spring-websocket')
    { version { strictly springFrameworkVersion } }
    implementation ('org.springframework:spring-webmvc')
    { version { strictly springFrameworkVersion } }
    implementation ('org.springframework:spring-webflux')
    { version { strictly springFrameworkVersion } }

    implementation "org.springframework.boot:spring-boot-starter-web:${springBootVersion}"
    testImplementation "org.springframework.boot:spring-boot-starter-test:${springBootVersion}"

    // Spring Framework (Security)
    def springSecurityVersion = '5.7.11'
    implementation("org.springframework.security:spring-security-core")
    { version { strictly springSecurityVersion } }
    implementation("org.springframework.security:spring-security-web")
    { version { strictly springSecurityVersion } }
    implementation("org.springframework.security:spring-security-ldap")
    { version { strictly springSecurityVersion } }
    implementation("org.springframework.security:spring-security-crypto")
    { version { strictly springSecurityVersion } }
    testImplementation("org.springframework.security:spring-security-test")
    { version { strictly springSecurityVersion } }

    implementation "org.springframework.security.oauth:spring-security-oauth2:2.5.2.RELEASE"

    implementation"org.springframework.boot:spring-boot-starter-actuator:${springBootVersion}"
    implementation('org.springframework.retry:spring-retry:1.3.3')

    runtimeOnly("org.springframework.boot:spring-boot-properties-migrator:${springBootVersion}")

    // Kafka
    def KafkaClientVersion = '3.6.1'
    implementation('org.springframework.kafka:spring-kafka:')
    { version { strictly('2.9.13') } }
    testImplementation('org.springframework.kafka:spring-kafka-test')
    { version { strictly('2.9.13') } }
    implementation('org.apache.kafka:kafka-clients')
    { version { strictly(KafkaClientVersion) } }
    testImplementation('org.apache.kafka:kafka-streams')
    { version { strictly(KafkaClientVersion) } }
    implementation('org.apache.kafka:kafka-server-common')
    { version { strictly(KafkaClientVersion) } }
    testImplementation('org.apache.kafka:kafka-server-common')
    { version { strictly(KafkaClientVersion) } }
    implementation('org.apache.kafka:kafka-storage')
    { version { strictly(KafkaClientVersion) } }
    testImplementation('org.apache.kafka:kafka-tools')
    { version { strictly(KafkaClientVersion) } }
    testImplementation('org.apache.kafka:connect-json')
    { version { strictly(KafkaClientVersion) } }
    testImplementation('org.apache.kafka:connect-api')
    { version { strictly(KafkaClientVersion) } }

    testImplementation("org.apache.kafka:kafka_2.13:${KafkaClientVersion}:test")
}

问题原因 #

1. 版本兼容性适配问题：Spring Kafka 2.9.x与Kafka 3.6.x的底层API交互存在细微适配差异，新版本Kafka的AdminClient对超时的敏感度更高，默认配置下容易触发超时。
2. 超时配置不匹配：仅设置@EmbeddedKafka的adminTimeout不足以覆盖AdminClient的所有超时参数，底层创建Topic的请求可能受限于request.timeout.ms等未调整的参数。
3. 测试资源不足：新版本Kafka的Embedded Broker启动需要更多内存和CPU资源，测试JVM默认配置不足会导致Broker启动缓慢，进而触发超时。
4. 上下文重建频繁：@DirtiesContext(classMode = BEFORE_EACH_TEST_METHOD)会在每次测试前重建ApplicationContext，频繁创建和销毁Embedded Kafka Broker，加剧资源竞争。

解决办法 #

1. 优化版本依赖管理 #

移除不必要的strictly版本锁定，让Spring Boot的依赖管理插件自动维护Spring Kafka与Kafka客户端的兼容版本：

// 替换原有Kafka依赖配置
implementation 'org.springframework.kafka:spring-kafka'
testImplementation 'org.springframework.kafka:spring-kafka-test'
implementation 'org.apache.kafka:kafka-clients'
testImplementation 'org.apache.kafka:kafka-streams'
// 移除其他Kafka相关组件的strictly锁定，由Spring Dependency Management自动管理

2. 增强EmbeddedKafka超时配置 #

在@EmbeddedKafka注解中增加Broker属性，调整AdminClient的超时参数：

@EmbeddedKafka(
    topics={"KafkaDataHelperNonSpyUnitTest.topic"},
    adminTimeout = 60,
    brokerProperties = {
        "admin.client.configs=request.timeout.ms=60000,default.api.timeout.ms=60000",
        "replica.lag.time.max.ms=60000"
    }
)

或者在测试属性文件hsql-test.properties中添加：

spring.kafka.admin.properties.request.timeout.ms=60000
spring.kafka.admin.properties.default.api.timeout.ms=60000

3. 增加测试JVM资源配置 #

在build.gradle的测试任务中增加内存分配：

test {
    jvmArgs '-Xmx1024m', '-Xms512m'
    // 可选：增加测试超时时间
    timeout = 120
}

4. 调整上下文重建策略 #

将@DirtiesContext的classMode改为BEFORE_CLASS，减少上下文重建频率：

@DirtiesContext(classMode = DirtiesContext.ClassMode.BEFORE_CLASS)

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