如何从分区分配器的实现中调用kafkaconsumer api

Question

我已经通过在我的spring boot应用程序中实现RangeAssignor实现了我自己的分区分配策略。我已经覆盖了它的subscriptionUserData方法并添加了一些用户数据。每当数据发生变化时，我都希望通过调用下面的kafkaconsumer apis来触发分区重新平衡kafkaconsumer apis强制重新平衡

我不确定如何获取 kafka 使用者的对象并调用此 api。请建议

Answer 1

您可以调用consumer.wakeup（）函数

consumer.wakeup（）是唯一可以安全地从不同html" target="_blank">线程调用的消费者方法。调用wakeup将导致轮询（）与WakeupException一起退出，或者如果线程没有等待轮询时调用了consumer.wakeup（），则在调用轮询（）时将在下一次迭代中引发异常。不需要处理WakeupException，但在退出线程之前，您必须调用consumer.close（）。如果需要，关闭消费者将提交偏移集，并将向组协调器发送消费者正在离开组的消息。消费者协调器将立即触发重新平衡

  Runtime.getRuntime().addShutdownHook(new Thread() {
        public void run() {
            System.out.println("Starting exit...");
            consumer.wakeup();   **//1**
            try {
                mainThread.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
} });
    ...
    Duration timeout = Duration.ofMillis(100);
    try {
        // looping until ctrl-c, the shutdown hook will cleanup on exit
        while (true) {
            ConsumerRecords<String, String> records =
                movingAvg.consumer.poll(timeout);
            System.out.println(System.currentTimeMillis() +
                "--  waiting for data...");
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s\n",
                    record.offset(), record.key(), record.value());
            }
            for (TopicPartition tp: consumer.assignment())
                System.out.println("Committing offset at position:" +
                    consumer.position(tp));
                movingAvg.consumer.commitSync();
        }
    } catch (WakeupException e) {
        // ignore for shutdown. **//2**
    } finally {
        consumer.close(); **//3**
        System.out.println("Closed consumer and we are done");
    }

1. ShutdownHook在一个单独的线程中运行，因此我们可以采取的唯一安全操作是调用wakeup来打破轮询循环。
2. 另一个调用wakeup的线程将导致轮询抛出WakeupException。您需要捕获异常以确保您的应用程序不会意外退出，但无需对其执行任何操作。
3. 在退出消费者之前，请确保干净地关闭它。

完整示例：

https://github.com/gwenshap/kafka-examples/blob/master/SimpleMovingAvg/src/main/java/com/shapira/examples/newconsumer/simplemovingavg/SimpleMovingAvgNewConsumer.java