Kafka-Streams verwenden Anwendungsfälle zum Hinzufügen eines globalen Speichers

Beim Definieren einer Topologie in Kafka-Streams kann ein globaler Statusspeicher hinzugefügt werden. Es wird ein Quellthema sowie ein ProcessorSupplier. Der Prozessor empfängt Datensätze und kann sie theoretisch transformieren, bevor er sie dem Speicher hinzufügt. Im Falle einer Wiederherstellung werden die Datensätze jedoch direkt aus dem Quellthema (Änderungsprotokoll) in den globalen Statusspeicher eingefügt, wobei eine eventuelle Transformation im Prozessor übersprungen wird.

   +-------------+             +-------------+              +---------------+
   |             |             |             |              |    global     |
   |source topic  ------------->  processor  +-------------->    state      |
   |(changelog)  |             |             |              |    store      |
   +-------------+             +-------------+              +---------------+
          |                                                         ^
          |                                                         |
          +---------------------------------------------------------+
              record directly inserted during restoration

StreamsBuilder # addGlobalStore (StoreBuilder storeBuilder, Zeichenfolgenthema, Verbrauchter Verbrauch, ProcessorSupplier stateUpdateSupplier) Fügt der Topologie einen globalen StateStore hinzu.

Gemäß Dokumentation

HINWEIS: Sie sollten den Prozessor nicht zum Einfügen transformierter Datensätze in den globalen Statusspeicher verwenden . Dieser Speicher verwendet das Quellthema als Änderungsprotokoll und fügt während der Wiederherstellung Datensätze direkt aus der Quelle ein . Dieser ProcessorNode sollte verwendet werden, um den StateStore auf dem neuesten Stand zu halten.

Parallel dazu ist derzeit ein schwerwiegender Fehler im kafka-Bug-Tracker geöffnet: Der auf addGlobalStore bereitgestellte benutzerdefinierte Prozessor KAFKA-7663 wird nicht verwendet, wenn der Status aus einem Thema wiederhergestellt wird, das genau erklärt, was in der Dokumentation angegeben ist, aber ein akzeptierter Fehler zu sein scheint.

Ich frage mich, ob KAFKA-7663 tatsächlich ein Fehler ist oder nicht. Der Dokumentation zufolge scheint es so konzipiert worden zu sein. In diesem Fall habe ich Schwierigkeiten, den Anwendungsfall zu verstehen.
Kann jemand die wichtigsten Anwendungsfälle dieser Low-Level-API erklären? Ich kann mir nur vorstellen, Nebenwirkungen zu verarbeiten, z. B. einige Protokollierungsvorgänge im Prozessor.

Bonusfrage: Wenn das Quellthema als Änderungsprotokoll des globalen Speichers fungiert und ein Datensatz aus dem Thema gelöscht wird, weil die Aufbewahrung abgelaufen ist, wird er dann aus dem globalen Statusspeicher entfernt? Oder erfolgt die Entfernung erst im Geschäft nach einer vollständigen Wiederherstellung des Geschäfts aus dem Änderungsprotokoll.

Ja, das ist ein ziemlich seltsamer kleiner Haken, aber die Dokumentation ist korrekt. Der Prozessor für einen globalen Statusspeicher darf nichts mit den Datensätzen tun, sondern sie im Speicher beibehalten.

AFAIK, dies ist kein philosophisches, sondern nur ein praktisches Thema. Der Grund ist einfach das beobachtete Verhalten ... Streams behandelt das Eingabethema als Änderungsprotokollthema für den Speicher und umgeht daher den Prozessor (sowie die Deserialisierung) während der Wiederherstellung.

Der Grund dafür, dass die Statuswiederherstellung jede Verarbeitung umgeht, besteht darin, dass die Daten in einem Änderungsprotokoll normalerweise mit den Daten im Speicher identisch sind. Es wäre also falsch, etwas Neues daran zu tun. Außerdem ist es effizienter, nur die Bytes vom Draht zu nehmen und sie in großen Mengen in die staatlichen Speicher zu schreiben. Ich sage "normalerweise", weil in diesem Fall das Eingabethema nicht genau wie ein normales Änderungsprotokollthema ist, da es seine Schreibvorgänge während des Speicherns nicht erhält.

Für das, was es wert ist, habe ich auch Schwierigkeiten, den Anwendungsfall zu verstehen. Scheinbar sollten wir entweder:

1. Befreien Sie sich von diesem Prozessor und werfen Sie die Binärdaten immer von der Leitung in die Speicher, genau wie bei der Wiederherstellung.
2. Entwerfen Sie globale Speicher neu, um beliebige Transformationen vor dem globalen Speicher zu ermöglichen. Wir könnten entweder:
   • Verwenden Sie weiterhin das Eingabethema und deserialisieren und rufen Sie die Prozessoren auch während der Wiederherstellung auf, ODER
   • Fügen Sie ein echtes Änderungsprotokoll für globale Speicher hinzu, sodass wir das Eingabethema abfragen, einige Transformationen anwenden und dann in den globalen Speicher und das globale Speicheränderungsprotokoll schreiben . Dann können wir das Änderungsprotokoll (nicht die Eingabe) für die Wiederherstellung und Replikation verwenden.

Übrigens, wenn Sie das letztere Verhalten möchten, können Sie es jetzt approximieren, indem Sie Ihre Transformationen anwenden und dann to(my-global-changelog)ein "Changelog" -Thema erstellen. Dann würden Sie den globalen Speicher erstellen, um my-global-changeloganstelle der Eingabe von Ihrem zu lesen .

Um Ihnen eine direkte Antwort zu geben, ist KAFKA-7663 kein Fehler. Ich werde das Ticket kommentieren und vorschlagen, es in eine Feature-Anfrage umzuwandeln.

Bonusantwort: Themen, die als Änderungsprotokolle für staatliche Geschäfte dienen, dürfen nicht mit Aufbewahrung konfiguriert werden. In der Praxis bedeutet dies, dass Sie unendliches Wachstum verhindern sollten, indem Sie die Komprimierung aktivieren und die Protokollaufbewahrung deaktivieren.

In der Praxis sind alte Daten, die nicht mehr gespeichert und gelöscht werden, kein "Ereignis", und die Verbraucher können nicht wissen, ob / wann dies geschieht. Daher ist es nicht möglich, Daten aus den Statusspeichern als Reaktion auf dieses Nichtereignis zu entfernen. Es würde passieren, wie Sie beschreiben ... die Aufzeichnungen würden einfach auf unbestimmte Zeit im globalen Laden liegen. Wenn / wenn eine Instanz ersetzt wird, wird die neue von der Eingabe wiederhergestellt und (offensichtlich) erhält nur Datensätze, die zu diesem Zeitpunkt im Thema vorhanden sind. Somit würde der Streams-Cluster als Ganzes eine inkonsistente Sicht auf den globalen Zustand erhalten. Aus diesem Grund sollten Sie die Aufbewahrung deaktivieren.

Der richtige Weg, um alte Daten aus dem Geschäft zu "löschen", besteht darin, einfach einen Grabstein für den gewünschten Schlüssel in das Eingabethema zu schreiben. Dies würde dann korrekt an alle Mitglieder des Clusters weitergegeben, während der Wiederherstellung korrekt angewendet UND von den Brokern korrekt komprimiert.

Ich hoffe das alles hilft. Bitte melden Sie sich auf jeden Fall für das Ticket an und helfen Sie uns, die API intuitiver zu gestalten!