java - Kafka 从相同的偏移量重启

Question

我有一个 kafka 消费者，它连接到一个具有 3 个分区的主题。一旦我从 kafka 获得记录，我想捕获偏移量和分区。重新启动时，我想从上次读取偏移量恢复消费者的位置

来自kafka文档:

每条记录都有自己的偏移量，因此要管理您自己的偏移量，您只需执行以下操作:

Configure enable.auto.commit=false

Use the offset provided with each ConsumerRecord to save your position.

On restart restore the position of the consumer using seek (TopicPartition, long).

这是我的示例代码:

constructor{    
    load data into offsetMap<partition,offset>
    initFlag=true;
}

Main method
{
    ConsumerRecords<String, String> records = consumer.poll(100);
    if(initFlag) // is this correct way to override offset position?
    {
        seekToPositions(offsetMap); 
        initFlag=false;
    }
    while(!shutdown)
    {
        for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                getOffsetPositions();// dump offsets and partitions to db/disk
        }   
   }
}

//get current offset and write to a file
public synchronized Map<Integer, Long> getOffsetPositions() throws Exception{

    Map<Integer, Long> offsetMap = new HashMap<Integer, Long>();
    //code to put partition and offset into map
    //write to disk or db

    }
} // Overrides the fetch offsets that the consumer

public synchronized void seekToPositions(Map<Integer, Long> offsetMap) {
            //code get partitions and offset from offsetMap
            consumer.seek(partition, offset);

    }

这是正确的做法吗？有没有更好的办法？

Answer 1

如果您提交偏移量，Kafka 将为您存储它们(默认情况下最多存储 24 小时)。

这样，如果您的消费者死亡，您可以在另一台机器上启动相同的代码，然后从您中断的地方继续。无需外部存储。

参见 https://kafka.apache.org/0102/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html 中的“抵消和消费者地位”

并推荐你考虑使用commitSync