Camel K を使用したインテグレーションの開発および管理

手順

1. 以下の例のように、Camel K Operator が OpenShift クラスターで稼働していることを確認します。

   oc get pod

   Copy to Clipboard Toggle word wrap

   NAME                               READY   STATUS    RESTARTS   AGE
   camel-k-operator-86b8d94b4-pk7d6   1/1     Running   0          6m28s

   Copy to Clipboard Toggle word wrap

2. kamel run コマンドを入力し、OpenShift のクラウドでインテグレーションを実行します。以下に例を示します。

   kamel run hello.camelk.yaml

   Copy to Clipboard Toggle word wrap

   integration "hello" created

   Copy to Clipboard Toggle word wrap

3. kamel get コマンドを入力し、インテグレーションの状態を確認します。

   <<<<<<< HEAD
   $ kamel get
   NAME PHASE KIT
   hello Building Kit myproject/kit-c4ci6mbe9hl5ph5c9sjg
   =======
   kamel get

   Copy to Clipboard Toggle word wrap

   NAME   PHASE        KIT
   hello  Building Kit kit-bqatqib5t4kse5vukt40
   >>>>>>> fusedoc-4502

   Copy to Clipboard Toggle word wrap

4. kamel describe コマンドを入力し、インテグレーションに関する詳細情報を出力します。

   <<<<<<< HEAD
   $ kamel describe it hello
   =======
   kamel describe integration hello

   Copy to Clipboard Toggle word wrap

   >>>>>>> fusedoc-4502
   Name:                hello
   Namespace:           myproject
   Creation Timestamp:  Fri, 13 Aug 2021 16:23:21 +0200
   Phase:               Building Kit
   Runtime Version:     1.7.1.fuse-800025-redhat-00001
   Kit:                 myproject/kit-c4ci6mbe9hl5ph5c9sjg
   Image:
   Version:             1.4.0
   Dependencies:
     camel:core
     camel:log
     camel:timer
     mvn:org.apache.camel.k:camel-k-runtime
     mvn:org.apache.camel.quarkus:camel-quarkus-yaml-dsl
   Sources:
     Name                        Language  Compression  Ref  Ref Key
     camel-k-embedded-flow.yaml  yaml      false
   Conditions:
     Type                          Status  Reason                        Message
     IntegrationPlatformAvailable  True    IntegrationPlatformAvailable  myproject/camel-k
     IntegrationKitAvailable       True    IntegrationKitAvailable       kit-c4ci6mbe9hl5ph5c9sjg
     CronJobAvailable              False   CronJobNotAvailableReason     different controller strategy used (deployment)
     DeploymentAvailable           True    DeploymentAvailable           deployment name is hello
     KnativeServiceAvailable       False   KnativeServiceNotAvailable    different controller strategy used (deployment)
     Ready                         True    ReplicaSetReady

   Copy to Clipboard Toggle word wrap

5. kamel log コマンドを入力して、ログを stdout に出力します。

   kamel log hello

   Copy to Clipboard Toggle word wrap

   ...
   [1] 2021-08-13 14:37:15,860 INFO [info] (Camel (camel-1) thread #0 - timer://yaml) Exchange[ExchangePattern: InOnly, BodyType: String, Body: Hello Camel K from yaml]
   ...

   Copy to Clipboard Toggle word wrap
6. Ctrl-C キーを押して、ターミナルでログインを終了します。

7. kamel delete を入力して、OpenShift にデプロイされたインテグレーションを削除します。

   kamel delete hello

   Copy to Clipboard Toggle word wrap

   Integration hello deleted

   Copy to Clipboard Toggle word wrap

手順

1. kamel run コマンドを入力し、--property オプションを使用してログレベルを指定します。以下に例を示します。

   kamel run --dev --property 'quarkus.log.category."org.apache.camel.support".level'=DEBUG Basic.java
   
   ...
   integration "basic" created
       Progress: integration "basic" in phase Initialization
       Progress: integration "basic" in phase Building Kit
       Progress: integration "basic" in phase Deploying
       Condition "IntegrationPlatformAvailable" is "True" for Integration basic: myproject/camel-k
       Integration basic in phase "Initialization"
       Integration basic in phase "Building Kit"
       Integration basic in phase "Deploying"
       Condition "IntegrationKitAvailable" is "True" for Integration basic: kit-c4dn5l62v9g3aopkocag
       Condition "DeploymentAvailable" is "True" for Integration basic: deployment name is basic
       Condition "CronJobAvailable" is "False" for Integration basic: different controller strategy used (deployment)
       Progress: integration "basic" in phase Running
       Condition "KnativeServiceAvailable" is "False" for Integration basic: different controller strategy used (deployment)
       Integration basic in phase "Running"
       Condition "Ready" is "False" for Integration basic
       Condition "Ready" is "True" for Integration basic
       [1] Monitoring pod basic-575b97f64b-7l5rl
       [1] 2021-08-17 08:35:22,906 DEBUG [org.apa.cam.sup.LRUCacheFactory] (main) Creating DefaultLRUCacheFactory
       [1] 2021-08-17 08:35:23,132 INFO  [org.apa.cam.k.Runtime] (main) Apache Camel K Runtime 1.7.1.fuse-800025-redhat-00001
       [1] 2021-08-17 08:35:23,134 INFO  [org.apa.cam.qua.cor.CamelBootstrapRecorder] (main) bootstrap runtime: org.apache.camel.quarkus.main.CamelMainRuntime
       [1] 2021-08-17 08:35:23,224 INFO  [org.apa.cam.k.lis.SourcesConfigurer] (main) Loading routes from: SourceDefinition{name='Basic', language='java', location='file:/etc/camel/sources/Basic.java', }
       [1] 2021-08-17 08:35:23,232 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Found RoutesBuilderLoader: org.apache.camel.dsl.java.joor.JavaRoutesBuilderLoader via: META-INF/services/org/apache/camel/java
       [1] 2021-08-17 08:35:23,232 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Detected and using RoutesBuilderLoader: org.apache.camel.dsl.java.joor.JavaRoutesBuilderLoader@68dc098b
       [1] 2021-08-17 08:35:23,236 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Found ResourceResolver: org.apache.camel.impl.engine.DefaultResourceResolvers$FileResolver via: META-INF/services/org/apache/camel/file
       [1] 2021-08-17 08:35:23,237 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Detected and using ResourceResolver: org.apache.camel.impl.engine.DefaultResourceResolvers$FileResolver@5b67bb7e
       [1] 2021-08-17 08:35:24,320 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Lookup Language with name simple in registry. Found: org.apache.camel.language.simple.SimpleLanguage@74d7184a
       [1] 2021-08-17 08:35:24,328 DEBUG [org.apa.cam.sup.EventHelper] (main) Ignoring notifying event Initializing CamelContext: camel-1. The EventNotifier has not been started yet: org.apache.camel.quarkus.core.CamelManagementEventBridge@3301500b
       [1] 2021-08-17 08:35:24,336 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Lookup Component with name timer in registry. Found: org.apache.camel.component.timer.TimerComponent@3ef41c66
       [1] 2021-08-17 08:35:24,342 DEBUG [org.apa.cam.sup.DefaultComponent] (main) Creating endpoint uri=[timer://java?period=1000], path=[java]
       [1] 2021-08-17 08:35:24,350 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Found ProcessorFactory: org.apache.camel.processor.DefaultProcessorFactory via: META-INF/services/org/apache/camel/processor-factory
       [1] 2021-08-17 08:35:24,351 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Detected and using ProcessorFactory: org.apache.camel.processor.DefaultProcessorFactory@704b2127
       [1] 2021-08-17 08:35:24,369 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Found InternalProcessorFactory: org.apache.camel.processor.DefaultInternalProcessorFactory via: META-INF/services/org/apache/camel/internal-processor-factory
       [1] 2021-08-17 08:35:24,369 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Detected and using InternalProcessorFactory: org.apache.camel.processor.DefaultInternalProcessorFactory@4f8caaf3
       [1] 2021-08-17 08:35:24,442 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Lookup Component with name log in registry. Found: org.apache.camel.component.log.LogComponent@46b695ec
       [1] 2021-08-17 08:35:24,444 DEBUG [org.apa.cam.sup.DefaultComponent] (main) Creating endpoint uri=[log://info], path=[info]
       [1] 2021-08-17 08:35:24,461 DEBUG [org.apa.cam.sup.EventHelper] (main) Ignoring notifying event Initialized CamelContext: camel-1. The EventNotifier has not been started yet: org.apache.camel.quarkus.core.CamelManagementEventBridge@3301500b
       [1] 2021-08-17 08:35:24,467 DEBUG [org.apa.cam.sup.DefaultProducer] (main) Starting producer: Producer[log://info]
       [1] 2021-08-17 08:35:24,469 DEBUG [org.apa.cam.sup.DefaultConsumer] (main) Build consumer: Consumer[timer://java?period=1000]
       [1] 2021-08-17 08:35:24,475 DEBUG [org.apa.cam.sup.DefaultConsumer] (main) Starting consumer: Consumer[timer://java?period=1000]
       [1] 2021-08-17 08:35:24,481 INFO  [org.apa.cam.imp.eng.AbstractCamelContext] (main) Routes startup summary (total:1 started:1)
       [1] 2021-08-17 08:35:24,481 INFO  [org.apa.cam.imp.eng.AbstractCamelContext] (main)     Started java (timer://java)
       [1] 2021-08-17 08:35:24,482 INFO  [org.apa.cam.imp.eng.AbstractCamelContext] (main) Apache Camel 3.10.0.fuse-800010-redhat-00001 (camel-1) started in 170ms (build:0ms init:150ms start:20ms)
       [1] 2021-08-17 08:35:24,487 INFO  [io.quarkus] (main) camel-k-integration 1.4.0 on JVM (powered by Quarkus 1.11.7.Final-redhat-00009) started in 2.192s.
       [1] 2021-08-17 08:35:24,488 INFO  [io.quarkus] (main) Profile prod activated.
       [1] 2021-08-17 08:35:24,488 INFO  [io.quarkus] (main) Installed features: [camel-bean, camel-core, camel-java-joor-dsl, camel-k-core, camel-k-runtime, camel-log, camel-support-common, camel-timer, cdi]
       [1] 2021-08-17 08:35:25,493 INFO  [info] (Camel (camel-1) thread #0 - timer://java) Exchange[ExchangePattern: InOnly, BodyType: String, Body: Hello Camel K from java]
       [1] 2021-08-17 08:35:26,479 INFO  [info] (Camel (camel-1) thread #0 - timer://java) Exchange[ExchangePattern: InOnly, BodyType: String, Body: Hello Camel K from java]
   ...

   Copy to Clipboard Toggle word wrap
2. Ctrl-C キーを押して、ターミナルでログインを終了します。

手順

1. 以下のコマンドを入力して、cluster-monitoring-config ConfigMap オブジェクトが openshift-monitoring project に存在するかどうかを確認します。

   $ oc -n openshift-monitoring get configmap cluster-monitoring-config

   Copy to Clipboard Toggle word wrap

2. cluster-monitoring-config ConfigMap がない場合は作成します。

   $ oc -n openshift-monitoring create configmap cluster-monitoring-config

   Copy to Clipboard Toggle word wrap

3. cluster-monitoring-config ConfigMap を編集します。

   $ oc -n openshift-monitoring edit configmap cluster-monitoring-config

   Copy to Clipboard Toggle word wrap

4. data:config.yaml: で、enableUserWorkload を true に設定します。

   apiVersion: v1
   kind: ConfigMap
   metadata:
     name: cluster-monitoring-config
     namespace: openshift-monitoring
   data:
     config.yaml: |
       enableUserWorkload: true

   Copy to Clipboard Toggle word wrap

手順

1. 以下のコマンドを入力して、Prometheus トレイトを有効にして Camel K インテグレーションを実行します。

   kamel run myIntegration.java -t prometheus.enabled=true

   Copy to Clipboard Toggle word wrap

   または、以下のようにインテグレーションプラットフォームを更新すると、Prometheus トレイトを 1 度グローバルに有効にすることができます。

   $ oc patch ip camel-k --type=merge -p '{"spec":{"traits":{"prometheus":{"configuration":{"enabled":true}}}}}'

   Copy to Clipboard Toggle word wrap
2. Prometheus で Camel K インテグレーションメトリクスのモニタリングを確認します。たとえば、埋め込み Prometheus の場合は、OpenShift 管理者または開発者 Web コンソールで Monitoring > Metrics と選択します。
3. 表示する Camel K メトリクスを入力します。たとえば、Administrator コンソールの Insert Metric at Cursor で application_camel_context_uptime_seconds を入力し、Run Queries をクリックします。
4. Add Query をクリックして追加のメトリクスを表示します。

手順

1. Camel MicroProfile Metrics コンポーネントアノテーションを使用して、カスタムメトリクスを Camel インテグレーションコードに登録します。以下の例は、Metrics.java インテグレーションを示しています。

   // camel-k: language=java trait=prometheus.enabled=true dependency=mvn:org.my/app:1.0 

   1

   
   
   import org.apache.camel.Exchange;
   import org.apache.camel.LoggingLevel;
   import org.apache.camel.builder.RouteBuilder;
   import org.apache.camel.component.microprofile.metrics.MicroProfileMetricsConstants;
   
   import javax.enterprise.context.ApplicationScoped;
   
   @ApplicationScoped
   public class Metrics extends RouteBuilder {
   
      @Override
      public void configure() {
           onException()
               .handled(true)
               .maximumRedeliveries(2)
               .logStackTrace(false)
               .logExhausted(false)
               .log(LoggingLevel.ERROR, "Failed processing ${body}")
               // Register the 'redelivery' meter
               .to("microprofile-metrics:meter:redelivery?mark=2")
               // Register the 'error' meter
               .to("microprofile-metrics:meter:error"); 

   2

   
   
           from("timer:stream?period=1000")
               .routeId("unreliable-service")
               .setBody(header(Exchange.TIMER_COUNTER).prepend("event #"))
               .log("Processing ${body}...")
               // Register the 'generated' meter
               .to("microprofile-metrics:meter:generated") 

   3

   
               // Register the 'attempt' meter via @Metered in Service.java
               .bean("service") 

   4

   
               .filter(header(Exchange.REDELIVERED))
                   .log(LoggingLevel.WARN, "Processed ${body} after ${header.CamelRedeliveryCounter} retries")
                   .setHeader(MicroProfileMetricsConstants.HEADER_METER_MARK, header(Exchange.REDELIVERY_COUNTER))
                   // Register the 'redelivery' meter
                   .to("microprofile-metrics:meter:redelivery") 

   5

   
               .end()
               .log("Successfully processed ${body}")
               // Register the 'success' meter
               .to("microprofile-metrics:meter:success"); 

   6

   
       }
   }

   Copy to Clipboard Toggle word wrap

   1

   Camel K モードラインを使用して、Prometheus トレイトと Maven 依存関係を自動的に設定します。

   2

   error: 処理されていないイベントの数に対応するエラー数のメトリック。

   3

   generated: 処理されるイベント数のメトリクス。

   4

   attempt: 受信イベントを処理するためにサービス Bean に対して実行された呼び出し数のメトリクス。

   5

   redelivery: イベントを処理するために行われた再試行回数のメトリクス。

   6

   success: 正常に処理されたイベント数のメトリクス。

2. 必要に応じて Camel MicroProfile Metrics アノテーションを実装ファイルに追加します。以下の例は、ランダムに失敗を生成する Camel K インテグレーションによって呼び出される service Bean を示しています。

   package com.redhat.integration;
   
   import java.util.Random;
   
   import org.apache.camel.Exchange;
   import org.apache.camel.RuntimeExchangeException;
   
   import org.eclipse.microprofile.metrics.Meter;
   import org.eclipse.microprofile.metrics.annotation.Metered;
   import org.eclipse.microprofile.metrics.annotation.Metric;
   
   import javax.inject.Named;
   import javax.enterprise.context.ApplicationScoped;
   
   @Named("service")
   @ApplicationScoped
   @io.quarkus.arc.Unremovable
   
   public class Service {
   
      //Register the attempt meter
      @Metered(absolute = true)
      public void attempt(Exchange exchange) { 

   1

   
         Random rand = new Random();
            if (rand.nextDouble() < 0.5) {
                throw new RuntimeExchangeException("Random failure", exchange); 

   2

   
         }
      }
    }

   Copy to Clipboard Toggle word wrap

   1

   @Metered MicroProfile Metrics アノテーションは meter を宣言し、名前はメトリクスメソッド名 (この場合は attempt) に基づいて自動的に生成されます。

   2

   この例は、メトリクスのエラーを生成するために無作為に失敗します。

3. 「Configuring Camel K integration metrics」の手順に従い、インテグレーションを実行し、Prometheus でカスタム Camel K メトリクスを確認します。

   この例では、Metrics.java ですでに Camel K モードラインが使用され、Prometheus と Service.java に必要な Maven 依存関係が自動的に設定されます。

手順

1. Operator メトリクスエンドポイントを対象とする PodMonitor リソースを作成し、Prometheus サーバーが Operator によって公開されるメトリクスをスクレープできるようにします。

   operator-pod-monitor.yaml

   apiVersion: monitoring.coreos.com/v1
   kind: PodMonitor
   metadata:
     name: camel-k-operator
     labels:
       app: "camel-k"
       camel.apache.org/component: operator
   spec:
     selector:
       matchLabels:
         app: "camel-k"
         camel.apache.org/component: operator
     podMetricsEndpoints:
       - port: metrics

   Copy to Clipboard Toggle word wrap

2. PodMonitor リソースを作成します。

   oc apply -f operator-pod-monitor.yaml

   Copy to Clipboard Toggle word wrap

手順

1. クラスター管理者アカウントで OpenShift Web コンソールにログインします。

2. Kafka と kamelets アプリケーションの接続に使用する OpenShift プロジェクトを作成します。

   1. Home > Projects を選択します。
   2. Create Project をクリックします。
   3. プロジェクトの名前 (例: my-camel-k-kafka) を入力し、続いて Create をクリックします。
3. 「Installing Camel K」の説明に従って、Camel K Operatorおよび Camel L CLI をインストールします。
4. Red Hat Integration - Camel K Operator がインストールされていることを確認するには、Operators > Installed Operators の順にクリックします。

1. コマンドラインから OpenShift クラスターにログインします。

2. 以下のように、kamelets で Kafka に接続するプロジェクトを開きます。

   oc project my-camel-k-kafka

   Copy to Clipboard Toggle word wrap

3. 必要な Operator がプロジェクトにインストールされていることを確認します。

   oc get csv

   Copy to Clipboard Toggle word wrap

   結果には、Red Hat Camel K および RHOAS Operator が表示され、それらが Succeeded フェーズにあることを示します。

4. Kafka インスタンスを準備し、RHOAS に接続します。

   1. 以下のコマンドを使用して RHOAS CLI にログインします。

      rhoas login

      Copy to Clipboard Toggle word wrap

   2. kafka-test などの kafka インスタンスを作成します。

      rhoas kafka create kafka-test

      Copy to Clipboard Toggle word wrap

      Kafka インスタンスを作成するプロセスの完了に数分かかる場合があります。

5. Kafka インスタンスのステータスを確認するには、以下を実行します。

   rhoas status

   Copy to Clipboard Toggle word wrap

   Web コンソール https://cloud.redhat.com/beta/application-services/streams/kafkas/ でステータスを表示することもできます。ステータスが ready の場合は、次の手順に進みます。

6. 新しい Kafka トピックを作成します。

   rhoas kafka topic create test-topic

   Copy to Clipboard Toggle word wrap

手順

1. 次のコマンドを入力して、サービスアカウントを作成します。この例では、my-service-acct という名前のサービスアカウントを作成し、クレデンシャルを JSON ファイルに保存します。

   rhoas service-account create --name "my-service-acct" --file-format json
   Service account my-service-account created
   Credentials saved to /home/developer/my-project/credentials.json

   Copy to Clipboard Toggle word wrap

   サービスアカウントのクレデンシャルが作成され、JSON ファイルに保存されます。サービスアカウントの作成時に、ファイル形式および場所を選択してクレデンシャルを保存できます。

2. サービスアカウントのクレデンシャルを確認するには、作成した credentials.json ファイルを表示します。

   cat credentials.json
   {
   "user":"srvc-acct-eb575691-b94a-41f1-ab97-50ade0cd1094",
   "password":"facf3df1-3c8d-4253-aa87-8c95ca5e1225"
   }

   Copy to Clipboard Toggle word wrap

   これらのクレデンシャルとブートストラップサーバーの URL を使用して、アプリケーションやサービスを Kafka インスタンスに接続します。

3. Kafka ブローカーの URL (ブートストラップ URL) を取得します。

   rhoas status kafka

   Copy to Clipboard Toggle word wrap

   コマンドは、以下のような出力を返します。

     Kafka
     ---------------------------------------------------------------
     ID:                     1ptdfZRHmLKwqW6A3YKM2MawgDh
     Name:                   my-kafka
     Status:                 ready
     Bootstrap URL:        my-kafka--ptdfzrhmlkwqw-a-ykm-mawgdh.kafka.devshift.org:443

   Copy to Clipboard Toggle word wrap

手順

1. application.properties ファイルを編集し、Kafka クレデンシャルを追加します。

   application.properties ファイル

   camel.component.kafka.brokers = <YOUR-KAFKA-BOOTSTRAP-URL-HERE>
   camel.component.kafka.security-protocol = SASL_SSL
   camel.component.kafka.sasl-mechanism = PLAIN
   camel.component.kafka.sasl-jaas-config=org.apache.kafka.common.security.plain.PlainLoginModule required username='<YOUR-SERVICE-ACCOUNT-ID-HERE>' password='<YOUR-SERVICE-ACCOUNT-SECRET-HERE>';
   consumer.topic=<TOPIC-NAME>
   producer.topic=<TOPIC-NAME>

   Copy to Clipboard Toggle word wrap

2. 以下のコマンドを実行して、application.properties ファイルに機密プロパティーが含まれるシークレットを作成します。

   oc create secret generic kafka-props --from-file application.properties

   Copy to Clipboard Toggle word wrap

   Camel K インテグレーションの実行時に、このシークレットを使用します。

手順

1. application-oauth.properties ファイルを編集し、Kafka クレデンシャルを追加します。

   application-oauth.properties ファイル

   camel.component.kafka.brokers = <YOUR-KAFKA-BOOTSTRAP-URL-HERE>
   camel.component.kafka.security-protocol = SASL_SSL
   camel.component.kafka.sasl-mechanism = OAUTHBEARER
   camel.component.kafka.sasl-jaas-config = org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
   oauth.client.id='<YOUR-SERVICE-ACCOUNT-ID-HERE>' \
   oauth.client.secret='<YOUR-SERVICE-ACCOUNT-SECRET-HERE>' \
   oauth.token.endpoint.uri="https://identity.api.openshift.com/auth/realms/rhoas/protocol/openid-connect/token" ;
   consumer.topic=<TOPIC-NAME>
   producer.topic=<TOPIC-NAME>

   Copy to Clipboard Toggle word wrap

2. 以下のコマンドを実行して、application.properties ファイルに機密プロパティーが含まれるシークレットを作成します。

   oc create secret generic kafka-props --from-file application-oauth.properties

   Copy to Clipboard Toggle word wrap

   Camel K インテグレーションの実行時に、このシークレットを使用します。

プロデューサーインテグレーションの実行

1. サンプルプロデューサーインテグレーションを作成します。これにより、トピックには 10 秒ごとにメッセージが表示されます。

   Sample SaslSSLKafkaProducer.java

   // kamel run --secret kafka-props SaslSSLKafkaProducer.java --dev
   // camel-k: language=java dependency=mvn:org.apache.camel.quarkus:camel-quarkus-kafka dependency=mvn:io.strimzi:kafka-oauth-client:0.7.1.redhat-00003
   
   import org.apache.camel.builder.RouteBuilder;
   import org.apache.camel.component.kafka.KafkaConstants;
   
   public class SaslSSLKafkaProducer extends RouteBuilder {
     @Override
     public void configure() throws Exception {
     log.info("About to start route: Timer -> Kafka ");
     from("timer:foo")
       .routeId("FromTimer2Kafka")
       .setBody()
         .simple("Message #${exchangeProperty.CamelTimerCounter}")
       .to("kafka:{{producer.topic}}")
       .log("Message correctly sent to the topic!");
     }
   }

   Copy to Clipboard Toggle word wrap

2. その後、プロデューサーインテグレーションを実行します。

   kamel run --secret kafka-props SaslSSLKafkaProducer.java --dev

   Copy to Clipboard Toggle word wrap

   プロデューサーは新しいメッセージを作成し、トピックにプッシュし、一部の情報をログに記録します。

   [2] 2021-05-06 08:48:11,854 INFO  [FromTimer2Kafka] (Camel (camel-1) thread #1 - KafkaProducer[test]) Message correctly sent to the topic!
   [2] 2021-05-06 08:48:11,854 INFO  [FromTimer2Kafka] (Camel (camel-1) thread #3 - KafkaProducer[test]) Message correctly sent to the topic!
   [2] 2021-05-06 08:48:11,973 INFO  [FromTimer2Kafka] (Camel (camel-1) thread #5 - KafkaProducer[test]) Message correctly sent to the topic!
   [2] 2021-05-06 08:48:12,970 INFO  [FromTimer2Kafka] (Camel (camel-1) thread #7 - KafkaProducer[test]) Message correctly sent to the topic!
   [2] 2021-05-06 08:48:13,970 INFO  [FromTimer2Kafka] (Camel (camel-1) thread #9 - KafkaProducer[test]) Message correctly sent to the topic!

   Copy to Clipboard Toggle word wrap

コンシューマーインテグレーションの実行

1. コンシューマーインテグレーションを作成します。

   Sample SaslSSLKafkaProducer.java

   // kamel run --secret kafka-props SaslSSLKafkaConsumer.java --dev
   // camel-k: language=java dependency=mvn:org.apache.camel.quarkus:camel-quarkus-kafka dependency=mvn:io.strimzi:kafka-oauth-client:0.7.1.redhat-00003
   
   import org.apache.camel.builder.RouteBuilder;
   
   public class SaslSSLKafkaConsumer extends RouteBuilder {
     @Override
     public void configure() throws Exception {
   	log.info("About to start route: Kafka -> Log ");
   	from("kafka:{{consumer.topic}}")
       .routeId("FromKafka2Log")
       .log("${body}");
     }
   }

   Copy to Clipboard Toggle word wrap

2. 別のシェルを開き、コマンドを使用してコンシューマーインテグレーションを実行します。

   kamel run --secret kafka-props SaslSSLKafkaConsumer.java --dev

   Copy to Clipboard Toggle word wrap

   コンシューマーは、トピックにあるイベントのロギングを開始します。

   [1] 2021-05-06 08:51:08,991 INFO  [FromKafka2Log] (Camel (camel-1) thread #0 - KafkaConsumer[test]) Message #8
   [1] 2021-05-06 08:51:10,065 INFO  [FromKafka2Log] (Camel (camel-1) thread #0 - KafkaConsumer[test]) Message #9
   [1] 2021-05-06 08:51:10,991 INFO  [FromKafka2Log] (Camel (camel-1) thread #0 - KafkaConsumer[test]) Message #10
   [1] 2021-05-06 08:51:11,991 INFO  [FromKafka2Log] (Camel (camel-1) thread #0 - KafkaConsumer[test]) Message #11

   Copy to Clipboard Toggle word wrap