1

我正在尝试增加消费者的数量以匹配我们正在读取的 Kafka 主题的分区数量。有三个分区，所以我将传入消息的分区配置为三个，如下所示：

mp:
  messaging:
    incoming:
      topicA:
        auto:
          offset:
            reset: earliest
        topics: TOPIC-NAME
        connector: smallrye-kafka
        value:
          deserializer: org.apache.kafka.common.serialization.StringDeserializer
        group:
          id: consumer-group
        partitions: 3

但是，我已经运行该应用程序一段时间了，似乎该应用程序只处理分区 0 中的消息，而不处理分区 1 和 2 中的消息。我在日志中看到它创建了三个消费者。

2021-06-16 23:35:59,826 INFO  [org.apa.kaf.cli.con.int.AbstractCoordinator] (vert.x-kafka-consumer-thread-0) [Consumer clientId=kafka-consumer-topicA-0, groupId=consumer-group] Successfully joined group with generation 15
2021-06-16 23:35:59,826 INFO  [org.apa.kaf.cli.con.int.AbstractCoordinator] (vert.x-kafka-consumer-thread-2) [Consumer clientId=kafka-consumer-topicA-2, groupId=consumer-group] Successfully joined group with generation 15
2021-06-16 23:35:59,826 INFO  [org.apa.kaf.cli.con.int.AbstractCoordinator] (vert.x-kafka-consumer-thread-1) [Consumer clientId=kafka-consumer-topicA-1, groupId=consumer-group] Successfully joined group with generation 15
2021-06-16 23:35:59,831 INFO  [org.apa.kaf.cli.con.int.ConsumerCoordinator] (vert.x-kafka-consumer-thread-1) [Consumer clientId=kafka-consumer-topicA-1, groupId=consumer-group] Adding newly assigned partitions: TOPIC-NAME-1
2021-06-16 23:35:59,831 INFO  [org.apa.kaf.cli.con.int.ConsumerCoordinator] (vert.x-kafka-consumer-thread-0) [Consumer clientId=kafka-consumer-topicA-0, groupId=consumer-group] Adding newly assigned partitions: TOPIC-NAME-0
2021-06-16 23:35:59,831 INFO  [org.apa.kaf.cli.con.int.ConsumerCoordinator] (vert.x-kafka-consumer-thread-2) [Consumer clientId=kafka-consumer-topicA-2, groupId=consumer-group] Adding newly assigned partitions: TOPIC-NAME-2

但它似乎在处理分区 0 中的消息：

2021-06-16 23:38:00,141 INFO  [MessageListener] (vert.x-worker-thread-2) Partition number:0; offset: 1593011
2021-06-16 23:38:00,282 INFO  [MessageListener] (vert.x-worker-thread-1) Partition number:0; offset: 1593012
2021-06-16 23:38:00,412 INFO  [MessageListener] (vert.x-worker-thread-4) Partition number:0; offset: 1593013
2021-06-16 23:38:00,543 INFO  [MessageListener] (vert.x-worker-thread-6) Partition number:0; offset: 1593014
2021-06-16 23:38:00,692 INFO  [MessageListener] (vert.x-worker-thread-8) Partition number:0; offset: 1593015
2021-06-16 23:38:00,838 INFO  [MessageListener] (vert.x-worker-thread-10) Partition number:0; offset: 1593016
2021-06-16 23:38:00,977 INFO  [MessageListener] (vert.x-worker-thread-12) Partition number:0; offset: 1593017
2021-06-16 23:38:01,131 INFO  [MessageListener] (vert.x-worker-thread-14) Partition number:0; offset: 1593018
2021-06-16 23:38:01,272 INFO  [MessageListener] (vert.x-worker-thread-16) Partition number:0; offset: 1593019
2021-06-16 23:38:01,406 INFO  [MessageListener] (vert.x-worker-thread-18) Partition number:0; offset: 1593020
2021-06-16 23:38:01,535 INFO  [MessageListener] (vert.x-worker-thread-0) Partition number:0; offset: 1593021
2021-06-16 23:38:01,670 INFO  [MessageListener] (vert.x-worker-thread-3) Partition number:0; offset: 1593022
2021-06-16 23:38:01,799 INFO  [MessageListener] (vert.x-worker-thread-5) Partition number:0; offset: 1593023

下面是监听器类的代码片段：

    @Incoming("topicA")
    @Blocking
    public CompletionStage<Void> consume(final IncomingKafkaRecord<String, String> message) {

        log.info("Partition number:" + message.getPartition() + "; offset: " + message.getOffset());
    
        return message.ack();
    }

这是小黑麦卡夫卡的错误吗？

4

1 回答 1

0

看来这些设置将使用多个分区：这将消耗所有消息。

mp.messaging.incoming.your-events.auto.offset.reset=earliest
mp.messaging.incoming.your-events.group.id=${quarkus.uuid}

如果您使用的是发射器，则无需上述设置即可使用；

int partition = 0;
Message<Integer> message = Message.of(value)
            .addMetadata(OutgoingKafkaRecordMetadata.<String>builder()
                .withKey(key)
                .withPartition(partition) // change for each partition, 0, 1, 2..                
                .withTopic("your-events")
                .build());
于 2021-12-22T22:31:09.807 回答