为什么不从头开始一个新的消费者组id

Question

我有一个 kafka 0.10 集群，其中包含几个向它们生成消息的主题。

当我使用 KafkaConsumer 和新组 ID 订阅主题时，我不会返回任何记录，但是如果我使用以相同组 ID 寻求开头的 ConsumerRebalanceListener 订阅主题，那么我会获得主题中的记录。

@Grab('org.apache.kafka:kafka-clients:0.10.0.0')

import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.clients.consumer.ConsumerRecords
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.PartitionInfo

Properties props = new Properties()
props.with {
    put("bootstrap.servers","***********:9091")
    put("group.id","script-test-noseek")
    put("enable.auto.commit","true")
    put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer")
    put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer")
    put("session.timeout.ms",30000)
}

KafkaConsumer consumer = new KafkaConsumer(props)
def topicMap = [:]
consumer.listTopics().each { topic, partitioninfo ->
        topicMap[topic] = 0
}

topicMap.each {topic, count ->
    def stopTime = new Date().time + 30_000
    def stop = false

    println "Starting topic: $topic"
    consumer.subscribe([topic])
    //consumer.subscribe([topic], new CRListener(consumer:consumer))
    while(!stop) {
        ConsumerRecords<String, String> records = consumer.poll(5_000)
        topicMap[topic] += records.size()
        consumer.commitAsync()
        if ( new Date().time > stopTime || records.size() == 0) {
            stop = true
        }
    }    
    consumer.unsubscribe()
}

def total = 0
println "------------------- Results -----------------------"
topicMap.each { k,v ->
    if ( v > 0 ) {
        println "Topic: ${k.padRight(64,' ')} Records: ${v}"
    }
    total += v
} 
println "==================================================="
println "Total: ${total}"
def dummy = "Process End"

class CRListener implements ConsumerRebalanceListener {
    KafkaConsumer consumer
    void onPartitionsAssigned(java.util.Collection partitions) {
        consumer.seekToBeginning(partitions)
    }
    void onPartitionsRevoked(java.util.Collection partitions) {
        consumer.commitSync()
    }
}

代码是 Groovy 2.4.x。我屏蔽了引导服务器。如果我取消注释消费者订阅行与侦听器，它会做我期望它做的事情。但事实上，我没有得到任何结果。

假设我为每次运行更改了组 ID，只是为了不在另一个执行停止的地方进行。

我看不出我做错了什么。任何帮助，将不胜感激。

Answer 1

如果您使用新的消费者组 ID 并希望从头开始阅读整个主题，则需要在属性中指定参数“auto.offset.reset=earliest”。（默认值为“最新”）

Properties props = new Properties()
props.with {
    // all other values...
    put("auto.offset.reset","earliest")
}

在消费者启动时会发生以下情况：

1. 寻找（有效）已提交的偏移量以供使用 group.id
2. 如果找到（有效）偏移量，则从那里恢复
3. 如果未找到（有效）偏移量，则根据 auto.offset.reset