重置卡夫卡消費者的上一次偏移量

Question

我想重置與給定使用者相對應的先前偏移量。

原因：我寫我的應用程序的spring-boot消費者代碼是使用「最早的」作爲自動偏移重置的值。由於現在偏移量已存儲在__consumer_offsets中，因此將自動偏移重置的值更改爲最新值不起作用。

注意：我使用的kafka版本高於0.9。不知道刪除使用者是否會有所幫助，因爲我知道偏移量現在存儲在主題__consumer_offsets中。

Answer 1

如果您的聽衆實施了ConsumerSeekAware，您可以尋求消費者。請參閱the documentation。

void registerSeekCallback(ConsumerSeekCallback callback); 

void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback); 

void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback); 

隨着回調提供：

void seek(String topic, int partition, long offset);