2.5. Apache Kafka のソース

手順

1. Developer パースペクティブで、+Add ページに移動し、Event Source を選択します。
2. Event Sources ページで、Type セクションの Kafka Source を選択します。

3. Kafka Source 設定を設定します。

   1. ブートストラップサーバー のコンマ区切りの一覧を追加します。
   2. トピック のコンマ区切りの一覧を追加します。
   3. コンシューマーグループ を追加します。
   4. 作成したサービスアカウントの Service Account Name を選択します。
   5. イベントソースの Sink を選択します。Sink は、チャネル、ブローカー、またはサービスなどの Resource、または URI のいずれかになります。
   6. Kafka イベントソースの Name を入力します。
4. Create をクリックします。

1. Developer パースペクティブで、Topology に移動します。

2. Kafka イベントソースおよびシンクを表示します。

   

   View larger image

   

手順

1. Kafka イベントソースが機能していることを確認するには、受信メッセージをサービスのログにダンプする Knative サービスを作成します。

   $ kn service create event-display \
       --image quay.io/openshift-knative/knative-eventing-sources-event-display

   Copy to Clipboard Toggle word wrap

2. KafkaSource CR を作成します。

   $ kn source kafka create <kafka_source_name> \
       --servers <cluster_kafka_bootstrap>.kafka.svc:9092 \
       --topics <topic_name> --consumergroup my-consumer-group \
       --sink event-display

   Copy to Clipboard Toggle word wrap
   注記

   このコマンドのプレースホルダー値は、ソース名、ブートストラップサーバー、およびトピックの値に置き換えます。

   --servers、--topics、および --consumergroup オプションは、Kafka クラスターへの接続パラメーターを指定します。--consumergroup オプションは任意です。

3. オプション: 作成した KafkaSource CR の詳細を表示します。

   $ kn source kafka describe <kafka_source_name>

   Copy to Clipboard Toggle word wrap

   出力例

   Name:              example-kafka-source
   Namespace:         kafka
   Age:               1h
   BootstrapServers:  example-cluster-kafka-bootstrap.kafka.svc:9092
   Topics:            example-topic
   ConsumerGroup:     example-consumer-group
   
   Sink:
     Name:       event-display
     Namespace:  default
     Resource:   Service (serving.knative.dev/v1)
   
   Conditions:
     OK TYPE            AGE REASON
     ++ Ready            1h
     ++ Deployed         1h
     ++ SinkProvided     1h

   Copy to Clipboard Toggle word wrap

検証手順

1. Kafka インスタンスをトリガーし、メッセージをトピックに送信します。

   $ oc -n kafka run kafka-producer \
       -ti --image=quay.io/strimzi/kafka:latest-kafka-2.7.0 --rm=true \
       --restart=Never -- bin/kafka-console-producer.sh \
       --broker-list <cluster_kafka_bootstrap>:9092 --topic my-topic

   Copy to Clipboard Toggle word wrap

   プロンプトにメッセージを入力します。このコマンドは、以下を前提とします。

   • Kafka クラスターが kafka namespace にインストールされている。
   • KafkaSource オブジェクトは、my-topic トピックを使用するように設定されている。

2. ログを表示して、メッセージが到達していることを確認します。

   $ oc logs $(oc get pod -o name | grep event-display) -c user-container

   Copy to Clipboard Toggle word wrap

   出力例

   ☁️  cloudevents.Event
   Validation: valid
   Context Attributes,
     specversion: 1.0
     type: dev.knative.kafka.event
     source: /apis/v1/namespaces/default/kafkasources/example-kafka-source#example-topic
     subject: partition:46#0
     id: partition:46/offset:0
     time: 2021-03-10T11:21:49.4Z
   Extensions,
     traceparent: 00-161ff3815727d8755848ec01c866d1cd-7ff3916c44334678-00
   Data,
     Hello!

   Copy to Clipboard Toggle word wrap

手順

1. KafkaSource オブジェクトを YAML ファイルとして作成します。

   apiVersion: sources.knative.dev/v1beta1
   kind: KafkaSource
   metadata:
     name: <source_name>
   spec:
     consumerGroup: <group_name> 

   1

   
     bootstrapServers:
     - <list_of_bootstrap_servers>
     topics:
     - <list_of_topics> 

   2

   
     sink:
     - <list_of_sinks> 

   3

   Copy to Clipboard Toggle word wrap

   1

   コンシューマーグループは、同じグループ ID を使用し、トピックからデータを消費するコンシューマーのグループです。

   2

   トピックは、データの保存先を提供します。各トピックは、1 つまたは複数のパーティションに分割されます。

   3

   シンクは、イベントがソースから送信される場所を指定します。

   重要

   OpenShift Serverless 上の KafkaSource オブジェクトの API の v1beta1 バージョンのみがサポートされます。非推奨となった v1alpha1 バージョンの API は使用しないでください。

   KafkaSource オブジェクトの例

   apiVersion: sources.knative.dev/v1beta1
   kind: KafkaSource
   metadata:
     name: kafka-source
   spec:
     consumerGroup: knative-group
     bootstrapServers:
     - my-cluster-kafka-bootstrap.kafka:9092
     topics:
     - knative-demo-topic
     sink:
       ref:
         apiVersion: serving.knative.dev/v1
         kind: Service
         name: event-display

   Copy to Clipboard Toggle word wrap

2. KafkaSource YAML ファイルを適用します。

   $ oc apply -f <filename>

   Copy to Clipboard Toggle word wrap

手順

1. 選択された namespace にシークレットとして証明書ファイルを作成します。

   $ oc create secret -n <namespace> generic <kafka_auth_secret> \
     --from-file=ca.crt=caroot.pem \
     --from-literal=password="SecretPassword" \
     --from-literal=saslType="SCRAM-SHA-512" \ 

   1

   
     --from-literal=user="my-sasl-user"

   Copy to Clipboard Toggle word wrap

   1

   SASL タイプは PLAIN、SCRAM-SHA-256、または SCRAM-SHA-512 です。

2. Kafka ソースを作成または変更して、次の spec 設定が含まれるようにします。

   apiVersion: sources.knative.dev/v1beta1
   kind: KafkaSource
   metadata:
     name: example-source
   spec:
   ...
     net:
       sasl:
         enable: true
         user:
           secretKeyRef:
             name: <kafka_auth_secret>
             key: user
         password:
           secretKeyRef:
             name: <kafka_auth_secret>
             key: password
         type:
           secretKeyRef:
             name: <kafka_auth_secret>
             key: saslType
       tls:
         enable: true
         caCert: 

   1

   
           secretKeyRef:
             name: <kafka_auth_secret>
             key: ca.crt
   ...

   Copy to Clipboard Toggle word wrap

   1

   パブリッククラウドの Kafka サービスを使用している場合は、caCert 仕様は必要ありません。