AMQ Streams With Kafka Connect on Openshift

Have you ever heard about AMQ Streams? It is a Kafka Platform based on Apache Kafka and powered by Red Hat. A big advantage of AMQ Streams is that as all Red Hat tools, it is open source. 

This article's about a common feature of Apache Kafka, called Kafka Connect. I'll explain how it works with OpenShift.

What Is Kafka Connect?

Kafka Connect is an open-source component of Apache Kafka®. It is a framework for connecting Kafka with external systems, such as databases, key-value stores, search indexes, and file systems. The Telegram platform is one of these systems and in this article, I will demonstrate how to use Kafka Connect deployed on OpenShift to get data from Telegram.

Kafka Connect API Architecture

Basically there are 2 types of Kafka Connectors: 

Sources:

A source connector collects data from a system and sends it to a Kafka topic, for example, we can get messages from Telegram and put it in a  Kafka topic. 

Synk:

A sink connector delivers data from Kafka topics into other systems, which might be indexed such as Elasticsearch, batch systems such as Hadoop, or any kind of database.

There are two ways to execute Kafka Connect: Standalone and Distributed. However, the operator supports only Distributed mode in OpenShift. That's not a restriction, because none Kafka distribution in production environments is recommended set standalone mode.

The Kafka Connector uses an environment independent of Kafka Broker, on OpenShift Kafka Connect API runs in a separated pod.  

In this example, we will use a source connector.

Before We Start

Firstly will be necessary to create a bot in the Telegram platform and generate an access token to him.  We need this token to generate a secret on OpenShift to Kafka Connect can receive messages from Telegram. To create this bot, see the article — Bots: An introduction for developers.

The necessary files to execute this demonstration is on GitHub. Click here to start the download. The files have a number that identifies the sequence of use of them in this article. Any person can download and reproduce this demonstration.

Let's Start

The first step is to deploy AMQ Streams on OpenShit. For this, use the Strimzi operator provided on the Red Hat Web Site. The link of the file to download is "Red Hat AMQ Streams 1.4.1 OpenShift Installation and Example Files".

If you prefer, I generated a package with the operator also available on GitHub.

Use the commands below to install AMQ Stream (Kafka) on OpenShift. 

$ oc new-project kafkaproject

$ oc apply -f ./operator/ 

$ oc apply -f ./kafka.yaml

What have we done? Firstly, we have created a project named 'kafkaproject' and deploy the Strimzi operator. When the operator was finished we deploy AMQ Streams through the kafka.yaml template.

The result of these commands is bellow,

The next step is to generate the Openshift secret using Telegram access token. Create the property file, see the example below. I chose 'telegram-credentials.properties' filename. Feel free to choose your filename.

xxxxxxxxxx

token=1011765240:AAG-PdMojD1pZWUpNySa8rHDjxym3CVyqTxd

Create the file telegram-credentials.yaml and put the base64 representation of file telegram-credentials.properties. 

xxxxxxxxxx

apiVersion: v1

data:

  telegram-credentials.properties: ##BASE64 telegram-credentials.properties##

kind: Secret

metadata:

  name: telegram-credentials

  namespace: kafkaproject

type: Opaque

Use the below commands to create the secret and deploy the Kafka Connect. Making a reference with RHEL version, the kafka-connect.yaml is similar to execute connect-distributed.sh

xxxxxxxxxx

 $ oc apply -f ./4-telegram-credentials.yaml 

 $ oc apply -f ./5-kafka-connect.yaml

Notice that a new pod was created on OpenShift.

After deploying Kafka Connect, we need to inform the parameters to connect to the Telegram. For this, we use the file 6-telegram-connector.yaml. See below:

x

apiVersion: kafka.strimzi.io/v1alpha1

kind: KafkaConnector

metadata:

  name: telegram-connector

  labels:

    strimzi.io/cluster: my-connect-cluster

spec:

  class: org.apache.camel.kafkaconnector.CamelSourceConnector

  tasksMax: 1

  config:

    key.converter: org.apache.kafka.connect.storage.StringConverter

    value.converter: org.apache.kafka.connect.storage.StringConverter

    camel.source.kafka.topic: telegram-topic

    camel.source.url: telegram:bots

    camel.component.telegram.authorizationToken: ${file:/opt/kafka/external-configuration/telegram-credentials/telegram-credentials.properties:token}    

Execute the command below:

xxxxxxxxxx

  oc apply -f 6-telegram-connector.yaml 

After completion, and if the credentials are correct, any message sent to your bot will be received by Kafka Connect and saved in a topic called telegram-topic, which was defined in property camel.source.kafka.topic in the file 6-telegram-connector.yaml. Notice how the API token is mounted from a secret in the property camel.component.telegram.authorizationToken.

You can go to the Telegram app and talk with the bot, in this example I create a bot called @amqstreams. Run the receiver on a Kafka topic telegram-topic to see the messages sent to the bot.

To consume the messages on the topic,  we can use the command below:

xxxxxxxxxx

kubectl run kafka-consumer -ti --image=strimzi/kafka:0.16.0-kafka-2.4.0 \

--rm=true --restart=Never \

-- bin/kafka-console-consumer.sh \

--bootstrap-server my-cluster-kafka-bootstrap:9092 \

--topic telegram-topic \

--from-beginning

A good point to notice here is the property camel.source.url. Internally, when we provide the value , telegram-bots, to property, the connector created a camel route using a Camel Telegram component.

There are a lot of connectors available for Kafka Connet, in our hands-on, we used a Camel Connector, this one allows us to use all power of Apache Camel together Kafka Connect. In this example, we didn't write any code. However, if necessary, we can create customized routes to connect to anything and use the same steps used here to put dada on the Apache Kafka.

Final Considerations

The Kafka Connect is a powerful member of the Kafka Ecosystem. In this article I showed a simple integration, however, Kafka Connect allows us to interact with a lot of systems or even other message brokers.

Guard this advice: “If there is data, we can connect and get it.“