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

Question

@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()
    }
}

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

Answer 1

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

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

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

1. 查找（有效的）提交偏移量以便使用group.id
2. 如果找到（有效）偏移量，请从那里继续
3. 如果未找到（有效）偏移量，请根据auto.offset.reset
设置偏移量