{{announcement.body}}
{{announcement.title}}

AMQ Streams With Kafka Connect on Openshift

DZone 's Guide to

AMQ Streams With Kafka Connect on Openshift

Receiving messages from Telegram with Kafka Connect

· Big Data Zone ·
Free Resource

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 ConnectI'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.

Sources and Synks workflow

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".

Download AMQ Streams

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. 

PowerShell


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,

Resulting clusters

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.

Plain Text
 




xxxxxxxxxx
1


 
1
token=1011765240:AAG-PdMojD1pZWUpNySa8rHDjxym3CVyqTxd



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

Plain Text
 




xxxxxxxxxx
1


 
1
apiVersion: v1
2
data:
3
  telegram-credentials.properties: ##BASE64 telegram-credentials.properties##
4
kind: Secret
5
metadata:
6
  name: telegram-credentials
7
  namespace: kafkaproject
8
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

Shell
 




xxxxxxxxxx
1


 
1
 $ oc apply -f ./4-telegram-credentials.yaml 
2
 $ oc apply -f ./5-kafka-connect.yaml



Notice that a new pod was created on OpenShift.

New pod 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:

Plain Text
 




x


 
1
apiVersion: kafka.strimzi.io/v1alpha1
2
kind: KafkaConnector
3
metadata:
4
  name: telegram-connector
5
  labels:
6
    strimzi.io/cluster: my-connect-cluster
7
spec:
8
  class: org.apache.camel.kafkaconnector.CamelSourceConnector
9
  tasksMax: 1
10
  config:
11
    key.converter: org.apache.kafka.connect.storage.StringConverter
12
    value.converter: org.apache.kafka.connect.storage.StringConverter
13
    camel.source.kafka.topic: telegram-topic
14
    camel.source.url: telegram:bots
15
    camel.component.telegram.authorizationToken: ${file:/opt/kafka/external-configuration/telegram-credentials/telegram-credentials.properties:token}    



Execute the command below:

Shell
 




xxxxxxxxxx
1


 
1
  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:

Shell
 




xxxxxxxxxx
1


1
kubectl run kafka-consumer -ti --image=strimzi/kafka:0.16.0-kafka-2.4.0 \
2
--rm=true --restart=Never \
3
-- bin/kafka-console-consumer.sh \
4
--bootstrap-server my-cluster-kafka-bootstrap:9092 \
5
--topic telegram-topic \
6
--from-beginning



Hello world message


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.

Topics:
apache camel, apache kafka, kafka connectors, kafka streams, kubernetes, openshift, red hat middleware

Opinions expressed by DZone contributors are their own.

{{ parent.title || parent.header.title}}

{{ parent.tldr }}

{{ parent.urlSource.name }}