如何检查Kafka消费者是否准备好了

Question

我将 Kafka 提交策略设置为最新且缺少前几条消息。如果我在开始将消息发送到输入主题之前先睡20秒，那么一切都按预期工作。我不确定问题是否与消费者需要很长时间进行分区重新平衡有关。有没有办法在开始轮询之前知道消费者是否准备好了？

Answer 1

多亏了阿列克谢（我也投了赞成票），我似乎基本上按照同样的想法解决了我的问题。

只是想分享我的经验……在我们的案例中，我们根据要求使用Kafka。

我尝试了…KafkaConsumer。分配（）重复（使用Thread.sleep（100）），但似乎没有帮助。添加KafkaConsumer。投票（50）似乎已经为消费者（群体）做好了准备，并且也收到了第一个响应。测试了几次，现在它一直在工作。

顺便说一句，测试需要停止应用程序

PS：只需调用poll（50）不带赋值（） 获取逻辑可能无法保证消费者（组）已经准备好了。

Answer 2

您可以执行以下操作：

我有一个从kafka主题读取数据的测试
所以您不能在多线程环境中使用KafkaConsumer，但您可以传递参数“AtomicReference assignment”，在使用者线程中更新它，然后在另一个线程中读取它例如，项目中的工作代码被截取以进行测试：

    private void readAvro(String readFromKafka,
                      AtomicBoolean needStop,
                      List<Event> events,
                      String bootstrapServers,
                      int readTimeout) {
    // print the topic name
    AtomicReference<Set<TopicPartition>> assignment = new AtomicReference<>();
    new Thread(() -> readAvro(bootstrapServers, readFromKafka, needStop, events, readTimeout, assignment)).start();

    long startTime = System.currentTimeMillis();
    long maxWaitingTime = 30_000;
    for (long time = System.currentTimeMillis(); System.currentTimeMillis() - time < maxWaitingTime;) {
        Set<TopicPartition> assignments = Optional.ofNullable(assignment.get()).orElse(new HashSet<>());
        System.out.println("[!kafka-consumer!] Assignments [" + assignments.size() + "]: "
                + assignments.stream().map(v -> String.valueOf(v.partition())).collect(Collectors.joining(",")));
        if (assignments.size() > 0) {
            break;
        }
        try {
            Thread.sleep(1_000);
        } catch (InterruptedException e) {
            e.printStackTrace();
            needStop.set(true);
            break;
        }
    }
    System.out.println("Subscribed! Wait summary: " + (System.currentTimeMillis() - startTime));
}

private void readAvro(String bootstrapServers,
                      String readFromKafka,
                      AtomicBoolean needStop,
                      List<Event> events,
                      int readTimeout,
                      AtomicReference<Set<TopicPartition>> assignment) {

    KafkaConsumer<String, byte[]> consumer = (KafkaConsumer<String, byte[]>) queueKafkaConsumer(bootstrapServers, "latest");
    System.out.println("Subscribed to topic: " + readFromKafka);
    consumer.subscribe(Collections.singletonList(readFromKafka));

    long started = System.currentTimeMillis();
    while (!needStop.get()) {
        assignment.set(consumer.assignment());
        ConsumerRecords<String, byte[]> records = consumer.poll(1_000);
        events.addAll(CommonUtils4Tst.readEvents(records));

        if (readTimeout == -1) {
            if (events.size() > 0) {
                break;
            }
        } else if (System.currentTimeMillis() - started > readTimeout) {
            break;
        }
    }

    needStop.set(true);

    synchronized (MainTest.class) {
        MainTest.class.notifyAll();
    }
    consumer.close();
}

P. S.
需要停止-全局标志，停止所有正在运行的线程，如果有成功的失败
事件-对象列表，我想检查
readTimeout-我们将等待多少时间，直到读取所有数据，如果readTimeout==-1，然后停止，当我们读取任何东西

Answer 3

>

您可以使用消费者。assignment（），它将返回一组分区，并验证是否分配了该主题可用的所有分区。

如果您使用的是spring-kafka项目，您可以包括spring-kafka测试依赖性，并使用下面的方法等待主题分配，但您需要有容器<代码>ContainerTestUtils。waitForAssignment（对象容器，int分区）