切换为Docker本地Kafka集群后Kafka Streams启动即退出问题

• 核心原因是单节点本地Kafka集群不满足Kafka Streams默认的副本因子要求：Kafka Streams创建内部changelog主题、状态存储主题时默认使用3副本配置，你本地Docker部署的Kafka只有1个broker，主题创建失败会直接抛出致命错误，导致Streams启动后立刻终止。Confluent Cloud集群有多可用节点，满足3副本要求所以可以正常运行。
• 你的docker-compose配置中Kafka服务缺少监听器绑定参数，同时Streams启动代码没有添加优雅关闭钩子和主线程阻塞逻辑，异常抛出时没有保留足够的错误上下文，程序直接终止。
• 代码执行顺序为先运行Producer写入测试数据，再启动Streams实例，因此你能观察到数据生产成功的现象，随后Streams启动失败JVM直接退出。

1. 修正Docker Compose Kafka配置 #

找到kafka服务的environment配置段，补全监听器和事务日志相关配置，适配单节点环境：

environment:
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:29092,PLAINTEXT_HOST://0.0.0.0:9092
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
      - KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1
      - KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1

修改完成后执行以下命令重建集群，清除旧的错误配置残留：

docker-compose down -v
docker-compose up -d

注意：宿主机运行Java程序连接Kafka时，bootstrap.servers要配置为localhost:9092，Schema Registry地址配置为http://localhost:8081，不能使用容器内部主机名kafka、schema-registry，宿主机无法解析Docker内部网络域名。

2. 调整Kafka Streams配置适配单节点环境 #

在加载完Streams配置后，添加单节点环境专用参数，将内部主题副本因子设为1：

final Properties streamsProps = StreamsUtils.loadProperties();
streamsProps.put(StreamsConfig.APPLICATION_ID_CONFIG, "extractor-windowed-streams");
// 新增单节点适配配置
streamsProps.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 1);
// 以下为本地测试可选优化配置
streamsProps.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
streamsProps.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 10*1024*1024);

3. 补全Streams启动逻辑 #

原有代码调用kafkaStreams.start()后主线程直接执行完毕，没有优雅关闭逻辑，也没有等待Streams持续运行，替换main方法末尾的启动代码为标准实现：

KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), streamsProps);
// 注册JVM关闭钩子，实现优雅停机
Runtime.getRuntime().addShutdownHook(new Thread(kafkaStreams::close));

TopicLoader.runProducer();
kafkaStreams.start();

// 阻塞主线程，等待Streams持续运行
try {
    kafkaStreams.stateLock().lock();
    while (kafkaStreams.state().isRunningOrRebalancing()) {
        kafkaStreams.stateLock().await();
    }
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
} finally {
    kafkaStreams.stateLock().unlock();
}

本地测试时也可以简化为在kafkaStreams.start()后添加Thread.currentThread().join();实现主线程阻塞。

完成以上修改后重新运行程序，即可看到Streams实例持续运行，消费测试数据并输出窗口聚合结果，不会出现启动后立刻退出的问题。

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