Chapter 5. Authenticating Camel K against Kafka You can authenticate Camel K against Apache Kafka. The following example demonstrates how to set up a Kafka Topic and use it in a simple Producer/Consumer pattern Integration. 5.1. Setting up Kafka To set up Kafka, you must: Install the required OpenShift operators Create a Kafka instance Create a Kafka topic Use the Red Hat product mentioned below to set up Kafka: Red Hat Advanced Message Queuing (AMQ) streams - A self-managed Apache Kafka offering. AMQ Streams is based on open source Strimzi and is included as part of Red Hat Integration . AMQ Streams is a distributed and scalable streaming platform based on Apache Kafka that includes a publish/subscribe messaging broker. Kafka Connect provides a framework to integrate Kafka-based systems with external systems. Using Kafka Connect, you can configure source and sink connectors to stream data from external systems into and out of a Kafka broker. 5.1.1. Setting up Kafka by using AMQ streams AMQ Streams simplifies the process of running Apache Kafka in an OpenShift cluster. 5.1.1.1. Preparing your OpenShift cluster for AMQ Streams To use Camel K or Kamelets and Red Hat AMQ Streams, you must install the following operators and tools: Red Hat Integration - AMQ Streams operator - Manages the communication between your Openshift Cluster and AMQ Streams for Apache Kafka instances. Red Hat Integration - Camel K operator - Installs and manages Camel K - a lightweight integration framework that runs natively in the cloud on OpenShift. Camel K CLI tool - Allows you to access all Camel K features. Prerequisites You are familiar with Apache Kafka concepts. You can access an OpenShift 4.6 (or later) cluster with the correct access level, the ability to create projects and install operators, and the ability to install the OpenShift and the Camel K CLI on your local system. You installed the OpenShift CLI tool ( oc ) so that you can interact with the OpenShift cluster at the command line. Procedure To set up Kafka by using AMQ Streams: Log in to your OpenShift cluster's web console. Create or open a project in which you plan to create your integration, for example my-camel-k-kafka . Install the Camel K operator and Camel K CLI as described in Installing Camel K . Install the AMQ streams operator: From any project, select Operators > OperatorHub . In the Filter by Keyword field, type AMQ Streams . Click the Red Hat Integration - AMQ Streams card and then click Install . The Install Operator page opens. Accept the defaults and then click Install . Select Operators > Installed Operators to verify that the Camel K and AMQ Streams operators are installed. steps Setting up a Kafka topic with AMQ Streams 5.1.1.2. Setting up a Kafka topic with AMQ Streams A Kafka topic provides a destination for the storage of data in a Kafka instance. You must set up a Kafka topic before you can send data to it. Prerequisites You can access an OpenShift cluster. You installed the Red Hat Integration - Camel K and Red Hat Integration - AMQ Streams operators as described in Preparing your OpenShift cluster . You installed the OpenShift CLI ( oc ) and the Camel K CLI ( kamel ). Procedure To set up a Kafka topic by using AMQ Streams: Log in to your OpenShift cluster's web console. Select Projects and then click the project in which you installed the Red Hat Integration - AMQ Streams operator. For example, click the my-camel-k-kafka project. Select Operators > Installed Operators and then click Red Hat Integration - AMQ Streams . Create a Kafka cluster: Under Kafka , click Create instance . Type a name for the cluster, for example kafka-test . Accept the other defaults and then click Create . The process to create the Kafka instance might take a few minutes to complete. When the status is ready, continue to the step. Create a Kafka topic: Select Operators > Installed Operators and then click Red Hat Integration - AMQ Streams . Under Kafka Topic , click Create Kafka Topic . Type a name for the topic, for example test-topic . Accept the other defaults and then click Create . 5.1.2. Creating a secret by using the SASL/Plain authentication method You can create a secret with the credentials that you obtained (Kafka bootstrap URL, service account ID, and service account secret). Procedure Edit the application.properties file and add the Kafka credentials. application.properties file Run the following command to create a secret that contains the sensitive properties in the application.properties file: You use this secret when you run a Camel K integration. See Also The Camel K Kafka Basic Quickstart 5.1.3. Creating a secret by using the SASL/OAUTHBearer authentication method You can create a secret with the credentials that you obtained (Kafka bootstrap URL, service account ID, and service account secret). Procedure Edit the application-oauth.properties file and add the Kafka credentials. application-oauth.properties file Run the following command to create a secret that contains the sensitive properties in the application.properties file: You use this secret when you run a Camel K integration. See Also The Camel K Kafka Basic Quickstart 5.2. Running a Kafka integration Running a producer integration Create a sample producer integration. This fills the topic with a message, every 10 seconds. Sample SaslSSLKafkaProducer.java Then run the procedure integration. The producer will create a new message and push into the topic and log some information. Running a consumer integration Create a consumer integration. Sample SaslSSLKafkaProducer.java Open another shell and run the consumer integration using the command: A consumer will start logging the events found in the Topic:

Commands:
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>
create secret generic kafka-props --from-file application.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" ; camel.component.kafka.additional-properties[sasl.login.callback.handler.class]=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler consumer.topic=<TOPIC-NAME> producer.topic=<TOPIC-NAME>
create secret generic kafka-props --from-file application-oauth.properties
// 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 #USD{exchangeProperty.CamelTimerCounter}") .to("kafka:{{producer.topic}}") .log("Message correctly sent to the topic!"); } }
kamel run --secret kafka-props SaslSSLKafkaProducer.java --dev
[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!
// 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("USD{body}"); } }
kamel run --secret kafka-props SaslSSLKafkaConsumer.java --dev
[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