Wie man den Verbraucher auf einen bestimmten Offset in Golang einstellen kann Kafka 10

Question

Ich muss den Producer veranlassen, von der letzten Nachricht zu starten, die er bearbeitet hat, bevor er abgestürzt ist. Glücklicherweise habe ich nur ein Thema mit einer Partition und einem Consumer.

Dazu habe ich versucht https://github.com/Shopify/sarama, aber es scheint noch nicht verfügbar zu sein. Ich verwende jetzt https://godoc.org/github.com/bsm/sarama-cluster, die mir erlauben, jeden Nachrichtenoffset zu begehen.

Ich kann den letzten festgeschriebenen Offset nicht abrufen Ich kann nicht herausfinden, wie man einen Sarama consumer von diesem Offset zu starten. Der einzige Parameter, den ich bisher gefunden habe, ist Config.Producer.Offsets.Initial.

1. Wie wird der letzte festgeschriebene Offset abgerufen?
2. Wie kann der Verbraucher von der letzten Nachricht starten, deren Offset festgeschrieben wurde? OffsetNewest wird es von der letzten produzierten Nachricht starten, nicht die letzte verarbeitet b Verbraucher.
3. Ist es möglich, nur mit Shopify/sarama und nicht mit bsm/sarama-cluster zu arbeiten?

Vielen Dank im Voraus

P. S. Ich benutze Kafka 10.0, also sind die Offsets Speicher in einem Kafka und nicht im Tierpfleger.

EDIT1: Teillösung: holen alle Nachrichten seit sarama.OffsetOldest und alle von ihnen überspringen, bis wir eine nicht verarbeitete eine gefunden.

Answer 1

Wenn der Offset für eine Partition bereits gespeichert wurde, setzt der sarama-cluster den Verbrauch von diesem Offset wieder her. Die Option Config.Producer.Offsets.Initial wird nur verwendet, wenn kein gespeicherter Offset vorhanden ist (zum ersten Mal für eine Verbrauchergruppe).

Sie können dies überprüfen, indem Sie die folgende Zeile am Anfang Ihrer main() Funktion hinzugefügt:

sarama.Logger = log.New(os.Stdout, "sarama: ", log.LstdFlags) 

Dann werden Sie so etwas wie die folgenden in der Ausgabe sehen:

cluster/consumer CID-17db1be4-a162-411c-a106-4d198191176a consume sample/0 from 12

Die 12 ist der Offset, von dem Sarama für diese Partition ausgeht (Beispiel/0).