java—kakaconsumer.poll(timeoutms)和kafkaconsumer.poll(duration.zero)之间的差异

bt1cpqcv 于 2021-06-04 发布在 Kafka

关注(0)|答案(1)|浏览(556)

我从一开始就在读Kafka的主题。为了使用 seekToBeginning() 我首先需要对poll（）进行一个伪调用。以下是我的代码片段：

// Subscribe
    consumer.subscribe(Collections.singleton(TOPIC_NAME));
    // Seek to beginning
    // consumer.poll(Duration.ZERO);
    consumer.poll(0);
    consumer.seekToBeginning(Collections.singleton(new TopicPartition(TOPIC_NAME, 0)));

使用 consumer.poll(0) 很好用。当我使用 consumer.poll(Duration.ZERO) 导致以下异常：

[main] INFO org.apache.kafka.clients.consumer.internals.SubscriptionState - [Consumer clientId=consumer-group2-1, groupId=group2] Seeking to EARLIEST offset of partition test-lc-1-0
Exception in thread "main" java.lang.IllegalStateException: No current assignment for partition test-lc-1-0
    at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignedState(SubscriptionState.java:368)
    at org.apache.kafka.clients.consumer.internals.SubscriptionState.lambda$requestOffsetReset$3(SubscriptionState.java:615)
    at java.base/java.util.Collections$SingletonSet.forEach(Collections.java:4797)
    at org.apache.kafka.clients.consumer.internals.SubscriptionState.requestOffsetReset(SubscriptionState.java:613)
    at org.apache.kafka.clients.consumer.KafkaConsumer.seekToBeginning(KafkaConsumer.java:1659)
    at com.ahmed.ConsumeProtobuf.main(ConsumeProtobuf.java:49)

我看了两个api的实现。最后，两个api都使用 0 作为论据。知道为什么poll（duration.zero）会失败吗？
谢谢你，艾哈迈德。

Java apache-kafka kafka-consumer-api

来源：https://stackoverflow.com/questions/63530978/difference-between-kakaconsumer-polltimeoutms-and-kafkaconsumer-pollduration

1条答案

按热度按时间

laik7k3q1#

启动消费者时，正确的方法是使用 ConsumerRebalanceListener .
例如，类似于：

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(configs);) {
    consumer.subscribe(Collections.singleton(TOPIC_NAME), new ConsumerRebalanceListener() {
        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {}

        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            consumer.seekToBeginning(partitions);
        }
    });
    while (true) {
        consumer.poll(Duration.ofSeconds(1L));
        ...
    }
}

赞(0）回复(0）举报 2021-06-04

我来回答

java—kakaconsumer.poll(timeoutms)和kafkaconsumer.poll(duration.zero)之间的差异

1条答案

相关问题

热门标签

最新问答