Testcontainers-Java迁移KafkaContainer到ConfluentKafkaContainer的注意事项

在Spring Boot 3.4.3升级过程中，许多开发者会遇到需要将Testcontainers中的KafkaContainer迁移到ConfluentKafkaContainer的情况。这个迁移过程看似简单，但实际上存在一些关键配置差异需要注意，否则可能导致Schema Registry等组件无法正常连接Kafka服务。

问题背景

Testcontainers项目在1.20.5版本中标记KafkaContainer为废弃状态，推荐使用专门针对Confluent平台优化的ConfluentKafkaContainer。这个变更背后有几个技术考量：

1. KafkaContainer是基于Apache Kafka官方镜像设计的通用容器
2. ConfluentKafkaContainer则是针对Confluent平台(包含Schema Registry等组件)优化的专用实现
3. 两者在网络配置和端口分配策略上有显著差异

关键差异分析

从问题描述中可以看到，迁移后Schema Registry无法连接Kafka，核心原因在于网络配置的变化：

1. 端口分配策略不同：

   • KafkaContainer内部使用9092端口
   • ConfluentKafkaContainer内部使用9093端口(BROKER协议)

2. 监听器配置差异：

   // KafkaContainer典型配置
   advertised.listeners = PLAINTEXT://localhost:9092
   
   // ConfluentKafkaContainer典型配置
   advertised.listeners = PLAINTEXT://localhost:51721,BROKER://495d7a51aff8:9093
   

3. 网络协议变化：

   • KafkaContainer使用单一PLAINTEXT协议
   • ConfluentKafkaContainer区分BROKER(内部)和PLAINTEXT(外部)协议

解决方案

要使Schema Registry能够正确连接ConfluentKafkaContainer，需要进行以下调整：

1. 修改连接地址： 将Schema Registry的bootstrap.servers配置指向内部BROKER监听器：

   .withEnv(
       "SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS",
       "BROKER://" + kafka.networkAliases[0] + ":9093",
   )
   

2. 协议映射配置： 确保协议映射正确：

   .withEnv(
       "SCHEMA_REGISTRY_KAFKASTORE_SECURITY_PROTOCOL", 
       "PLAINTEXT"
   )
   

3. 完整配置示例：

   val schemaRegistry = GenericContainer(schemaRegistryImage)
       .withNetwork(network)
       .withExposedPorts(8081)
       .withEnv("SCHEMA_REGISTRY_HOST_NAME", "schema-registry")
       .withEnv("SCHEMA_REGISTRY_LISTENERS", "http://0.0.0.0:8081")
       .withEnv(
           "SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS",
           "BROKER://" + kafka.networkAliases[0] + ":9093"
       )
       .withEnv(
           "SCHEMA_REGISTRY_KAFKASTORE_SECURITY_PROTOCOL",
           "PLAINTEXT"
       )
       .waitingFor(Wait.forHttp("/subjects").forStatusCode(200))
   

深入理解

这个问题的本质在于ConfluentKafkaContainer采用了更接近生产环境的配置方式：

1. 安全隔离：区分内部(BROKER)和外部(PLAINTEXT)通信
2. 端口专业化：9093用于Broker间通信，9092用于客户端通信
3. 协议映射：通过listener.security.protocol.map配置不同协议的映射关系

对于测试环境，开发者也可以简化配置，强制使用单一协议：

val kafka = ConfluentKafkaContainer(kafkaImage)
    .withNetwork(network)
    .withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT")
    .withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "BROKER")

最佳实践

1. 版本对齐：确保Kafka镜像版本与Schema Registry版本兼容
2. 网络隔离：始终使用自定义Network确保容器间通信
3. 健康检查：为关键服务添加waitingFor条件
4. 日志分析：遇到问题时首先检查容器日志中的配置输出

通过理解这些底层机制，开发者可以更灵活地在测试环境中配置Confluent平台组件，确保集成测试的可靠性。