2.3. Connecting a data source to a Kafka topic in a kamelet binding

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

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

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

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

kamelet バインディングのソースの場合は、データソース kamelet を指定し (たとえば、coffee-source kamelet はコーヒーに関するデータが含まれるイベントを生成します)、kamelet のパラメーターを設定します。

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: coffees-to-kafka
spec:
  source:
    ref:
      kind: Kamelet
      apiVersion: camel.apache.org/v1alpha1
      name: coffee-source
    properties:
      period: 5000
  sink:

kamelet binding のシンクの場合は、kafka-sink kamelets およびその必要なパラメーターを指定します。

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

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

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

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

KameletBinding をリソースとして OpenShift namespace に追加します。

oc apply -f <kamelet binding filename>

以下に例を示します。

oc apply -f coffees-to-kafka.yaml

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

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

oc get kameletbindings

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

oc get integrations

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

kamel logs <integration> -n <project>

以下に例を示します。

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

結果は以下の出力に似ています。

...
[1] 2021-06-21 07:48:12,696 INFO  [io.quarkus] (main) camel-k-integration 1.4.0 on JVM (powered by Quarkus 1.13.0.Final) started in 2.790s.