Abonnieren von Kafka

Einrichten des Kafka

Sie können Ihren Kafka-Nachrichtenproduzenten so einrichten, dass Nachrichten an Kafka-Ereignissubskriptionen Code Engine gesendet werden. Verwenden Sie Ihre Code Engine-Ereignissubskription Kafka, um Anwendungen oder Jobs auszulösen, wenn eine Kafka-Nachricht empfangen wird.

Erstellen Sie zunächst eine Event Streams-Serviceinstanz für Ihren Event-Streaming-Service. Während Sie die Konsole oder die Befehlszeilenschnittstelle verwenden können, wird in den folgenden Schritten beschrieben, wie Sie den Ereigniserzeuger Event Streams mit der Befehlszeilenschnittstelle einrichten.

1. Erstellen Sie eine Serviceinstanz für Event Streams. Der Name des Event Streams-CLI-Service lautet messagehub. Erstellen Sie für dieses Beispiel eine Event Streams-Serviceinstanz namens myeventstream.

   ibmcloud resource service-instance-create myeventstream messagehub lite us-south
   

2. Erstellen Sie einen Serviceschlüssel, um Berechtigungsnachweise für Ihre Serviceinstanz bereitzustellen.

   ibmcloud resource service-key-create myeventstream-key Manager --instance-name myeventstream
   

   Beispielausgabe

   Creating service key of service instance myeventstream under account <user_account>...
   OK
   Service key crn:v1:bluemix:public:messagehub:us-south:a/e43abfcbd191404cb17ef650e9681dd3:c0736069-3f4a-438a-b614-6846877d692d:resource-key:4c8edfdb-abcd-abcd-abcd-abcdabcdabcd was created.
   
   Name:          myeventstream-key
   ID:            crn:v1:bluemix:public:messagehub:us-south:a/e43abfcbd191404cb17ef650e9681dd3:c0736069-3f4a-438a-b614-6846877d692d:resource-key:4c8edfdb-abcd-abcd-abcd-abcdabcdabcd
   Created At:    Mon Mar 21 18:36:09 UTC 2022
   State:         active
   Credentials:
               api_key:                  abcdeH9tu3qE5Sn8VbJfcDEWtjR_l0iPisB3abcdefgh
               apikey:                   abcdeH9tu3qE5Sn8VbJfcDEWtjR_l0iPisB3abcdefgh
               iam_apikey_description:   Auto-generated for key crn:v1:bluemix:public:messagehub:us-south:a/e43abfcbd191404cb17ef650e9681dd3:c0736069-3f4a-438a-b614-6846877d692d:resource-key:4c8edfdb-abcd-abcd-abcd-abcdabcdabcd
               iam_apikey_name:          myeventstream-key
               iam_role_crn:             crn:v1:bluemix:public:iam::::serviceRole:Manager
               iam_serviceid_crn:        crn:v1:bluemix:public:iam-identity::a/e43abfcbd191404cb17ef650e9681dd3::serviceid:ServiceId-3e99caa5-b174-4f04-9845-5c5d783b8bc7
               instance_id:              c0736069-3f4a-438a-b614-6846877d692d
               kafka_admin_url:          https://abcdabcdabcdabcd.svc07.us-south.eventstreams.cloud.ibm.com
               kafka_brokers_sasl:       [broker-1-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 broker-2-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093]
               kafka_http_url:           https://abcdabcdabcdabcd.svc07.us-south.eventstreams.cloud.ibm.com
               password:                 abcdeH9tu3qE5Sn8VbJfcDEWtjR_l0iPisB3abcdefgh
               user:                     token
   

   Notieren Sie sich die Werte für user, password und kafka-brokers_sasl für Ihren Serviceschlüssel. Sie benötigen diese Informationen, wenn Sie Ihre Subskription für Code Engine Kafka einrichten. Die Werte für password und apikey sind im Serviceschlüssel für Ihre Event Streams-Serviceinstanz identisch. Sie können auch den Befehl ibmcloud resource service-key myeventstream-key verwenden, um die Serviceschlüsselinformationen abzurufen.

3. Initialisieren Sie das Plug-in Event Streams relativ zu Ihrer Event Streams-Serviceinstanz.

   ibmcloud es init --instance-name myeventstream
   

4. Erstellen Sie ein Event Streams Thema.

   ibmcloud es topic-create kafka-topic1
   

Einrichten einer Code Engine-Beispielapp zum Erzeugen von Kafka-Nachrichten

Richten Sie für dieses Lernprogramm eine Code Engine-Anwendung als Ereigniserzeuger von Kafka-Nachrichten ein. Der Zweck dieser kafka-sender-app-App besteht darin, eine Verbindung zu Ihrer Event Streams-Instanz herzustellen und Kafka-Nachrichten an einen Empfänger (Kafka-Konsument) zu erzeugen (zu senden). Diese App, die Ereignisse für Kafka-Nachrichten erzeugt, verwendet die Code Engine Kafka-Absenderbeispielapp, um Kafka-Nachrichten zu senden. Dieses Beispielabsender-Image erfordert die Umgebungsvariable BROKERS und einen geheimen Schlüssel, der die password-Berechtigungsnachweise enthält.

1. Erstellen Sie einen geheimen Schlüssel mit Berechtigungsnachweisen, die für die Code Engine Kafka-Beispiele erforderlich sind. Erstellen Sie beispielsweise den geheimen Schlüssel kafka-subscription-secret, um die Berechtigungsnachweise zu enthalten, die sowohl für die Kafka-Beispielsender-App als auch für die Kafka-Ereignissubskription erforderlich sind, die das Empfängerbeispiel Kafka verwendet. Diese Berechtigungsnachweise sind für die Kafka-Beispiel-Sender-App und die Ereignissubskription Code Engine Kafka erforderlich, um mit der Serviceinstanz für Event Streamszu kommunizieren. Obwohl es nicht erforderlich ist, dass Sie diesen geheimen Schlüssel erstellen, bevor Sie die Kafka-Sender-App und die Ereignissubskription erstellen, vereinfacht diese Aktion die erforderlichen Schritte.

   Fügen Sie zum Erstellen des geheimen Schlüssels kafka-subscription-secret eine Literalumgebungsvariable für password und username hinzu. Weitere Informationen finden Sie unter Geheimen Schlüssel mit der Befehlszeilenschnittstelle erstellen.

   • Geben Sie den Schlüssel username mit dem Wert user an, der in den Details der Serviceberechtigungsnachweise in der Serviceinstanz Event Streams aufgelistet ist. Für die Serviceinstanz von Event Streams lautet dieser Wert token. Dieser Schlüssel ist für die Authentifizierung zwischen der Kafka-Ereignissubskription Code Engine und dem Kafka-Nachrichtenbroker erforderlich.

   • Geben Sie den Schlüssel password mit dem Wert apikey an, der in den Details der Serviceberechtigungsnachweise in der Serviceinstanz Event Streams aufgelistet ist. Dieser Schlüssel ist für das Senderbeispiel und zum Aktivieren der Kommunikation zwischen der Ereignissubskription Code Engine Kafka und dem Kafka-Nachrichtenbroker erforderlich.

     ibmcloud ce secret create --name kafka-subscription-secret --from-literal password=<value_of_apikey> --from-literal username=<value_of_user>
     

     Beispiel:

     ibmcloud ce secret create --name kafka-subscription-secret --from-literal password=abcdeH9tu3qE5Sn8VbJfcDEWtjR_l0iPisB3abcdefgh --from-literal username=token
     

2. Erstellen Sie die kafka-sender-app mit den folgenden Informationen.

   • Geben Sie die Option --image an, um auf das Container-Image icr.io/codeengine/kafka-sender zu verweisen. Dieses Image wird aus sender.go erstellt, das über die -Beispiele für IBM Cloud Code Engine GitHub repoverfügbar ist. Diese Beispiel-Sender-App erfordert die password-Berechtigungsnachweise, die in Ihrem kafka-subscription-secret gespeichert sind, und die Umgebungsvariable BROKERS.

   • Geben Sie die Option --env-from-secret an, um auf den vollständigen geheimen Schlüssel kafka-subscription-secret zu verweisen, der die password-Berechtigungsnachweise enthält.

   • Geben Sie die Option --env an, um eine Literalumgebungsvariable ( BROKERS) hinzuzufügen, und geben Sie den Namen eines der Broker-Hosts an, die in den Details der Serviceberechtigungsnachweise in der Serviceinstanz Event Streams aufgelistet sind. Wenn Sie jedoch mehrere Broker-Hostnamen angeben möchten, verwenden Sie das Format --env BROKERS-broker1,broker2,broker3.

   • (optional) Geben Sie die Option --min-scale=1 an, damit die Anwendung immer eine laufende Instanz hat und nicht auf Null skaliert. Eine solche Konfiguration, bei der die Anwendung stets über eine aktive Instanz verfügt, ist nützlich, wenn Sie Protokolle anzeigen. Wenn Sie in einer Produktionsumgebung arbeiten, berücksichtigen Sie die Kosten für das Beibehalten einer aktiven Instanz Ihrer App oder ob Code Engine automatisch auf null skaliert werden soll. Standardmäßig wird die App auf null skaliert, wenn sie nicht verwendet wird.

     ibmcloud ce app create --name kafka-sender-app --image icr.io/codeengine/kafka-sender --env-from-secret kafka-subscription-secret --env BROKERS=broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --min-scale 1
     

3. Führen Sie nach der Bereitstellung dieser App den Befehl app get aus, um zu bestätigen, dass sich die App im Status ready befindet.

   ibmcloud ce app get -n kafka-sender-app
   

Sie haben die App kafka-sender-app erstellt, um Kafka Nachrichten für Code Engine-Ereignissubskriptionen zu erstellen, und Sie haben den geheimen Schlüssel kafka-subscription-secret erstellt, der die erforderlichen Berechtigungsnachweise enthält.

Code Engine Kafka-Subskription einrichten

Damit Code Engine mit Kafka-Ereignissen arbeiten kann, richten Sie eine Code Engine Kafka-Ereignissubskription ein, um eine Verbindung zu Kafka-Ereignisbrokern herzustellen und für Kafka-Ereignisse empfangsbereit zu sein. Richten Sie außerdem eine App Code Engine als Empfänger der Kafka-Ereignisse ein. Die Kafka-Ereignissubskription definiert die Beziehung zwischen dem Kafka-Produzenten (Sender) und dem Konsumenten (Empfänger) von Ereignissen.

Die Kafka-Ereignissubskription Code Engine stellt eine Verbindung zu Ihrem Kafka-Nachrichtenbroker her und sendet HTTP-Post-Anforderungen für jede eingehende Kafka-Nachricht an die Empfängeranwendung. Weitere Informationen zu den Informationen, die in Kafka-Ereignissen enthalten sind, finden Sie unter HTTP-Header und -Hauptteil für Ereignisse, die an Apps zugestellt werden.

1. Erstellen Sie eine Anwendung Code Engine, die als Ereigniskonsument von Kafka-Nachrichten fungiert und die Kafka-Ereignisse empfängt. Erstellen Sie beispielsweise eine Anwendung mit dem Namen kafka-receiver-app, die das icr.io/codeengine/kafka-receiver-Image verwendet. Dieses Image wird aus receiver.go erstellt, das über die -Beispiele für IBM Cloud Code Engine GitHub repoverfügbar ist. Dieses Beispiel erfordert keine Umgebungsvariablen. Sie können optional die Option --min-scale=1 angeben, so dass die Anwendung immer eine laufende Instanz hat und nicht auf Null skaliert. Eine solche Konfiguration, bei der die Anwendung stets über eine aktive Instanz verfügt, ist nützlich, wenn Sie Protokolle anzeigen. Wenn Sie in einer Produktionsumgebung arbeiten, berücksichtigen Sie die Kosten für das Beibehalten einer aktiven Instanz Ihrer App oder ob Code Engine automatisch auf null skaliert werden soll. Standardmäßig wird die App auf null skaliert, wenn sie nicht verwendet wird.

   ibmcloud ce app create -n kafka-receiver-app --image icr.io/codeengine/kafka-receiver --min-scale 1
   

   Ereignisse werden standardmäßig an die Stamm-URL der Zielanwendung weitergeleitet. Sie können Ereignisse an ein anderes Ziel innerhalb der App senden, indem Sie die Option --path verwenden. Wenn Ihre Subskription beispielsweise --path /event angibt, wird das Ereignis an https://<base application URL>/events gesendet.

2. Führen Sie nach der Bereitstellung dieser App den Befehl app get aus, um zu bestätigen, dass sich die App im Status ready befindet.

   ibmcloud ce app get -n kafka-receiver-app
   

3. Erstellen Sie eine Code Engine Kafka-Ereignissubskription für Ihre Kafka-Ereignisse mit dem Befehl ibmcloud ce sub kafka create. Verwenden Sie den geheimen Schlüssel kafka-subscription-secret, den Sie zuvor erstellt haben, um auf den Nachrichtenbroker zuzugreifen. Geben Sie die Brokerinformationen basierend auf den Serviceberechtigungsinformationen für Ihre Kafka-Ressource an. In diesem Beispiel können Sie die Brokerinformationen aus der Ausgabe des Befehls ibmcloud resource service-key myeventstream-key abrufen. Beachten Sie, dass Sie für jeden Broker für Ihr Thema die Option --broker angeben müssen. Die Option --destination gibt die Ressource Code Engine an, die die Ereignisse empfängt.

   ibmcloud ce sub kafka create --name mykafkasubscription --destination kafka-receiver-app --secret kafka-subscription-secret --topic kafka-topic1 --broker broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker  broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-1-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-2-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   

4. Anzeige der Details des Kafka.

   ibmcloud ce sub kafka get -n mykafkasubscription
   

   Beispielausgabe

   Getting Kafka event subscription 'mykafkasubscription'...
   OK
   
   Name:          mykafkasubscription
   [...]
   Destination Type:                 app
   Destination:                      kafka-receiver-app
   Brokers:
   broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   broker-1-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   broker-2-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   Consumer Group:                   knative-kafka-source-a4072fe1-1dfa-4470-9d07-bf7a0ff8e340
   Topics:
   kafka-topic1
   Secret key reference (user):      kafka-subscription-secret.username
   Secret key reference (password):  kafka-subscription-secret.password
   Ready:                            true
   
   Conditions:
   Type                     OK    Age  Reason
   ConnectionEstablished    true  24s
   InitialOffsetsCommitted  true  24s
   Ready                    true  24s
   Scheduled                true  24s
   SinkProvided             true  24s
   
   Events:
   Type     Reason           Age  Source                  Messages
   Normal   FinalizerUpdate  26s  kafkasource-controller  Updated "mykafkasubscription" finalizers
   

Abonnement testen

Nachdem Ihre Kafka-Ereignissubskription, die auf die kafka-receiver-app-Anwendung verweist, erstellt wurde, verwenden Sie kafka-sender-app, um Nachrichtenereignisse an die Empfängeranwendung zu senden.

1. Rufen Sie die öffentliche URL der Ziel-App ab, kafka-sender-app indem Sie den Befehl ibmcloud ce app get mit der Option --output url verwenden, um die URL Ihrer App zu suchen.

   ibmcloud ce app get -n kafka-sender-app --output url
   

   Beispielausgabe

   https://kafka-sender-app.abcdabcdabc.us-south.codeengine.appdomain.cloud
   

2. Führen Sie die Kafka-Ereigniserzeuger-App kafka-sender-app aus, um Ereignisse an die Zielanwendung Code Engine zu senden. Rufen Sie die Anwendung kafka-sender-app mit curl auf und geben Sie Werte für das Topic und die Anzahl der Nachrichten an. Verwenden Sie die Ausgabe des Befehls ibmcloud ce app get, um die öffentliche URL Ihrer Anwendung zu finden. Achten Sie darauf, den Wert für 'curl' in Anführungszeichen einzuschließen, um sicherzustellen, dass er als einzelne Zeichenfolge behandelt wird. Beispiel:

   curl "<public_URL_of_Kafka_sender_app>?topic=<your_topic_name>&num=<number_of_messages_to_produce>"
   

   Beispiel:

   curl "https://kafka-sender-app.abcdabcdabc.us-south.codeengine.appdomain.cloud?topic=kafka-topic1&num=1"
   

3. Ereignisse in Protokollen anzeigen. Wenn Ihre Kafka-Ereignissubskription mit einem Broker, gültigen Topics und einem gültigen geheimen Zugriffsschlüssel erstellt wird und Sie über eine Kafka-Anwendung verfügen, die Nachrichten zu diesem Topic erzeugt (z. B. kafka-sender-app), können Sie Ereignisse in Protokollen für Ihre Zielanwendung Code Engine anzeigen, die Kafka-Nachrichten wie kafka-receiver-app empfängt. Wenn Sie die Empfängerapp Kafka (icr.io/codeengine/kafka-receiver) verwenden, suchen Sie in den Protokollen für die Empfängeranwendung nach Event data, um die empfangenen Nachrichten anzuzeigen.

   ibmcloud ce app logs -n kafka-receiver-app
   

   Beispielausgabe

   Getting logs for all instances of application 'kafka-receiver-app'...
   OK
   
   kafka-receiver-app-00001-deployment-66976f7988-9xttm/user-container:
   2022/03/31 22:19:45 Listening on port 8080
   2022/03/31 22:19:46 ----------
   2022/03/31 22:19:46 Path: /
   2022/03/31 22:19:46 Header: Accept-Encoding=[gzip]
   2022/03/31 22:19:46 Header: Ce-Id=[partition:0/offset:167]
   2022/03/31 22:19:46 Header: Ce-Source=[/apis/v1/namespaces/glxo4k7nj7d/kafkasources/mykafkasubscription#kafka-topic1]
   2022/03/31 22:19:46 Header: Ce-Specversion=[1.0]
   2022/03/31 22:19:46 Header: Ce-Subject=[partition:0#167]
   2022/03/31 22:19:46 Header: Ce-Time=[2022-03-31T22:19:36.499Z]
   2022/03/31 22:19:46 Header: Ce-Type=[dev.knative.kafka.event]
   2022/03/31 22:19:46 Header: Content-Length=[8]
   2022/03/31 22:19:46 Header: Forwarded=[for=172.30.208.213;proto=http, for=127.0.0.6]
   2022/03/31 22:19:46 Header: K-Proxy-Request=[activator]
   2022/03/31 22:19:46 Header: Traceparent=[00-b033708685c715a7c2384cdf05797785-65540b0937e9b0ce-00]
   2022/03/31 22:19:46 Header: User-Agent=[Go-http-client/1.1]
   2022/03/31 22:19:46 Header: X-B3-Parentspanid=[e1a785d7fdbead6c]
   2022/03/31 22:19:46 Header: X-B3-Sampled=[1]
   2022/03/31 22:19:46 Header: X-B3-Spanid=[abcde9901e6bf83f]
   2022/03/31 22:19:46 Header: X-B3-Traceid=[abcde490a426573772fa0bf60caf5ddb]
   2022/03/31 22:19:46 Header: X-Envoy-Attempt-Count=[1]
   2022/03/31 22:19:46 Header: X-Forwarded-For=[172.30.208.213, 127.0.0.6, 127.0.0.6]
   2022/03/31 22:19:46 Header: X-Forwarded-Proto=[http]
   2022/03/31 22:19:46 Header: X-Request-Id=[abcdeb4e-c5ac-abcd-abcd-60e6278abcde]
   2022/03/31 22:19:46 Event data: test1: 1
   

   Beachten Sie, dass die Protokollinformationen für Anwendungen nur eine Stunde lang gültig sind. Weitere Informationen zum Anzeigen von Protokollen für Apps (oder Jobs) finden Sie unter Protokolle anzeigen.

Abonnement aktualisieren

Verwenden Sie den Befehl ibmcloud ce subscription kafka update, um eine Ereignissubskription über die Befehlszeilenschnittstelle zu aktualisieren. Im folgenden Beispiel wird der Themenname aktualisiert.

ibmcloud ce sub kafka update -n mykafkasubscription --topic kafka-topic2

Sie können den Befehl ibmcloud ce subscription kafka update verwenden, um die Werte für die Subskription Kafka zu aktualisieren. Sie können den Wert für die Consumergruppe jedoch nicht mit diesem Befehl ändern. Wenn Sie die Subskription aktualisieren möchten, um auf ein anderes Topic zu verweisen, stellen Sie sicher, dass das Topic Kafka vorhanden ist, bevor Sie die Subskription aktualisieren.

Aufräumen für das Kafka

Möchten Sie Ihr Kafka-Abonnement, die sendenden und empfangenden Apps und den geheimen Schlüssel löschen? Sie können die Befehle ibmcloud ce app delete, ibmcloud ce sub kafka delete und ibmcloud ce sub kafka delete verwenden. Sie können optional die Option -f verwenden, um das Löschen der Komponente ohne Bestätigung zu erzwingen.

Wenn Sie die Kafka-Subskription löschen, wird die App, auf die die Subskription verweist, nicht gelöscht.

Zum Entfernen des Abonnements lautet der Befehl wie folgt:

ibmcloud ce sub kafka delete --name mykafkasubscription -f

Gehen Sie wie folgt vor, um Ihre Kafka-Nachrichtenempfanganwendung zu entfernen:

ibmcloud ce app delete --name kafka-receiver-app -f

Ebenso können Sie die kafka-sender-app entfernen.

ibmcloud ce app delete --name kafka-sender-app -f

Gehen Sie wie folgt vor, um kafka-subscription-secret zu entfernen:

ibmcloud ce secret delete --name kafka-subscription-secret -f

Möchten Sie Ihre Serviceinstanz für die Serviceinstanz von Event Streams löschen? Die Option --recursive gibt an, dass alle Ressourcen für die Serviceinstanz, einschließlich des zugehörigen Serviceschlüssels, entfernt werden sollen.

ibmcloud resource service-instance-delete myeventstream --recursive -f

Nächste Schritte

Weitere Informationen zum Arbeiten mit Kafka-Ereignissubskriptionen enthält der Abschnitt Mit dem Ereigniserzeuger Kafka arbeiten.

Benötigen Sie weitere Codebeispiele? Sehen Sie sich die Beispiele für IBM Cloud Code Engine GitHub repoan.