.NET中TestContainers部署Kafka+Schema Registry：无法解析Kafka主机

尝试通过TestContainers搭建Kafka与Schema Registry环境，容器已成功启动，但从容器外部操作时出现异常：连接Kafka Admin Client后，获取元数据等命令可正常执行，但创建Topic的命令超时失败，报错无法解析kafka:9092主机。

报错信息：

[thrd:kafka:9092/1]: kafka:9092/1: Failed to resolve 'kafka:9092': Name of service not known (after 12ms in state CONNECT)
[thrd:app]: rdkafka#producer-1: kafka:9092/1: Failed to resolve 'kafka:9092': Name or service not known (after 12ms in state CONNECT))

用户配置代码：

[Test]
public async Test my_setup()
{
    var network = new NetworkBuilder().WithName(Guid.NewGuid().ToString("D")).Build();
    await network.CreateAsync();

    var kafkaContainer = BuildKafkaContainer(network);
    await _kafkaContainer.StartAsync();

    var schemaRegistryContainer = BuildSchemaRegistryContainer(network);
    await _schemaRegistryContainer.StartAsync();

    var kafkaAdminClient = new AdminClientBuilder(new AdminClientConfig {
        BootstrapServers = kafkaContainer.GetBootstrapAddress(),
        SecurityProtocol = SecurityProtocol.Plaintext,
    }).Build();

    // CALL 1: 该命令可正常执行，返回预期的空数据
    var meta = kafkaAdminClient.GetMetadata(TimeSpan.FromSeconds(10));

    // CALL 2: 该命令最终超时，调试输出中的错误信息如下：
    //
    // [thrd:kafka:9092/1]: kafka:9092/1: Failed to resolve 'kafka:9092': Name of service not known (after 12ms in state CONNECT)"
    // [thrd:app]: rdkafka#producer-1: kafka:9092/1: Failed to resolve 'kafka:9092': Name or service not known (after 12ms in state CONNECT))
    // .. 消息重复 ..
    await KafkaAdminClient.CreateTopicsAsync(new[] { new TopicSpecification
    {
        Name = KafkaTopicName,
        ReplicationFactor = 1,
        NumPartitions = 1,
    }}, new CreateTopicsOptions
    {
        OperationTimeout = TimeSpan.FromSeconds(60),
        RequestTimeout = TimeSpan.FromSeconds(60)
    });
}

private KafkaContainer BuildKafkaContainer(INetwork network)
{
    return new KafkaBuilder()
        .WithImage("confluentinc/cp-kafka:7.5.2")
        .WithNetworkAliases("kafka")
        .WithNetwork(network)
        .WithEnvironment("KAFKA_LISTENERS", $"PLAINTEXT://kafka:{KafkaBuilder.KafkaPort},BROKER://0.0.0.0:{KafkaBuilder.BrokerPort}")
        .WithStartupCallback((container, ct) =>
        {
            // 与默认WithStartupCallback实现一致，直到...
            const char lf = '\n';
            var startupScript = new StringBuilder();
            startupScript.Append("#!/bin/bash");
            startupScript.Append(lf);
            startupScript.Append($"echo 'clientPort={KafkaBuilder.ZookeeperPort}' > zookeeper.properties");
            startupScript.Append(lf);
            startupScript.Append("echo 'dataDir=/var/lib/zookeeper/data' >> zookeeper.properties");
            startupScript.Append(lf);
            startupScript.Append("echo 'dataLogDir=/var/lib/zookeeper/log' >> zookeeper.properties");
            startupScript.Append(lf);
            startupScript.Append("zookeeper-server-start zookeeper.properties &");
            startupScript.Append(lf);

            // ... 到此处：我们覆盖KAFKA_ADVERTISED_LISTENERS
            startupScript.Append($"export KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:{KafkaBuilder.KafkaPort},BROKER://{container.IpAddress}:{KafkaBuilder.BrokerPort}");

            // ... 回到复制默认实现
            startupScript.Append(lf);
            startupScript.Append("echo '' > /etc/confluent/docker/ensure");
            startupScript.Append(lf);
            startupScript.Append("/etc/confluent/docker/run");
            return container.CopyAsync(Encoding.Default.GetBytes(startupScript.ToString()), KafkaBuilder.StartupScriptFilePath, Unix.FileMode755, ct);
        })
        .Build();
}

private IContainer BuildSchemaRegistryContainer(INetwork network)
{
    return new ContainerBuilder()
        .WithImage("confluentinc/cp-schema-registry:7.5.2")
        .WithNetwork(network)
        .WithExposedPort(8081)
        .WithPortBinding(8081, true)
        .WithEnvironment("SCHEMA_REGISTRY_HOST_NAME", "schema-registry")
        .WithEnvironment("SCHEMA_REGISTRY_LISTENERS", "http://0.0.0.0:8081")
        .WithEnvironment("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", $"PLAINTEXT://kafka:{KafkaBuilder.KafkaPort}")
        .WithWaitStrategy(Wait.ForUnixContainer().UntilMessageIsLogged("Server started, listening for requests..."))
        .Build();
}

核心原因 #

Kafka的ADVERTISED_LISTENERS配置错误导致：

1. 第一次获取元数据成功，是因为你用GetBootstrapAddress()拿到了容器映射到主机的地址，请求能正常到达Kafka容器。
2. 但Kafka返回的元数据里，Broker地址是kafka:9092（来自PLAINTEXT://kafka:{KafkaBuilder.KafkaPort}这条广告监听配置），主机环境无法解析Docker内部的网络别名，后续创建Topic的请求尝试连接该地址时就会出现域名解析失败。

解决步骤 #

1. 修正ADVERTISED_LISTENERS配置 #

需要让Kafka同时提供两种可访问地址：

• 内部容器通信（如Schema Registry连接Kafka）：保留kafka:9092
• 主机外部访问：使用主机可解析的地址（比如localhost:映射端口）

修改BuildKafkaContainer中的export KAFKA_ADVERTISED_LISTENERS语句：

export KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:${KafkaBuilder.KafkaPort},BROKER://localhost:${KafkaBuilder.BrokerPort}

这样Kafka返回的元数据会包含localhost:BrokerPort，主机环境可以直接访问。

2. 简化配置（可选） #

无需自定义StartupCallback，TestContainers的KafkaBuilder提供了更简洁的配置方式，自动处理监听和广告地址：

private KafkaContainer BuildKafkaContainer(INetwork network)
{
    return new KafkaBuilder()
        .WithImage("confluentinc/cp-kafka:7.5.2")
        .WithNetworkAliases("kafka")
        .WithNetwork(network)
        .WithListener("PLAINTEXT", KafkaBuilder.KafkaPort)
        .WithExternalListener()
        .Build();
}

这种方式会自动正确配置LISTENERS和ADVERTISED_LISTENERS，避免手动配置出错。

3. 验证配置 #

修改后重新运行测试：

• 检查获取元数据返回的Broker地址是否为主机可访问的地址（如localhost:xxxx）
• 创建Topic的请求会使用正确地址连接，不再出现解析错误

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