未订阅主题的 Kafka Streams 警告答案

【问题标题】：Kafka Streams Warning for unsubscribed topics未订阅主题的 Kafka Streams 警告
【发布时间】：2020-01-21 05:07:32
【问题描述】：

在我的 kafka 流应用程序中，我订阅了一个主题列表，比如说（Topic1、Topic2、Topic3）。总体而言，代理有超过 3 个主题（100 个），其他流应用程序正在使用这些主题。

在我的应用程序（正在使用 Topic1、Topic2 和 Topic3）中，我不断看到与其他主题相关的警告。

14:02:33.910 478517 [Application-468d913c-bca0-4ad9-baec-5b51ffa597d2-StreamThread-4] WARN  o.a.kafka.clients.NetworkClient - [Consumer clientId=Application-468d913c-bca0-4ad9-baec-5b51ffa597d2-StreamThread-4-consumer, groupId=Application] 1389 partitions have leader brokers without a matching listener, including [Topic4, Topic5, Topic6, Topic7, ..., Topic8-Link-Store-changelog-5 ]

消息通过以下kafka代码打印 https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java

@Override
        public void handleSuccessfulResponse(RequestHeader requestHeader, long now, MetadataResponse response) {
            // If any partition has leader with missing listeners, log up to ten of these partitions
            // for diagnosing broker configuration issues.
            // This could be a transient issue if listeners were added dynamically to brokers.
            List<TopicPartition> missingListenerPartitions = response.topicMetadata().stream().flatMap(topicMetadata ->
                topicMetadata.partitionMetadata().stream()
                    .filter(partitionMetadata -> partitionMetadata.error() == Errors.LISTENER_NOT_FOUND)
                    .map(partitionMetadata -> new TopicPartition(topicMetadata.topic(), partitionMetadata.partition())))
                .collect(Collectors.toList());
            if (!missingListenerPartitions.isEmpty()) {
                int count = missingListenerPartitions.size();
                log.warn("{} partitions have leader brokers without a matching listener, including {}",
                        count, missingListenerPartitions.subList(0, Math.min(10, count)));
            }

这是预期的吗？为什么这些警告消息会发布在我的应用程序中？

【问题讨论】：

标签： apache-kafka kafka-consumer-api apache-kafka-streams

【解决方案1】：

注意

如果将侦听器动态添加到代理，这可能是一个暂时性问题

无论如何，这表明代理上定义的listeners 属性存在问题。

例如，您连接到PLAINTEXT://kafka:9092，但服务器实际上以PLAINTEXT_SASL:// 开头

【讨论】：