Gibt es eine Standardmethode zum Zurücksetzen von Offsets bei Verwendung der Annotation @KafkaListener? - Apache-Kafka, Frühlings-Kafka

Ich benutze das KafkaListener Interface in meiner Spring Boot App, was gut funktioniert. Offsets werden von Kafka selbst gespeichert.

Nehmen wir einmal an, einer der Konsumenten eines Themas stellt eine neue Version bereit und verschraubt 2 Stunden Nachrichten. Dann reparieren sie die App und wollen die neue Version mit einem Offset von vor zwei Stunden starten.

Ich könnte die consumer.seek () mit einem früheren consumer.offsetsForTimes () Aufruf verwenden, aber das ist nur geradlinig, wenn ich den Abfragemechanismus benutze, nicht wenn der Spring KafkaListener verwendet wird.

Ich habe versucht, einen ConsumerRebalanceListener zu verwenden, aber um die Offsets zu setzen, muss der Consumer aktiv sein, was ich nicht habe, wenn ich den ConsumerRebalanceListener instanziiere ...

Muss ich zum Polling wechseln? Kann ich den Kunden erreichen, bevor mein KafkaListener das erste Mal aufgerufen wird, aber nachdem dem Kunden eine oder mehrere Partitionen zugewiesen wurden?

Antworten:

Nicht gleichzeitig; das bevorstehende 2.0 Veröffentlichung führt ein ConsumerAwareMessageListener, die unterstützt, die Consumer als ein Argument in die @KafkaListener Methode.