Читаю вот эту статью:
https://www.safaribooksonline.com/library/view/kafka-the-definitive/9781491936153/ch04.html#callout_kafka_consumers__reading_data_from_kafka_CO2-1
Тут решается проблема того, чтобы сохранять в базу результат обработки сообщений и оффсет для топика атомарно. Для этой цели решено хранить оффсет в базе и записывать в одной транзакции результат обработки сообщений и оффсет для топика.
В общем-то идея понятна, но код как-то не очень:
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
public class SaveOffsetsOnRebalance implements
ConsumerRebalanceListener {
public void onPartitionsRevoked(Collection<TopicPartition>
partitions) {
commitDBTransaction(); 1
}
public void onPartitionsAssigned(Collection<TopicPartition>
partitions) {
for(TopicPartition partition: partitions)
consumer.seek(partition, getOffsetFromDB(partition)); 2
}
}
}
consumer.subscribe(topics, new SaveOffsetOnRebalance(consumer));
consumer.poll(0);
for (TopicPartition partition: consumer.assignment())
consumer.seek(partition, getOffsetFromDB(partition)); 3
while (true) {
ConsumerRecords<String, String> records =
consumer.poll(100);
for (ConsumerRecord<String, String> record : records)
{
processRecord(record);
storeRecordInDB(record);
storeOffsetInDB(record.topic(), record.partition(),
record.offset()); 4
}
commitDBTransaction();
}
1
We use an imaginary method here to commit the transaction in the database. The idea here is that the database records and offsets will be inserted to the database as we process the records, and we just need to commit the transactions when we are about to lose the partition to make sure this information is persisted.
2
We also have an imaginary method to fetch the offsets from the database, and then we seek() to those records when we get ownership of new partitions.
3
When the consumer first starts, after we subscribe to topics, we call poll() once to make sure we join a consumer group and get assigned partitions, and then we immediately seek() to the correct offset in the partitions we are assigned to. Keep in mind that seek() only updates the position we are consuming from, so the next poll() will fetch the right messages. If there was an error in seek() (e.g., the offset does not exist), the exception will be thrown by poll().
4
Another imaginary method: this time we update a table storing the offsets in our database. Here we assume that updating records is fast, so we do an update on every record, but commits are slow, so we only commit at the end of the batch. However, this can be optimized in different ways.
SaveOffsetsOnRebalance
я так понял
onPartitionsRevoked вызывается когда один из партишенов этого листенера уходит к другому консумеру при ребалансе, ну или вообще его удалили
onPartitionsAssigned - вызывается когда этому консумеру назначили партишен
Ок commitDBTransaction() - коммитит в базу все результаты которые мы обработали а также оффсет.
consumer.seek выставляет консумеру оффсет для партишена.
Я правильно понял как это работает?
