Einstellen und Abrufen von benutzerdefinierten Kafka-Header-Eigenschaften

Sie können benutzerdefinierte Kafka-Header-Eigenschaften verwenden, um Metadaten mit einer Kafka-Nachricht zu verknüpfen.

Vorbereitungen

Lesen Sie die folgenden Abschnitte:

Informationen zu dieser Task

Kafka mit benutzerdefinierten Header-Eigenschaften können Sie der Nachricht Kafka Metadaten hinzufügen, die dann bei der Verarbeitung der Nachricht verwendet werden können. Die Headereigenschaften können beispielsweise Informationen zum Format der Daten enthalten, wie z. B. einen Schemanamen.

Bei den benutzerdefinierten Kopfzeileneigenschaften von Kakfa handelt es sich um eine Reihe von name:value -Paaren, die bei der Veröffentlichung mit einer Kafka -Nachricht verknüpft und beim Abruf der Nachricht abgerufen werden können. IBM App Connect Enterprise unterstützt „ Kafka “-Header-Eigenschaften, die aus einem Namen im Java-String-Format und einem Wert bestehen.

Diese Eigenschaften werden in der LocalEnvironment in einem Ordner namens KafkaHeader festgelegt. Für den KafkaProducer knoten werden alle Eigenschaftswerte mit Ausnahme von BLOB-Werten in UTF-8 Strings umgewandelt, bevor die Bytes in der Header-Eigenschaft in der Kafka Nachricht gespeichert werden. Bei BLOB-Werten werden die Bytes in der Header-Eigenschaft der Nachricht Kafka gespeichert, ohne umgewandelt zu werden. Für KafkaConsumer und KafkaRead werden die Werte der Header-Eigenschaften in der empfangenen Nachricht standardmäßig in String-Werte umgewandelt, wenn sie in die lokale Umgebung geschrieben werden. Um die Werte als Bytes anstelle von Zeichenketten zu erhalten, setzen Sie die Eigenschaft Header scheme auf dem Knoten auf Bytes.

Bevor Sie den KafkaProducer -Knoten aufrufen, können Sie die angepassten Kafka -Headereigenschaften im Ordner für die lokale Environment.Destination.Kafka.Output.KafkaHeader -Umgebung festlegen. Zum Beispiel:

# To set Kafka custom header properties called 'Name' and 'Occupation'
      SET OutputLocalEnvironment.Destination.Kafka.Output.KafkaHeader.Name = 'Bob';
      SET OutputLocalEnvironment.Destination.Kafka.Output.KafkaHeader.Occupation = 'Builder';

Bei der Rückkehr von einem KafkaConsumer knoten wird der Ordner "Lokale Umgebung" LocalEnvironment.Kafka.Input mit allen benutzerdefinierten Kopfzeileneigenschaften aktualisiert, die in der empfangenen Nachricht enthalten sind, die dann mithilfe eines Rechenknotens verarbeitet werden kann. Zum Beispiel:

 # To retrieve the value of the custom header property called 'Name'
        SET Name = InputLocalEnvironment.Kafka.Input.KafkaHeader.Name;

Bei der Rückkehr von einem KafkaRead knoten wird der Ordner "Lokale Umgebung" LocalEnvironment.Kafka.Read mit allen benutzerdefinierten Kopfzeileneigenschaften aktualisiert, die in der empfangenen Nachricht enthalten sind, die dann mithilfe eines Rechenknotens verarbeitet werden kann. Zum Beispiel:

# To retrieve the value of the custom header property called 'Occupation'
        SET Occupation = InputLocalEnvironment.Kafka.Read.KafkaHeader.Occupation;

Weitere Informationen zu den Kafka-Knoten finden Sie in folgenden Abschnitten:

Informationen zu Eigenschaften, die im Flow dynamisch überschrieben werden können, enthält der Abschnitt Lokale Umgebungsvariablen mit Kafka -Knoten verwenden.