3.3. Apache Kafka のシンク

手順

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

   Kafka シンク YAML

   apiVersion: eventing.knative.dev/v1alpha1
   kind: KafkaSink
   metadata:
     name: <sink-name>
     namespace: <namespace>
   spec:
     topic: <topic-name>
     bootstrapServers:
      - <bootstrap-server>

2. Kafka シンクを作成するには、KafkaSink YAML ファイルを適用します。

   $ oc apply -f <filename>

3. シンクが仕様で指定されるようにイベントソースを設定します。

   API サーバーソースに接続された Kafka シンクの例

   apiVersion: sources.knative.dev/v1alpha2
   kind: ApiServerSource
   metadata:
     name: <source-name> 1
     namespace: <namespace> 2
   spec:
     serviceAccountName: <service-account-name> 3
     mode: Resource
     resources:
     - apiVersion: v1
       kind: Event
     sink:
       ref:
         apiVersion: eventing.knative.dev/v1alpha1
         kind: KafkaSink
         name: <sink-name> 4

   1

   イベントソースの名前。

   2

   イベントソースの namespace。

   3

   イベントソースのサービスアカウント。

   4

   Kafka シンクの名前。

手順

1. Developer パースペクティブで、+Add ビューに移動します。
2. Eventing カタログ で Event Sink をクリックします。
3. カタログ項目で KafkaSink を検索してクリックします。
4. イベントシンクの作成 をクリックします。

5. フォームビューで、ホスト名とポートの組み合わせであるブートストラップサーバーの URL を入力します。

   
6. イベントデータを送信するトピックの名前を入力します。
7. イベントシンクの名前を入力します。
8. Create をクリックします。

検証

1. Developer パースペクティブで、Topology ビューに移動します。
2. 作成したイベントシンクをクリックして、右側のパネルに詳細を表示します。

手順

1. KafkaSink オブジェクトと同じ namespace に証明書ファイルをシークレットとして作成します。

   重要

   証明書とキーは PEM 形式である必要があります。

   • 暗号化なしで SASL を使用した認証の場合:

     $ oc create secret -n <namespace> generic <secret_name> \
       --from-literal=protocol=SASL_PLAINTEXT \
       --from-literal=sasl.mechanism=<sasl_mechanism> \
       --from-literal=user=<username> \
       --from-literal=password=<password>

   • SASL を使用した認証と TLS を使用した暗号化の場合:

     $ oc create secret -n <namespace> generic <secret_name> \
       --from-literal=protocol=SASL_SSL \
       --from-literal=sasl.mechanism=<sasl_mechanism> \
       --from-file=ca.crt=<my_caroot.pem_file_path> \ 1
       --from-literal=user=<username> \
       --from-literal=password=<password>

     1

     パブリッククラウドで管理される Kafka サービスを使用している場合は、ca.crt を省略してシステムのルート CA セットを使用できます。

   • TLS を使用した認証と暗号化の場合:

     $ oc create secret -n <namespace> generic <secret_name> \
       --from-literal=protocol=SSL \
       --from-file=ca.crt=<my_caroot.pem_file_path> \ 1
       --from-file=user.crt=<my_cert.pem_file_path> \
       --from-file=user.key=<my_key.pem_file_path>

     1

     パブリッククラウドで管理される Kafka サービスを使用している場合は、ca.crt を省略してシステムのルート CA セットを使用できます。

2. KafkaSink オブジェクトを作成または変更し、auth 仕様にシークレットへの参照を追加します。

   apiVersion: eventing.knative.dev/v1alpha1
   kind: KafkaSink
   metadata:
      name: <sink_name>
      namespace: <namespace>
   spec:
   ...
      auth:
        secret:
          ref:
            name: <secret_name>
   ...

3. KafkaSink オブジェクトを適用します。

   $ oc apply -f <filename>