Kafka sprint 插件未分发所有分区答案

【问题标题】：Kafka sprint plugin is not distributing all partitionsKafka sprint 插件未分发所有分区
【发布时间】：2018-08-03 13:22:13
【问题描述】：

假设 Kafka 主题 (my-topic) 有 8 个分区，并且我有一个侦听器组 (my-topic-group)，它包含 8 到 10 个不同进程在不同的机器上运行。特定分区 (my-topic-2) 没有被任何侦听器使用。

这是 Kafa Producer (Jar-1) 的代码

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

KafkaProducer producer ;
.....


producer.send(new ProducerRecord('my-topic', student_id % 8, null, payload));

这里是 Spring (Jar-2) 中 Kafa 监听器配置的代码

@EnableKafka
@Configuration
public class SpringBootKafka {

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();

        String servers ;
        .....

        props.put("bootstrap.servers", servers);
        props.put("group.id", "my-topic-group");
        props.put("key.deserializer", StringDeserializer.class);
        props.put("value.deserializer", StringDeserializer.class);
        props.put("auto.offset.reset", "earliest");

        return props;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

Kafka 监听器（Jar-2）

@Component
public class EventsReceiver {
    @KafkaListener(topics = "my-topic")
    public void receive(ConsumerRecord<String, String> consumerRecord) {

        String message = consumerRecord.value();

    }
}

最初 Jar-2 部署在 1 台机器（docker）中，慢慢地我们将 pod 的数量增加到 10 个。没有一个 pod 监听 my-topic-partition-2。而且不止一个正在收听my-topic-partition-7。所以我在监听器中遗漏了一些 kafka 事件。

【问题讨论】：

你为什么要自己使用student_id % 8？默认分区器已经对总主题进行了取模
不知道为什么，但这不是我的代码。我只是从 kakfa 消费。
这是您在问题中粘贴的代码的一部分...您是什么意思？
无论如何，你怎么知道消息甚至在分区 2 中？如果你不消费，那么你需要检查数据是否真的存在
Jar-1 不是我的代码，jar-2 是我的代码。在 spring-boot 重新加载后，我没有看到任何连接到 partition-2 的进程（来自记录器）。我在 Perl 中编写了另一个代码来读取所有分区并打印数据和分区信息，其中包含所有数据（包括分区 2）。我假设 spring 配置有问题。

标签： java spring spring-boot apache-kafka spring-kafka

【解决方案1】：

查看日志；当实例出现时，您应该会看到类似的内容：

Instance 1

partitions assigned: [so51673658-2, so51673658-1, so51673658-4, so51673658-3, so51673658-6, so51673658-5, so51673658-8, so51673658-7, so51673658-9, so51673658-0]
partitions revoked: [so51673658-2, so51673658-1, so51673658-4, so51673658-3, so51673658-6, so51673658-5, so51673658-8, so51673658-7, so51673658-9, so51673658-0]
partitions assigned: [so51673658-2, so51673658-1, so51673658-4, so51673658-3, so51673658-0]
partitions revoked: [so51673658-2, so51673658-1, so51673658-4, so51673658-3, so51673658-0]
partitions assigned: [so51673658-4, so51673658-6, so51673658-5]

Instance 2

partitions assigned: [so51673658-6, so51673658-5, so51673658-8, so51673658-7, so51673658-9]
partitions revoked: [so51673658-6, so51673658-5, so51673658-8, so51673658-7, so51673658-9]
partitions assigned: [so51673658-8, so51673658-7, so51673658-9]

Instance 3

partitions assigned: [so51673658-2, so51673658-1, so51673658-3, so51673658-0]

【讨论】：

【解决方案2】：

我发现了这个问题，它在 2 个不同的数据中心运行，并且都指向同一个 Kafka 集群。分区 2 被分配到第二个数据中心。

【讨论】：