Werte von ktable emittieren und ändern

Question

Ich versuche, das folgende Problem mit Kafka zu lösen. Es gibt ein Thema. nennen wir es src-topic. Ich erhalte von Zeit zu Zeit Aufzeichnungen von diesem Thema. Ich möchte diese Werte in einer ktable speichern und die Werte, die in der ktable gespeichert sind, alle 10 Sekunden an dst-topic ausgeben. Wenn ich zum ersten Mal einen Wert von dieser ktable aussende, dann möchte ich 1 an den von mir emittierten Datensatz anhängen. Jedes weitere Mal möchte ich 0 an den ausgegebenen Datensatz anhängen.

Ich suche eine korrekte und vorzugsweise idiomatische Lösung für dieses Problem. Eine der Lösungen, die ich sehe, ist, einen Datensatz mit 1 zu senden, wenn ich aus src-Thema aufnehmen und dann in der ktable den Datensatz mit 0 angehängt speichern. Ein anderer Thread wird von diesem ktable lesen und die Datensätze regelmäßig ausgeben. Das Problem bei diesem Ansatz ist, dass es eine Wettlaufsituation hat.

Jeder Rat wird geschätzt.

Answer 1

Es gibt keine direkte Möglichkeit, dies zu tun. Hinweis: Ein KTable ist ein Changelog Stream (es kann intern einen Tabellenstatus haben - nicht alle KTables haben einen Zustand -, aber das ist ein Implementierungsdetail).

Also ein KTable ist ein Stream und Sie können einen Stream nicht spülen ... Und weil der Zustand (falls vorhanden) intern ist, können Sie den Zustand auch nicht löschen.

Der Zugriff auf den Status kann nur über interaktive Abfragen erfolgen, die auch einen Bereichsscan ermöglichen. Dies wird jedoch nichts Downstream ausgeben, sondern gibt die Daten an den "Nicht-Streams-Teil" Ihrer Anwendung.

Ich denke, Sie müssen Low-Level-Prozessor-API verwenden, um das gewünschte Ergebnis zu erhalten.