java - 在动态创建的主题上向 Kafka 发送消息会出现错误 LEADER_NOT_AVAILABLE

Question

我正在尝试使用以下代码发送有关动态创建的主题的消息:

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

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void send(String topic, String payload) {
        String topicName =  topic + Integer.valueOf(RandomStringUtils.randomNumeric(1, 4))%100;
        kafkaTemplate.send(topicName, payload);
        LOGGER.info("sent payload to topic='{}'", topicName);
    }
}

当我用 topic = 'somethingnew' 调用这个 send 方法时，我得到以下错误，(有趣的是它能够发送一些消息，但对于许多它给出错误，并且在一段时间后，它开始发送消息也在那些主题上)但是当我重新启动服务器，并使用 topic = 'somethingnew' 再次调用发送方法时，它工作正常!卡夫卡有一些潜在的错误吗？我使用的是 spring boot 版本:1.5.4.RELEASE。

2018-02-13 09:52:35.927  INFO [betsync-adapter-service,,,] 22241 --- [      Thread-28] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 0.10.1.1
2018-02-13 09:52:35.927  INFO [betsync-adapter-service,,,] 22241 --- [      Thread-28] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : f10ef2720b03b247
2018-02-13 09:52:36.067  WARN [betsync-adapter-service,,,] 22241 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : Error while fetching metadata with correlation id 0 : {betsyncDataTopicNew123.5=LEADER_NOT_AVAILABLE}
2018-02-13 09:52:36.272  INFO [betsync-adapter-service,,,] 22241 --- [      Thread-28] c.b.b.a.k.message.KafkaMessageSender     : sent payload to topic='betsyncDataTopicNew123.5'
2018-02-13 09:52:36.372  WARN [betsync-adapter-service,,,] 22241 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : Error while fetching metadata with correlation id 3 : {betsyncDataTopicNew123.93=LEADER_NOT_AVAILABLE}
2018-02-13 09:52:36.475  INFO [betsync-adapter-service,,,] 22241 --- [      Thread-28] c.b.b.a.k.message.KafkaMessageSender     : sent payload to topic='betsyncDataTopicNew123.93'
2018-02-13 09:52:36.583  WARN [betsync-adapter-service,,,] 22241 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : Error while fetching metadata with correlation id 6 : {betsyncDataTopicNew123.4=LEADER_NOT_AVAILABLE}
2018-02-13 09:52:36.686  INFO [betsync-adapter-service,,,] 22241 --- [      Thread-28] c.b.b.a.k.message.KafkaMessageSender     : sent payload to topic='betsyncDataTopicNew123.4'
2018-02-13 09:52:36.794  WARN [betsync-adapter-service,,,] 22241 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : Error while fetching metadata with correlation id 9 : {betsyncDataTopicNew123.7=LEADER_NOT_AVAILABLE}
2018-02-13 09:52:36.897  INFO [betsync-adapter-service,,,] 22241 --- [      Thread-28] c.b.b.a.k.message.KafkaMessageSender     : sent payload to topic='betsyncDataTopicNew123.7'
2018-02-13 09:52:37.008  WARN [betsync-adapter-service,,,] 22241 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : Error while fetching metadata with correlation id 12 : {betsyncDataTopicNew123.0=LEADER_NOT_AVAILABLE}
2018-02-13 09:52:37.110  INFO [betsync-adapter-service,,,] 22241 --- [      Thread-28] c.b.b.a.k.message.KafkaMessageSender     : sent payload to topic='betsyncDataTopicNew123.0'
2018-02-13 09:52:37.220  WARN [betsync-adapter-service,,,] 22241 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : Error while fetching metadata with correlation id 15 : {betsyncDataTopicNew123.1=LEADER_NOT_AVAILABLE}
2018-02-13 09:52:37.320  INFO [betsync-adapter-service,,,] 22241 --- [      Thread-28] c.b.b.a.k.message.KafkaMessageSender     : sent payload to topic='betsyncDataTopicNew123.1'
2018-02-13 09:52:37.431  WARN [betsync-adapter-service,,,] 22241 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : Error while fetching metadata with correlation id 18 : {betsyncDataTopicNew123.10=LEADER_NOT_AVAILABLE}
2018-02-13 09:52:37.534  INFO [betsync-adapter-service,,,] 22241 --- [      Thread-28] c.b.b.a.k.message.KafkaMessageSender     : sent payload to topic='betsyncDataTopicNew123.10'
2018-02-13 09:52:37.534  INFO [betsync-adapter-service,,,] 22241 --- [      Thread-28] c.b.b.a.k.message.KafkaMessageSender     : sent payload to topic='betsyncDataTopicNew123.4'
2018-02-13 09:52:37.642  WARN [betsync-adapter-service,,,] 22241 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : Error while fetching metadata with correlation id 22 : {betsyncDataTopicNew123.71=LEADER_NOT_AVAILABLE}
2018-02-13 09:52:37.747  INFO [betsync-adapter-service,,,] 22241 --- [      Thread-28] c.b.b.a.k.message.KafkaMessageSender     : sent payload to topic='betsyncDataTopicNew123.71'
2018-02-13 09:52:37.859  WARN [betsync-adapter-service,,,] 22241 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : Error while fetching metadata with correlation id 25 : {betsyncDataTopicNew123.2=LEADER_NOT_AVAILABLE}
2018-02-13 09:52:37.962  INFO [betsync-adapter-service,,,] 22241 --- [      Thread-28] c.b.b.a.k.message.KafkaMessageSender     : sent payload to topic='betsyncDataTopicNew123.2'
2018-02-13 09:52:37.963  INFO [betsync-adapter-service,,,] 22241 --- [      Thread-28] c.b.b.a.k.message.KafkaMessageSender     : sent payload to topic='betsyncDataTopicNew123.7'
2018-02-13 09:52:38.074  WARN [betsync-adapter-service,,,] 22241 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : Error while fetching metadata with correlation id 29 : {betsyncDataTopicNew123.33=LEADER_NOT_AVAILABLE}
2018-02-13 09:52:38.176  INFO [betsync-adapter-service,,,] 22241 --- [      Thread-28] c.b.b.a.k.message.KafkaMessageSender     : sent payload to topic='betsyncDataTopicNew123.33'
2018-02-13 09:52:38.176  INFO [betsync-adapter-service,,,] 22241 --- [      Thread-28] c.b.b.a.k.message.KafkaMessageSender     : sent payload to topic='betsyncDataTopicNew123.1'
2018-02-13 09:52:38.285  WARN [betsync-adapter-service,,,] 22241 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : Error while fetching metadata with correlation id 33 : {betsyncDataTopicNew123.17=LEADER_NOT_AVAILABLE}

这只是警告，我们没有收到任何异常，所以不知道消息是否丢失。

Answer 1

使用 AdminClient按需创建主题。

Spring for Apache Kafka 提供了一个方便的KafkaAdmin可以为 NewTopic 创建主题beans 在应用程序上下文中，但它也可用于创建 AdminClient这样您就可以手动创建主题 - docs here . Spring Boot 2. 0 auto 配置为你注册一个。

一旦你有了 AdminClient例如，使用 createTopics() 之一方法。查看 AdminClient javadocs 获取更多信息。请务必等待 CreateTopicsResult的 Future<?> s 完成。