我在做一个测试，使用Apache Kafka 4.1.2。它需要在一个 local-health-status-v0 主题中发送一条消息。

该主题由Spring Boot 4.0.6 @Config 创建：

@Bean
public NewTopic healthStatusTopic() {
   String healthStatusTopicName = String.format(HEALTH_STATUS_TOPIC_FMT, this.environmentService.topicEnvironment(), this.environmentService.version());

   return TopicBuilder.name(healthStatusTopicName)
      .partitions(this.healthTopicPartitions)
      .replicas(this.healthTopicReplicas)
      .build();
}

并且成功。我可以执行：

kafka-topics.sh  --list --bootstrap-server localhost:9092

local-health-status-v0

该程序在这个主题中发送了一条消息，来自一个 @Service，使用 KafkaTemplate：

@Service
public class VivacityService {
/** Logger */
private static final Logger LOGGER = LoggerFactory.getLogger(VivacityService.class);

/** Kafka template */
private final KafkaTemplate<String, String> kafkaTemplate;

/** Topic name for the live status of members */
private static final String TOPIC = "local-health-status-v0";

@Autowired
public VivacityService(KafkaTemplate<String, String> kafkaTemplate) {
    this.kafkaTemplate = kafkaTemplate;
}

/**
 * Publish that the orchestrator is alive.
 */
public void alive() {
    String message = "Alive: orchestrator";
    LOGGER.info("Publishing: {}", message);

    CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(TOPIC, message);
    SendResult<String, String> r = future.join();
    LOGGER.info("Producer record: {}", r.getProducerRecord());
    LOGGER.info("Record Metadata: {}", r.getRecordMetadata());
}

它记录了一些日志，让我相信它已经成功：

[Producer clientId=beaufort-orchestrator-producer-1] Instantiated an idempotent produ
Kafka version: 4.1.2                                                                 
Kafka commitId: c82fd9b934b4c1e6                                                     
Kafka startTimeMs: 1780544373688                                                     
[Producer clientId=beaufort-orchestrator-producer-1] Cluster ID: 4L6G9nxwQreSVeTuSsh_
[Producer clientId=beaufort-orchestrator-producer-1] ProducerId set to 0 with epoch 0
Producer record: ProducerRecord(topic=local-health-status-v0, partition=null, headers
Record Metadata: local-health-status-v0-0@0

But a kafka-console-consumer.sh shows no messages in the topic:

kafka-console-consumer.sh  --topic local-health-status-v0 --from-beginning --bootstrap-server localhost:9092

Processed a total of 0 messages

如果我查看Apache Kafka的日志，它没有错误，但出现了数百条这样的信息：

INFO Sent auto-creation request for Set(__consumer_offsets) to the active controller. (kafka.server.DefaultAutoTopicCreationManager)
INFO Sent auto-creation request for Set(__consumer_offsets) to the active controller. (kafka.server.DefaultAutoTopicCreationManager)
INFO Sent auto-creation request for Set(__consumer_offsets) to the active controller. (kafka.server.DefaultAutoTopicCreationManager)
INFO Sent auto-creation request for Set(__consumer_offsets) to the active controller. (kafka.server.DefaultAutoTopicCreationManager)

我的Apache Kafka：

KAFKA_VERSION="4.1.2"
CONTAINER_NAME="kafka-beaufort"
CLUSTER_ID="4L6G9nxwQreSVeTuSsh_Hg"

docker run -d \
  --name $CONTAINER_NAME \
  -p 9092:9092 \
  -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_NODE_ID=1 \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
  -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e CLUSTER_ID=$CLUSTER_ID \
  apache/kafka:$KAFKA_VERSION

解决方案

根本原因是在单节点KRaft集群（没有ZooKeeper）的情况下，Kafka无法创建它内部的 __consumer_offsets 主题。该主题供消费者跟踪他们已读取的消息。默认情况下，该主题需要3 个副本 (offsets.topic.replication.factor=3)，但你只有一个broker——因此Kafka会一直重试，这恰好就是你在日志中看到的情况：

INFO发送自动创建请求Set(__consumer_offsets) 给活动控制器。 (kafka.server.DefaultAutoTopicCreationManager)

要修复它，请在你的docker run命令中添加以下环境变量：

KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1

别忘了先删除旧容器，否则破损状态会一直存在。