如何在使用動態組重新平衡時指定卡夫卡消費者的起始位置？

Question

是否有可能從指定的偏移量啓動0.9或0.10卡夫卡消費者，同時仍然使用具有動態重新平衡功能的消費羣體？

這裏是迄今發現：

案例1：如果我們使用consumer.assign（...）方法來手動指定分區的消費者 - 我們能做到以下操作：

consumer.seek(<specificPartition>, <myCustomOffset>); or: 
consumer.seekToBeginning(<specificPartition>); 
consumer.seekToEnd(<specificPartition>); 

基本上，我們可以完全控制啓動消費者表單的哪個位置，但是它的代價是沒有由卡夫卡動態地完成分區重新分配。情況2：如果我們使用consumer.subscribe（...）方法 - 卡夫卡將管理重新平衡，但是，我們不能做任何的以上三個選項... :( 因此，我們嘗試了以下爲「黑客」周圍 - 在消費者的啓動時間，前進入投票（）循環：

// get coordinator from the private field of the consumer: 
ConsumerCoordinator coordinator = (ConsumerCoordinator) FieldUtils.readField(consumer, "coordinator", true); 
// make sure all partitions are already 
coordinator.ensurePartitionAssignment(); 
// get the list of partitions assigned to this specific consumer: 
Set<TopicPartition> assignedTopicPartitions = consumer.assignment() 
// now we can go ahead and do the same three actions (seek(), sequined() or seekToBeginning()) on those partitions only for this consumer as above. 
for (TopicPartition assignedPartition: assignedTopicPartitions) { 
    consumer.seek(<assignedPartition>, <myCustomOffset>) // or whatever 
... 
} 
// now start the poll() loop: 
while (true) { 
    ConsumerRecords<String, String> records = consumer.poll(pollIntervalMs); 
    for (ConsumerRecord<String, String> record : records) { 
     // processMessage(record.value(), record.offset()); 
    } 
} 

這對我的品味來說太過黑客，而且，我也不確定這種邏輯是否會在實際的重新平衡過程中持續下去，比如新的消費者被加入到羣組中。

有人可以驗證這種方法或建議一個更好的方法來完成我們需要的東西嗎？

謝謝！

Answer 1

而不是使用ConsumerCoordinator你可以做一個初始的poll()（並且不處理任何東西）來分配分區。之後，使用seek()並開始您的輪詢循環，如代碼中所示。