Kafka从相同偏移量重新启动

Question

我有一个单一的Kafka消费者，它连接到一个有3个分区的主题。一旦我从Kafka那里得到一张唱片，我就想捕捉偏移量和分区。在重新启动时，我希望从上次读取的偏移量恢复使用者的位置

摘自Kafka文档：

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

配置enable.auto.commit=false

下面是我的示例代码：

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