2.4. Connecting a Kafka topic to a data sink in a kamelet binding

任意のエディターで、以下の基本構造で YAML ファイルを作成します。

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name:
spec:
  source:
  sink:

kamelet binding の名前を追加します。この例では、バインディングが kafka-source kamelet を log-sink kamelet に接続するため、名前は kafka-to-log になります。

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: kafka-to-log
spec:
  source:
  sink:

kamelet バインディングのソースの場合は、kafka-source kamelet を指定し、そのパラメーターを設定します。

たとえば、Kafka クラスターが OpenShift Streams にある場合 (securityProtocol パラメーターを設定する必要はありません)。

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: kafka-to-log
spec:
  source:
    ref:
      kind: Kamelet
      apiVersion: camel.apache.org/v1alpha1
      name: kafka-source
    properties:
      bootstrapServers: "broker.url:9092"
      password: "testpassword"
      topic: "test-topic"
      user: "testuser"
  sink:

たとえば、Kafka クラスターが AMQ Streams にある場合は、securityProtocol パラメーターを “PLAINTEXT” に設定する必要があります。

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: kafka-to-log
spec:
  source:
    ref:
      kind: Kamelet
      apiVersion: camel.apache.org/v1alpha1
      name: kafka-source
    properties:
      bootstrapServers: "broker.url:9092"
      password: "testpassword"
      topic: "test-topic"
      user: "testuser"
      securityProtocol: "PLAINTEXT"
  sink:

kamelet バインディングのシンクでは、データコンシューマー kamelet (例: log-sink kamelet) を指定し、kamelet のパラメーターを設定します。以下に例を示します。

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: kafka-to-log
spec:
  source:
    ref:
      kind: Kamelet
      apiVersion: camel.apache.org/v1alpha1
      name: kafka-source
    properties:
      bootstrapServers: "broker.url:9092"
      password: "testpassword"
      topic: "test-topic"
      user: "testuser"
      securityProtocol: "PLAINTEXT" // only for AMQ streams
  sink:
    ref:
      kind: Kamelet
      apiVersion: camel.apache.org/v1alpha1
      name: log-sink

kamelet バインディングをリソースとして OpenShift namespace に追加します。

oc apply -f <kamelet binding filename>

以下に例を示します。

oc apply -f kafka-to-log.yaml

Camel K Operator は、KameletBinding リソースを使用して Camel K インテグレーションを生成し、実行します。ビルドに数分かかる場合があります。

kamelet バインディングリソースのステータスを表示するには、次のコマンドを実行します。

oc get kameletbindings

インテグレーションの状態を表示するには、以下を実行します。

oc get integrations

インテグレーションのログを表示するには、以下を実行します。

kamel logs <integration> -n <project>

以下に例を示します。

kamel logs kafka-to-log -n my-camel-k-kafka

この出力では、以下の例のようにコーヒーイベントが表示されるはずです。

2021-07-20 17:15:56,725 INFO  [log-sink-E80C5C904418150-0000000000000001] (Camel (camel-1) thread #0 - timer://tick) {"id":7259,"uid":"a4ecb7c2-05b8-4a49-b0d2-d1e8db5bc5e2","blend_name":"Postmodern Symphony","origin":"Huila, Colombia","variety":"Kona","notes":"delicate, chewy, black currant, red apple, star fruit","intensifier":"balanced"}

実行中のインテグレーションを停止するには、関連付けられた kamelet バインディングリソースを削除します。

oc delete kameletbindings/<kameletbinding-name>

以下に例を示します。

oc delete kameletbindings/kafka-to-log