Buildkite流水线中如何配置Kafka容器的BootstrapServers？

我使用Test Containers搭建了如下简化版的集成测试环境：

public class IntegrationTestFactory : WebApplicationFactory<Program>, IAsyncLifetime
{
    private static readonly string NetworkName = Guid.NewGuid().ToString();

    private readonly INetwork _network = new NetworkBuilder()
        .WithName(NetworkName)
        .Build();

    private readonly PostgreSqlContainer _postgreSqlContainer = new PostgreSqlBuilder()
        .WithImage("postgres:14.7")
        .WithName("psql")
        .WithHostname("psql")
        .WithCleanUp(true)
        .WithWaitStrategy(Wait.ForUnixContainer().UntilPortIsAvailable(5432))
        .WithPortBinding(5432, 5432)
        .WithEnvironment(
            new ReadOnlyDictionary<string, string>(new Dictionary<string, string>
            {
                { "POSTGRES_DB", "test-db" },
                { "POSTGRES_USER", "test-db-user" },
                { "POSTGRES_PASSWORD", "test-db-password" }
            }))
        .WithNetwork(NetworkName)
        .Build();

    private readonly IContainer _zookeeperContainer = new ContainerBuilder()
        .WithImage("confluentinc/cp-zookeeper:5.5.0")
        .WithHostname("zookeeper")
        .WithName("zookeeper")
        .WithCleanUp(true)
        .WithEnvironment(
            new ReadOnlyDictionary<string, string>(new Dictionary<string, string>
            {
                { "ZOOKEEPER_CLIENT_PORT", "2181" },
                { "ZOOKEEPER_TICK_TIME", "2000" }
            }))
        .WithPortBinding(2181, 2181)
        .WithWaitStrategy(Wait.ForUnixContainer().UntilPortIsAvailable(2181))
        .WithNetwork(NetworkName)
        .Build();

    private readonly IContainer _kafkaContainer = new ContainerBuilder()
        .WithImage("confluentinc/cp-kafka:5.5.0")
        .WithHostname("broker")
        .WithName("broker")
        .WithNetwork(NetworkName)
        .WithEnvironment(
            new ReadOnlyDictionary<string, string>(new Dictionary<string, string>
            {
                { "KAFKA_BROKER_ID", "1" },
                { "KAFKA_ZOOKEEPER_CONNECT", "zookeeper:2181" },
                { "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT" },
                { "KAFKA_LISTENERS", "PLAINTEXT://broker:29092,PLAINTEXT_HOST://0.0.0.0:9092" },
                { "KAFKA_ADVERTISED_LISTENERS", "PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092" },
                { "KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1" },
                { "KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "0" },
                { "KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1" },
                { "KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1" },
                { "KAFKA_AUTO_CREATE_TOPICS_ENABLE", "true" }
           }))
        .WithPortBinding(9092, 9092)
        .WithWaitStrategy(Wait.ForUnixContainer().UntilPortIsAvailable(9092))
        .WithCleanUp(true)
        .Build();

    private readonly IContainer _schemaRegistryContainer = new ContainerBuilder()
        .WithImage("confluentinc/cp-schema-registry:5.5.0")
        .WithHostname("schema-registry")
        .WithName("schema-registry")
        .WithCleanUp(true)
        .WithEnvironment(
            new ReadOnlyDictionary<string, string>(new Dictionary<string, string>
            {
                { "SCHEMA_REGISTRY_HOST_NAME", "schema-registry" },
                { "SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "broker:29092" },
                { "SCHEMA_REGISTRY_LISTENERS", "http://0.0.0.0:8081" }
            }))
        .WithPortBinding(8081, 8081)
        .WithWaitStrategy(Wait.ForUnixContainer().UntilPortIsAvailable(8081))
        .WithNetwork(NetworkName)
        .Build();

    private readonly IContainer _restProxyContainer = new ContainerBuilder()
        .WithImage("confluentinc/cp-kafka-rest:6.2.0")
        .WithHostname("rest-proxy")
        .WithName("rest-proxy")
        .WithCleanUp(true)
        .WithEnvironment(
            new ReadOnlyDictionary<string, string>(new Dictionary<string, string>
            {
                { "KAFKA_REST_HOST_NAME", "rest-proxy" },
                { "KAFKA_REST_BOOTSTRAP_SERVERS", "broker:29092" },
                { "KAFKA_REST_LISTENERS", "http://0.0.0.0:8082" },
                { "KAFKA_REST_SCHEMA_REGISTRY_URL", "http://schema-registry:8081" }
            }))
        .WithPortBinding(8082, 8082)
        .WithWaitStrategy(Wait.ForUnixContainer().UntilPortIsAvailable(8082))
        .WithNetwork(NetworkName)
        .Build();

    protected override void ConfigureWebHost(IWebHostBuilder builder)
    {
        builder.ConfigureTestServices(services =>
        {
            var isBuildkite = Environment.GetEnvironmentVariable("BUILDKITE") == "true";
            var host = isBuildkite ? "172.17.0.1" : "localhost";
        
            // Replace DbContext
            AppContext.SetSwitch("Npgsql.EnableLegacyTimestampBehavior", true);
            services.RemoveDbContext<ApplicationDbContext>();
            var appConnectionString =
             $"Host={host};Port=5432;Database=test-db;Username=test-db-user;Password=test-db-password;";
        
            services.AddDbContext<ApplicationDbContext>(options => options.UseNpgsql(appConnectionString, sqlOptions =>
            {
                sqlOptions.EnableRetryOnFailure();
                sqlOptions.UseNodaTime();
            }));
        
            // Override Kafka config
            services.AddKafkaProducer(new ConfigurationBuilder().Build(), options =>
            {
                options.TopicName = "MyTopicName";
                options.RetryCount = 1;
                options.RetryTimeoutInSeconds = 1;
                options.BootstrapServers = $"{host}:9092";
                options.SchemaRegistry = $"http://{host}:8081";
            });
        });
    }
}

本地环境使用localhost作为主机值可正常运行集成测试，但在Buildkite流水线中，使用172.17.0.1可连接PostgreSQL实例，却无法访问Kafka Broker。尝试了172.17.0.1:9092、172.17.0.1:29092、broker:9092等多个BootstrapServers值均失败，使用172.17.0.1:9092时出现错误：

localhost:9092: Connect to ipv4#127.0.0.1:9092 failed: Connection refused (after 0ms in state CONNECT)

请问在Buildkite流水线中，正确的BootstrapServers值应为多少？

解决方案 #

问题出在Kafka的KAFKA_ADVERTISED_LISTENERS配置上。你当前设置的PLAINTEXT_HOST://localhost:9092会导致：当测试程序通过172.17.0.1:9092连接到Kafka后，Kafka会返回localhost:9092作为实际通信地址，但在Buildkite环境里，localhost指向的是测试程序所在的主机/容器，而非Kafka容器，因此出现连接拒绝。

具体修改步骤 #

1. 调整Kafka容器的广告监听地址：将KAFKA_ADVERTISED_LISTENERS中的localhost替换为Buildkite环境下能访问到Kafka的主机地址（即172.17.0.1），保留内部通信的PLAINTEXT://broker:29092不变。

修改后的Kafka容器环境变量片段：

.WithEnvironment(
    new ReadOnlyDictionary<string, string>(new Dictionary<string, string>
    {
        // 其他配置保持不变
        { "KAFKA_ADVERTISED_LISTENERS", "PLAINTEXT://broker:29092,PLAINTEXT_HOST://172.17.0.1:9092" },
        // 其他配置保持不变
   }))

2. 测试程序的BootstrapServers保持不变：仍然使用172.17.0.1:9092作为BootstrapServers值。

更灵活的配置方案 #

为了避免硬编码地址，可通过环境变量动态设置广告地址：

• 在Buildkite流水线中设置环境变量KAFKA_ADVERTISED_HOST=172.17.0.1
• 代码中读取该变量，默认使用localhost：

var advertisedHost = Environment.GetEnvironmentVariable("KAFKA_ADVERTISED_HOST") ?? "localhost";
// ...
{ "KAFKA_ADVERTISED_LISTENERS", $"PLAINTEXT://broker:29092,PLAINTEXT_HOST://{advertisedHost}:9092" },

这样本地和Buildkite环境都能自动适配，无需修改代码。

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