Apache Kafka and Low-Code: An Introduction and a Practical Tutorial on Integrating IBM Event Streams and Watson IoT Platform with Joget

Apache Kafka and Low-Code: An Introduction and a Practical Tutorial on Integrating IBM Event Streams and Watson IoT Platform with Joget

This will be followed by a practical tutorial on using a visual low-code approach to rapidly develop an application integrated with Kafka for an Internet of Things (IoT) use case. In this tutorial, we’ll integrate the Joget low code platform with IBM Event Streams (a hosted Kafka service) and IBM Watson IoT Platform.

1. Part 2: Create IBM Event Streams Service – Step 1: Create an Event Streams service instance

   Now that we have a general understanding of Apache Kafka, let’s setup an actual Kafka cluster. For this tutorial we’ll use IBM Event Streams, a managed Kafka service for the enterprise provided by IBM Cloud.

    

   IBM Cloud provides a free tier which includes the Event Streams service, so you can get started for free. Once you have registered for an IBM Cloud account, you can create a new Events Streams service.

    

   Login to the IBM Cloud console, and select the Event Streams service in the Catalog.

    

   

    

   Choose the Lite plan, and click on Create.

    

2. Part 2: Create IBM Event Streams Service – Step 2: Create Topic

   Once the service has been created, click on Create a topic in the Getting started page.

    

   

    

   Enter a topic name (e.g. kafka-java-console-sample-topic) and for testing purposes, select 1 partition and a day for message retention.

   

   

   

   

    

3. Part 2: Create IBM Event Streams Service – Step 3: Create Credentials

   In the Service credentials page, click on New credential. Select the Manager role and Auto Generate the Service ID.

    

   

   

    

   The credentials are provided in JSON format. Copy the JSON, which contains connection information e.g. API Key, Broker URLs, etc) to be used later.

    

   {
     “api_key“: “p111111ZVt9Q8DQPluE6jYLYhmMmKLaKtXCjWkpSI6OZ”,
     “apikey”: “p111111ZVt9Q8DQPluE6jYLYhmMmKLaKtXCjWkpSI6OZ”,
     “iam_apikey_description”: “Auto-generated for key 8111111a-0a93-4a7a-9981-1500b1543e22”,
     “iam_apikey_name”: “Service credentials-1”,
     “iam_role_crn”: “crn:v1:bluemix:public:iam::::serviceRole:Manager”,
     “iam_serviceid_crn”: “crn:v1:bluemix:public:iam-identity::a/3111111935dc885883a45beb7fde7546::serviceid:ServiceId-11111114-12e4-4fa8-94b1-1f3bbc37ed09”,
     “instance_id”: “31111114-0ff1-42d7-b58b-a7f4fead3796”,
     “kafka_admin_url“: “https://t111111q7ljf7f4n.svc02.us-south.eventstreams.cloud.ibm.com”,
     “kafka_brokers_sasl“: [
       “broker-1-t111111q7ljf7f4n.kafka.svc02.us-south.eventstreams.cloud.ibm.com:9093”,
       “broker-2-t111111q7ljf7f4n.kafka.svc02.us-south.eventstreams.cloud.ibm.com:9093”,
       “broker-5-t111111q7ljf7f4n.kafka.svc02.us-south.eventstreams.cloud.ibm.com:9093”,
       “broker-3-t111111q7ljf7f4n.kafka.svc02.us-south.eventstreams.cloud.ibm.com:9093”,
       “broker-4-t11111q7ljf7f4n.kafka.svc02.us-south.eventstreams.cloud.ibm.com:9093”,
       “broker-0-t11111q7ljf7f4n.kafka.svc02.us-south.eventstreams.cloud.ibm.com:9093”
     ],
     “kafka_http_url”: “https://t11111q7ljf7f4n.svc02.us-south.eventstreams.cloud.ibm.com”,
     “password“: “p111111ZVt9Q8DQPluE6jYLYhmMmKLaKtXCjWkpSI6OZ”,
     “user“: “token”
   }

    

4. Part 2: Create IBM Event Streams Service – Step 4: Test the Service with a Sample Application

   Now that we have a running Kafka service, let’s test it with a sample application (consumer and producer) provided by IBM in GitHub. Clone and build the GitHub repository:

    

   $ git clone https://github.com/ibm-messaging/event-streams-samples.git

   $ cd event-streams-samples/kafka-java-console-sample

   $ gradle clean && gradle build

    

   Once the project has been built, let’s first run the consumer application in a command terminal. Replace the kafka_brokers_sasl and api_key values from the JSON obtained previously.

    

   $ java -jar ./build/libs/kafka-java-console-sample-2.0.jar <kafka_brokers_sasl> <api_key> -consumer

    

   The output will be as follows:

    

   [2020-01-10 16:18:07,383] INFO Using command line arguments to find credentials. (com.eventstreams.samples.EventStreamsConsoleSample)
   [2020-01-10 16:18:07,385] INFO Kafka Endpoints: broker-1-t111111q7ljf7f4n.kafka.svc02.us-south.eventstreams.cloud.ibm.com:9093 (com.eventstreams.samples.EventStreamsConsoleSample)
   [2020-01-10 16:18:08,522] INFO Creating the topic kafka-java-console-sample-topic (com.eventstreams.samples.EventStreamsConsoleSample)
   [2020-01-10 16:18:14,626] INFO Topic kafka-java-console-sample-topic already exists (com.eventstreams.samples.EventStreamsConsoleSample)
   [2020-01-10 16:18:17,297] INFO [Partition(topic = kafka-java-console-sample-topic, partition = 0, leader = 2, replicas = [2,3,4], isr = [2,3,4], offlineReplicas = [])] (com.eventstreams.samples.ConsumerRunnable)
   [2020-01-10 16:18:17,303] INFO EventStreamsConsoleSample will run until interrupted. (com.eventstreams.samples.EventStreamsConsoleSample)
   [2020-01-10 16:18:17,303] INFO class com.eventstreams.samples.ConsumerRunnable is starting. (com.eventstreams.samples.ConsumerRunnable)
   [2020-01-10 16:18:20,309] INFO No messages consumed (com.eventstreams.samples.ConsumerRunnable)

   Open another command terminal, and run the producer application:

    

   $ java -jar ./build/libs/kafka-java-console-sample-2.0.jar <kafka_brokers_sasl> <api_key> -producer
   

   The output will be as follows:

    

   [2020-01-10 16:19:21,655] INFO Using command line arguments to find credentials. (com.eventstreams.samples.EventStreamsConsoleSample)
   [2020-01-10 16:19:21,656] INFO Kafka Endpoints: broker-1-t335226q7ljf7f4n.kafka.svc02.us-south.eventstreams.cloud.ibm.com:9093 (com.eventstreams.samples.EventStreamsConsoleSample)
   [2020-01-10 16:19:22,192] INFO Creating the topic kafka-java-console-sample-topic (com.eventstreams.samples.EventStreamsConsoleSample)
   [2020-01-10 16:19:26,767] INFO Topic kafka-java-console-sample-topic already exists (com.eventstreams.samples.EventStreamsConsoleSample)
   [2020-01-10 16:19:29,016] INFO [Partition(topic = kafka-java-console-sample-topic, partition = 0, leader = 2, replicas = [2,3,4], isr = [2,3,4], offlineReplicas = [])] (com.eventstreams.samples.ProducerRunnable)
   [2020-01-10 16:19:29,017] INFO EventStreamsConsoleSample will run until interrupted. (com.eventstreams.samples.EventStreamsConsoleSample)
   [2020-01-10 16:19:29,017] INFO class com.eventstreams.samples.ProducerRunnable is starting. (com.eventstreams.samples.ProducerRunnable)
   [2020-01-10 16:19:32,321] INFO Message produced, offset: 0 (com.eventstreams.samples.ProducerRunnable)
   
   

   When you see the Message produced output, the consumer application should show the corresponding output when it receives the messages:

    

   headers = RecordHeaders(headers = [], isReadOnly = false), key = key, value = This is a test message #0) (com.eventstreams.samples.ConsumerRunnable)
   [2020-01-10 16:19:34,690] INFO Message consumed: ConsumerRecord(topic = kafka-java-console-sample-topic, partition = 0, leaderEpoch = 0, offset = 1, CreateTime = 1578644374326, serialized key size = 3, serialized value size = 25,
   
   

    

   At this point, we’ve confirmed that the Kafka cluster works well with consumer and producer applications. More detailed instructions are available in the official getting started tutorial.

5. Part 3: Design a Joget App with Kafka Plugin – Step 1: Design a New App

   Let’s now visually design an app using the Joget platform. Joget provides an extensible plugin architecture, so we developed sample Kafka producer and consumer plugins.

    

   Firstly, setup a Joget platform, either by downloading and installing on-premise, signing up for Joget Cloud, running on Docker, or other deployment options such as using the Joget certified operator on Red Hat OpenShift.

    

   NOTE: The sample Kafka plugins currently do not work in a multi-tenant Joget environment, so if you wish to run the plugin on Joget Cloud, it needs to be on a dedicated plan.

    

   In the Joget App Center, login as an administrator and click on Design New App. Key in an App ID and App Name and click on Save.

    

   

    

6. Part 3: Design a Joget App with Kafka Plugin – Step 2: Configure a Kafka Consumer Plugin

   First, ensure that the Kafka plugin has been uploaded to the platform. In Settings, select Manage Plugins, click on Upload Plugin, select the Kafka plugin file and click on Upload.

    

   Back in the app, select Properties & Export and click on Set Default Plugin Properties.

    

   

    

   Select the Kafka Consumer Audit Trail plugin, and key in the Kafka configuration. The Bootstrap Servers and API Key are obtained from the credentials JSON downloaded in the previous IBM Event Streams tutorial.

    

   

    

   In the BeanShell Script for Received Messages, you can enter a BeanShell script (the variables key and value are injected when a message is received). As an example, a simple script to output the message to the log file:

    

   import org.joget.commons.util.LogUtil;

   LogUtil.info(“KafkaConsumer”, “Message received ” + key + “=” + value);

    

   You can code more complex scripts according to your requirement, based on the BeanShell guide in the Joget Knowledge Base.

    

   Once the plugin configuration has been submitted, there will be a running background thread that is subscribed to the Kafka topic, with log output as below:

    

   INFO 17 Jan 2020 16:17:40 org.joget.plugin.base.PluginManager – Bundle org.joget.kafka.joget-kafka started
   INFO 17 Jan 2020 16:17:40 org.joget.kafka.ConsumerThreadMonitor – Waiting for platform init
   INFO 17 Jan 2020 16:17:40 org.joget.kafka.ConsumerThreadMonitor – Started thread monitor
   INFO 17 Jan 2020 16:17:45 org.joget.kafka.ConsumerThreadMonitor – Started Kafka consumer thread for app kafka
   INFO 17 Jan 2020 16:17:45 org.joget.kafka.ConsumerRunnable – Consumer is starting.
   INFO 17 Jan 2020 16:17:59 org.joget.kafka.ConsumerRunnable$1 – Partitions [kafka-java-console-sample-topic-0] assigned, consumer seeking to end.
   INFO 17 Jan 2020 16:18:02 org.joget.kafka.ConsumerRunnable$1 – current Position: 421
   INFO 17 Jan 2020 16:18:02 org.joget.kafka.ConsumerRunnable$1 – Seeking to end…
   INFO 17 Jan 2020 16:18:03 org.joget.kafka.ConsumerRunnable$1 – Seek from the current position: 421
   INFO 17 Jan 2020 16:18:03 org.joget.kafka.ConsumerRunnable$1 – Producer can now begin producing messages.
   
   

    

7. Part 3: Design a Joget App with Kafka Plugin – Step 3: Design Kafka Producer Process

   Let’s now create a simple producer process that sends a message to the Kafka cluster. Back in the Kafka app, select Processes and click on Design Process to launch the Process Builder. Add a Tool to the process, rename it appropriately and click on Deploy. Once the process has been deployed successfully, close the Process Builder.

    

   

   In the Processes page, select the Map Tools to Plugins tab, and click on Add/Edit Plugin.

   

   Select the Kafka Producer Tool, and key in the Kafka configuration. The Bootstrap Servers and API Key are obtained from the credentials JSON downloaded in the previous tutorial, and the Topic, Key and Value can be set according to the required message.

   

   Click on Run Process to run the process. When the tool is executed, you should see the log output indicating that the message has been sent to the Kafka cluster:

    

   INFO 17 Jan 2020 16:19:10 org.joget.kafka.ProducerRunnable – [Partition(topic = kafka-java-console-sample-topic, partition = 0, leader = 2, replicas = [2,3,4], isr = [4,3,2], offlineReplicas = [])]
   INFO 17 Jan 2020 16:19:10 org.joget.kafka.ProducerRunnable – class org.joget.kafka.ProducerRunnable is starting.
   INFO 17 Jan 2020 16:19:13 org.joget.kafka.ProducerRunnable – Message produced, offset: 421
   INFO 17 Jan 2020 16:19:13 org.joget.kafka.ProducerRunnable – class org.joget.kafka.ProducerRunnable is shutting down.
   INFO 17 Jan 2020 16:19:13 org.joget.kafka.ProducerRunnable – class org.joget.kafka.ProducerRunnable has shut down.
   
   

   The Kafka consumer plugin will receive the message and execute the configured BeanShell script. Here is the sample log output:

    

   INFO 17 Jan 2020 16:19:14 org.joget.kafka.ConsumerRunnable – Executing script. Offset: 421
   INFO 17 Jan 2020 16:19:14 org.joget.kafka.ConsumerRunnable – Key=key. Value=message
   INFO 17 Jan 2020 16:19:14 KafkaConsumer – Message received key=message
   INFO 17 Jan 2020 16:19:14 org.joget.kafka.ConsumerRunnable – Script execution complete.
   

   At this point, we have a Kafka producer and consumer running in a Joget app. You can enhance the app using the Form Builder and Userview Builder to add a message form and UI for the app, as the screenshots below demonstrate:

    

   

    

   

    

   

    

8. Part 4: Create IBM Watson IoT Platform Service – Step 1: Create IoT Platform Service

   IBM Watson IoT Platform is a managed cloud service to connect, control, monitor and analyze data from IoT devices using the MQTT protocol. As a part of IBM Cloud, it also has a free tier offering so you can get started for free.

    

   Login to the IBM Cloud console, and select the Internet of Things Platform Service in the Catalog.

    

   

   Choose the Lite plan, and click on Create. The service should be created within a few minutes.

   

    

9. Part 4: Create IBM Watson IoT Platform Service – Step 2: Add an IoT Device

   Now, let’s create a device in the service. Browse to Devices, and click on Create a device.

   

    

   Enter a Device Type, Device ID and any other optional information. You can use the defaults for all the other settings.

    

   

   

   

   

   Click on Finish, and the Device Credentials will be presented. Keep a copy of the credentials, especially the Authentication Token which is not recoverable.

   

   More detailed information can be found in the Getting started with Watson IoT Platform page.

    

10. Part 5: Integrate Watson IoT with Event Streams – Step 1: Generate API Key

    The Watson IoT platform provides data store connectors to connect it to other IBM Cloud services. Let’s setup a service binding and data store connector for Event Streams, so that device data will be automatically published to a Kafka topic.

     

    Browse to Apps, and click on Generate API Key. Set the Role to Standard Application and click on Generate Key.

    

    

    Make a note of the generated API Key and Authentication Token.

    

     

11. Part 5: Integrate Watson IoT with Event Streams – Step 2: Add Service Binding and Add Data Store Connector for Event Streams

    To set up a service binding between Watson IoT and Event Streams, you will need to use the Python SDK. On a machine with Python installed, install the SDK using the command:

    pip install wiotp-sdk

    Save the following Python code in a file bind.py (replace the credential values with your Event Streams credentials):

    import wiotp.sdk.application
    from wiotp.sdk.api.services import EventStreamsServiceBindingCredentials, EventStreamsServiceBindingCreateRequestoptions = wiotp.sdk.application.parseEnvVars()
    appClient = wiotp.sdk.application.ApplicationClient(options)serviceBinding = {
    “name”: “test-eventstreams”,
    “description”: “Test EventStreams instance”,
    “type”: “eventstreams”,
    “credentials”: {
    “api_key“: “p111111ZVt9Q8DQPluE6jYLYhmMmKLaKtXCjWkpSI6OZ”,
    “user“: “token”,
    “password“: “p111111ZVt9Q8DQPluE6jYLYhmMmKLaKtXCjWkpSI6OZ”,
    “kafka_admin_url“: “https://t111111q7ljf7f4n.svc02.us-south.eventstreams.cloud.ibm.com&#8221;,
    “kafka_brokers_sasl“: [
    “broker-1-t111111q7ljf7f4n.kafka.svc02.us-south.eventstreams.cloud.ibm.com:9093”,
    “broker-2-t111111q7ljf7f4n.kafka.svc02.us-south.eventstreams.cloud.ibm.com:9093″,
    ],
    },
    }service = appClient.serviceBindings.create(serviceBinding)# Set up the connector
    connector = appClient.dsc.create(name=”connector1″, serviceId=service.id, type=”eventstreams”)# Set up destinations
    destination = connector.destinations.create(name=”kafka-java-console-sample-topic”, partitions=1)# Set up rules
    rule1 = connector.rules.createEventRule(name=”allevents”, destinationName=”kafka-java-console-sample-topic”, typeId=”*”, eventId=”*”)

     

    Set the WIOTP_API_KEY and WIOTP_API_TOKEN variables to match your Watson IoT API Key, and execute the Python file.

     

    export WIOTP_API_KEY=’a-bmckzu-2uaorakza9′

    export WIOTP_API_TOKEN=’dI!tDqszBfl(pT)K)i’

    python bind.py

     

    NOTE: You can also use the IBM Watson IoT Platform Historical Data Storage Extension API to manage (create, list and delete) the service bindings and data store connectors.

12. Part 5: Integrate Watson IoT with Event Streams – Step 3: Activate Device Simulator for Testing

    To test the service binding between Watson IoT Platform and Event Streams, let’s set up a simulated device.

     

    Browse to Settings, select Device Simulator and enable the Activate Device Simulator option. You can use the default configuration to simulate events at a fixed frequency.

     

    

    

    

    

     

    Once the simulation has been created and enabled, you will see events being generated.

     

    If you have set up the data store connector correctly, you should see the messages published to the Kafka topic in the Event Streams service.

     

    When you run the sample IBM consumer application, you should see the Message consumed log messages:

     

    $ java -jar ./build/libs/kafka-java-console-sample-2.0.jar <kafka_brokers_sasl> <api_key> -consumer[2020-01-13 17:33:40,572] INFO Using command line arguments to find credentials. (com.eventstreams.samples.EventStreamsConsoleSample)
    [2020-01-13 17:33:40,576] INFO Kafka Endpoints: broker-1-t111111q7ljf7f4n.kafka.svc02.us-south.eventstreams.cloud.ibm.com:9093 (com.eventstreams.samples.EventStreamsConsoleSample)
    [2020-01-13 17:33:42,234] INFO Creating the topic kafka-java-console-sample-topic (com.eventstreams.samples.EventStreamsConsoleSample)
    [2020-01-13 17:33:48,058] INFO Topic kafka-java-console-sample-topic already exists (com.eventstreams.samples.EventStreamsConsoleSample)
    [2020-01-13 17:33:50,505] INFO [Partition(topic = kafka-java-console-sample-topic, partition = 0, leader = 2, replicas = [2,3,4], isr = [4,3,2], offlineReplicas = [])] (com.eventstreams.samples.ConsumerRunnable)
    [2020-01-13 17:33:50,511] INFO EventStreamsConsoleSample will run until interrupted. (com.eventstreams.samples.EventStreamsConsoleSample)
    [2020-01-13 17:33:50,512] INFO class com.eventstreams.samples.ConsumerRunnable is starting. (com.eventstreams.samples.ConsumerRunnable)
    [2020-01-13 17:33:59,527] INFO No messages consumed (com.eventstreams.samples.ConsumerRunnable)
    [2020-01-13 17:34:02,532] INFO No messages consumed (com.eventstreams.samples.ConsumerRunnable)
    [2020-01-13 17:34:02,865] INFO Message consumed: ConsumerRecord(topic = kafka-java-console-sample-topic, partition = 0, leaderEpoch = 6, offset = 106, CreateTime = 1578908009923, serialized key size = 139, serialized value size = 19, headers = RecordHeaders(headers = [], isReadOnly = false), key = {“orgId”:”bmckzu”,”format”:”json”,”timestamp”:”2020-01-13T09:33:28.095Z”,”deviceType”:”sensor”,”deviceId”:”sensor_1″,”eventType”:”event_1″}, value = {“randomNumber”:99}) (com.eventstreams.samples.ConsumerRunnable) 

    If you have the sample Joget Kafka app running, you should see the following log output showing the received messages:

     

    INFO 17 Jan 2020 16:31:58 org.joget.kafka.ConsumerRunnable – Executing script. Offset: 422
    INFO 17 Jan 2020 16:31:58 org.joget.kafka.ConsumerRunnable – Key={“orgId”:”bmckzu”,”format”:”json”,”timestamp”:”2020-01-17T08:31:56.169Z”,”deviceType”:”sensor”,”deviceId”:”sensor_1″,”eventType”:”event_1″}. Value={“randomNumber”:8}
    INFO 17 Jan 2020 16:31:58 KafkaConsumer – Message received {“orgId”:”bmckzu”,”format”:”json”,”timestamp”:”2020-01-17T08:31:56.169Z”,”deviceType”:”sensor”,”deviceId”:”sensor_1″,”eventType”:”event_1″}={“randomNumber”:8}
    INFO 17 Jan 2020 16:31:58 org.joget.kafka.ConsumerRunnable – Script execution complete.
    INFO 17 Jan 2020 16:31:58 org.joget.kafka.ConsumerRunnable – Script result: null
    INFO 17 Jan 2020 16:31:58 org.joget.kafka.ConsumerRunnable – Executing script. Offset: 423