Event-based microservices with Red Hat AMQ Streams

As part of Red Hat's AMQ offerings, Red Hat offers a Kafka-based event streaming solution both for traditional deployment and microservices-based deployment branded as Red Hat AMQ Streams. The Red Hat OpenShift AMQ Streams deployment option is based on Strimzi, an open source tool that makes Kafka deployment as a container on a Kubernetes platform easy because most of the deployment prerequisites are automated with the OpenShift Operator Framework.

In this article, we look at how to deploy Apache Kafka on Red Hat OpenShift 4, using reasonable sample microservice applications to showcase the endless possibility of innovation brought by OpenShift and Kafka.

Figure 1: Our deployment environment.

Figure 1 illustrates the environment we will deploy on OpenShift with a number of microservices, including:

• The Account Balance Service provides information on the account balance, which has its own MongoDB database services.
• The Credit Service performs credit transfer between accounts, storing the credit data in the credit Kafka topic.
• The Event Correlator listens to the credit topic, performs the necessary adjustment to the account balance, and updates the changes to the Account Balance Service via the REST API. At the same time, it sends the outcome of this process as a credit response to the Kafka topic credit-response.
• MongoDB Kafka Connect listens to the content in the credit-response topic and streams this information to the Credit Response DB.

Install AMQ Streams on Red Hat OpenShift

AMQ Streams installation is pretty straightforward on Red Hat OpenShift. The only issue I faced was configuring MongoDB Kafka Connect, and that was mostly due to a lack of detailed documentation and a bug in Kafka Connect. All of these issues are now structurally documented as the following:

1. Download the YAML installation files from the Red Hat Access website.

Note: We are installing these AMQ Streams using cluster admin. AMQ Streams includes several custom resources. By default, permission to create, edit, and delete these resources is limited to OpenShift cluster administrators. If you want to allow non-cluster administrators to manage AMQ Streams resources, you must assign them the Strimzi Administrator role.

2. Deploy the Kafka cluster using the Kafka Operator, which can watch Kafka resources for single and multiple namespaces. In our case, we deploy the Operator to watch for a single namespace. Once you download and unzip the installation files, navigate to the root folder, which contains two folders: examples and install.
3. Run the following from the command line to make changes to the provided YAML files for our single OpenShift namespace deployment:

On Linux:

$ sed -i 's/namespace: .*/namespace: my-kafka-example/' install/cluster-operator/*RoleBinding*.yaml

On macOS:

$ sed -i '' 's/namespace: .*/namespace: my-kafka-example/' install/cluster-operator/*RoleBinding*.yaml

4. Run the following command to deploy the Operator once the namespace is changed:

$ oc apply -f install/cluster-operator -n my-kafka-example

Note: You can also configure the Kafka Operator to watch for all namespaces. Please refer to the documentation for details.

5. Deploy the Kafka cluster once the Kafka Operator is deployed. There are two options: ephemeral and persistent. We will deploy a persistent Kafka cluster. Begin by opening the examples/kafka/kafka-persistent.yaml file and changing the Kafka cluster name in the Kafka.metadata.name property as follows:

apiVersion: kafka.strimzi.io/v1beta1
kind: Kafka
metadata:
  name: my-kafka-cluster
# ...

entityOperator:
    topicOperator:
      watchedNamespace: my-kafka-example
      reconciliationIntervalSeconds: 90
      zookeeperSessionTimeoutSeconds: 20
      topicMetadataMaxAttempts: 6
      image: registry.redhat.io/amq7/amq-streams-operator:1.3.0

oc apply -f examples/kafka/kafka-persistent.yaml

oc run kafka-producer -ti --image=registry.redhat.io/amq7/amq-streams-kafka-23:1.3.0 --rm=true --restart=Never -- bin/kafka-console-producer.sh --broker-list kafka-cluster-kafka-bootstrap:9092 --topic my-topic

oc run kafka-consumer -ti --image=registry.redhat.io/amq7/amq-streams-kafka-23:1.3.0 --rm=true --restart=Never -- bin/kafka-console-consumer.sh --bootstrap-server kafka-cluster-kafka-bootstrap:9092 --topic my-topic --from-beginning

oc new-app -f https://raw.githubusercontent.com/chengkuangan/creditresponsemongodb/master/mongodb-deployment-template.yaml

2019-11-07 12:23:07,617 INFO Cluster created with settings {hosts=[creditresponse:27017], mode=SINGLE, requiredClusterType=UNKNOWN, serverSelectionTimeout='30000 ms', maxWaitQueueSize=500} (org.mongodb.driver.cluster) [task-thread-mongodb-sink-0]
2019-11-07 12:23:07,842 INFO Cluster description not yet available. Waiting for 30000 ms before timing out (org.mongodb.driver.cluster) [task-thread-mongodb-sink-0]
2019-11-07 12:23:07,861 INFO Opened connection [connectionId{localValue:1, serverValue:220}] to creditresponse:27017 (org.mongodb.driver.connection) [cluster-ClusterId{value='5dc40cab2efd9074c7742e33', description='null'}-creditresponse:27017]
2019-11-07 12:23:07,863 INFO Monitor thread successfully connected to server with description ServerDescription{address=creditresponse:27017, type=STANDALONE, state=CONNECTED, ok=true, version=ServerVersion{versionList=[3, 2, 10]}, minWireVersion=0, maxWireVersion=4, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=null, roundTripTimeNanos=1014605} (org.mongodb.driver.cluster) [cluster-ClusterId{value='5dc40cab2efd9074c7742e33', description='null'}-creditresponse:27017]
2019-11-07 12:23:07,892 INFO Closed connection [connectionId{localValue:2}] to creditresponse:27017 because there was a socket exception raised by this connection. (org.mongodb.driver.connection) [task-thread-mongodb-sink-0]
2019-11-07 12:23:07,894 ERROR Error on mongodb operation (com.mongodb.kafka.connect.sink.MongoSinkTask) [task-thread-mongodb-sink-0]
com.mongodb.MongoSecurityException: Exception authenticating MongoCredential{mechanism=SCRAM-SHA-1, userName='creditresponse', source='admin', password=<hidden>, mechanismProperties={}}
  at com.mongodb.internal.connection.SaslAuthenticator.wrapException(SaslAuthenticator.java:173)

mongo --port 27017 -u admin -p creditresponse --authenticationDatabase admin
use admin
db.runCommand({createRole:"listDatabases",privileges:[{resource:{cluster:true}, actions:["listDatabases"]}],roles:[]})
db.createUser({
"user" : "creditresponse",
"pwd" : "creditresponse",
    "roles" : [
        {
            "role" : "listDatabases",
            "db" : "admin"
        },
        {
            "role" : "readWrite",
            "db" : "creditresponse"
        },
        {
            "role" : "read",
            "db" : "local"
        }
    ]
})

[source]
FROM registry.redhat.io/amq7/amq-streams-kafka-23:1.3.0
USER root:root
COPY ./mongo-plugins/ /opt/kafka/plugins/
USER kafka:kafka

docker login registry.redhat.io

docker build -t chengkuan/amq-streams-kafka-connect-23:1.3.0

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaConnect
metadata:
  name: mongodb-connect-cluster
spec:
  version: 2.3.0
  replicas: 1
  bootstrapServers: kafka-cluster-kafka-bootstrap:9093
  tls:
    trustedCertificates:
      - secretName: kafka-cluster-cluster-ca-cert
        certificate: ca.crt
  image: docker.io/chengkuan/amq-streams-kafka-connect-23:1.3.0

oc apply -f examples/kafka-connect/kafka-connect.yaml

oc port-forward service/mongodb-connect-cluster-connect-api 8083:8083

curl http://localhost:8083/connector-plugins

[{"class":"com.mongodb.kafka.connect.MongoSinkConnector","type":"sink","version":"0.2"},{"class":"com.mongodb.kafka.connect.MongoSourceConnector","type":"source","version":"0.2"},{"class":"org.apache.kafka.connect.file.FileStreamSinkConnector","type":"sink","version":"2.3.0.redhat-00003"},{"class":"org.apache.kafka.connect.file.FileStreamSourceConnector","type":"source","version":"2.3.0.redhat-00003"}]

{
    "name": "mongodb-sink",
    "config": {
        "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
        "tasks.max": 1,
        "topics": "credit-response",
        "connection.uri": "mongodb://creditresponse:creditresponse@creditresponse:27017",
        "database": "creditresponse",
        "collection": "response",
        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
        "key.converter.schemas.enable": false,
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter.schemas.enable": false,
        "max.num.retries": 3
    }
}

curl -d connect-mongodb-sink.json -H "Content-Type: application/json" -X POST http://localhost:8083/connectors

curl http://localhost:8083/connectors/mongodb-sink

The result:

{
    "name": "mongodb-sink",
    "config": {
        "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
        "key.converter.schemas.enable": "false",
        "database": "creditresponse",
        "tasks.max": "1",
        "topics": "credit-response",
        "max.num.retries": "3",
        "connection.uri": "mongodb://creditresponse:creditresponse@creditresponse:27017",
        "value.converter.schemas.enable": "false",
        "name": "mongodb-sink",
        "collection": "response",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "key.converter": "org.apache.kafka.connect.json.JsonConverter"
    },
    "tasks": [
        {
            "connector": "mongodb-sink",
            "task": 0
        }
    ],
    "type": "sink"
}

Deploy the Account Balance Service application

Run the following command to deploy the pods:

oc new-app https://raw.githubusercontent.com/chengkuangan/accountbalance/master/templates/deployment-templates.yaml

Deploy the Credit Service application

Run the following command to deploy the Credit Service. Ensure that KAFKA_BOOTSTRAP_SERVER points to the correct server:

oc new-app https://raw.githubusercontent.com/chengkuangan/creditservice/master/templates/creditservice.json -p KAFKA_BOOTSTRAP_SERVER=kafka-cluster-kafka-bootstrap:9092

Deploy the Event Correlator Service

oc new-app -f https://raw.githubusercontent.com/chengkuangan/eventcorrelator/master/templates/deployment-templates.yaml -p KAFKA_BOOTSTRAP_SERVER=kafka-cluster-kafka-bootstrap:9092

Watch the application in action

curl -H "Content-Type: application/json" -X POST http://accountbalance-my-kafka-example.apps.demo.ocp.internal/ws/pg/balance -d '{"accountId": "20191108-MY-00000001", "balance": 500.00, "lastUpdatedDate": 1563178274158 }'
curl -H "Content-Type: application/json" -X POST http://accountbalance-my-kafka-example.apps.demo.ocp.internal/ws/pg/balance -d '{"accountId": "20191108-MY-00000002", "balance": 700.00, "lastUpdatedDate": 1563178274158 }'

Query the account balance entries created above:

curl http://accountbalance-my-kafka-example.apps.demo.ocp.internal/ws/pg/balance/all

The result:

[
   {
      "_id":"5dc52069a3c1080001ebd539",
      "accountId":"20191108-MY-00000001",
      "balance":500.0,
      "lastUpdatedDate":1563178274158
   },
   {
      "_id":"5dc52076a3c1080001ebd53a",
      "accountId":"20191108-MY-00000002",
      "balance":700.0,
      "lastUpdatedDate":1563178274158
   }
]

Perform a credit transfer:

curl -H "Content-Type: application/json" -X POST http://creditservice-my-kafka-example.apps.demo.ocp.internal/ws/pg/credits -d '{"amount": 10.50, "sourceAccount": "20191108-MY-00000001", "targetAccount": "20191108-MY-00000002"}'

Query the balance after the credit transfer:

curl http://accountbalance-my-kafka-example.apps.demo.ocp.internal/ws/pg/balance/all

The result:

[
   {
      "_id":"5dc52069a3c1080001ebd539",
      "accountId":"20191108-MY-00000001",
      "balance":489.5,"lastUpdatedDate":1573200543543
   },
   {
      "_id":"5dc52076a3c1080001ebd53a",
      "accountId":"20191108-MY-00000002",
      "balance":710.5,
      "lastUpdatedDate":1573200543543
   }
]

Perform oc rsh into the Credit Response MongoDB. Use db.response.find() to see that the credit response is captured:

mongo --port 27017 -u admin -p creditresponse --authenticationDatabase admin

>use creditresponse
switched to db creditresponse

>show collections
response

>db.response.find()

Result:

{ 
   "_id" : ObjectId("5dc523f536d41402601d01a4"), 
   "sourceAccountRecordId" : "5dc52069a3c1080001ebd539", 
   "targetAccountRecordId" : "5dc52076a3c1080001ebd53a", 
   "sourceAccountId" : "20191108-MY-00000001", 
   "targetAccountId" : "20191108-MY-00000002", 
   "sourceAccountBalance" : 489.5, 
   "targetAccountBalance" : 710.5, 
   "creditRecordId" : "ykvlkqk2puzc5u", 
   "creditAmount" : 10.5, 
   "transactionDate" : NumberLong("1573200543543") 
}

References

• Red Hat AMQ Streams on OpenShift documentation
• Apache Kafka Connect documentation