Kafka consumer flow-trigger

9 minute read

Overview

This is a flow-trigger that enables triggering a flow that consumes messages from Apache Kafka. Using Apache Kafka is useful when implementing an event-driven microservice architecture.

It is part of the Kafka plugin, @axway/api-builder-plugin-ft-kafka. The plugin also contains a Kafka Producer flow-node. They can be used independently in that your application may only just consume messages from Kafka, it does not necessarily have to produce them.

You can install the Kafka plugin from the Plugins page, or execute the following command:

npm install @axway/api-builder-plugin-ft-kafka

For additional getting started information, refer to the Getting Started With API Builder.

Flow-trigger details

The following sections provide details of the available Kafka Consumer parameters.

Connection parameters

Trigger parameters

Quick start

We recommend that you first get familiar with Kafka Key Concepts.

To get started using the Kafka flow-trigger, you need to have access to a Kafka server. You can follow this guide, Running Kafka in Development for steps to run Kafka and Zookeeper in a Docker container.

Alternatively, if you do not wish to use Docker, you can download Kafka directly and extract into a directory. You must start zookeeper first:

// Start zookeeper

./bin/zookeeper-server-start.sh config/zookeeper.properties

After zookeeper starts, you can start Kafka:

// Start kafka

./bin/kafka-server-start.sh config/server.properties

When shutting down, shutdown Kafka first, and then zookeeper.

How to use the Kafka Consumer flow-trigger

Example - Consume JSON messages

For this example, we will create a “Consumer flow” and configure it to consume JSON messages from a Kafka topic, “messages”. Generally speaking, your application would be unlikely to read and write to the same topic. However, to demonstrate this example, we will also create a “Producer flow” to write JSON to the Kafka topic, “messages”.

Create a consumer flow

Follow the instructions on Create a new flow to create a “Consumer flow”. In this flow, you will want to drag the Kafka Consumer from the Flow-Nodes > Flow-Triggers panel on the left, into the flow graph on the right. Configure the Kafka Consumer flow-trigger as shown below. The Consumer group ID is integral to the scalability of Kafka. For example, you can configure multiple instances of API Builder with the same Consumer group ID to achieve scalability.

This configures the Kafka Consumer to receive JSON messages from the “messages” topic. However, we also need to configure the Connection details. Scroll the Kafka Consumer property panel to the bottom, and enter your Kafka server’s details. This example is using “localhost:9092” and is uniquely identifying this service as a unique “client-1” for Client ID.

We will also want to do something with the Kafka message in the flow, but first we need to understand what a Kafka Message actually is so that we can work with it. The Kafka Consumer outputs a Message value called $.request, which can be seen on the Outputs tab.

If you click the info icon , it will display the JSON schema for a Message. In this example, we are only interested in the value.

// JSON schema for Message

{
  "properties": {
    "value": {
      "description": "The message payload. The type of data depends on the configured \"Message format\"."
    },
    "key": {
      "description": "The message key encoded as a buffer.",
      "oneOf": [
        {
          "type": "object"
        },
        {
          "type": "null"
        }
      ]
    },
    "headers": {
      "type": "object",
      "description": "Key-value pairs of message headers.",
      "additionalProperties": {
        "type": "object",
        "description": "The header value encoded as a buffer."
      }
    },
    "offset": {
      "type": "string",
      "description": "The current consumer position."
    },
    "timestamp": {
      "type": "string",
      "description": "The number of milliseconds since the Unix epoch."
    }
  }
}

So to configure this flow to use the Message value, click on the Flow inputs tab, and edit the request input as shown below.

This configures the Kafka Consumer to provide the output Message value as input into the flow.

On the Flow inputs tab, it lists the optional named inputs that can be provided to the flow: request, config, and env. These keys are hard-coded into the flow and their JSONPath selectors are provided automatically. The request maps the output of the Kafka Consumer as input into the flow, so we want change the value to use $.request.value as the value of request (this selects the value property from the Message object). From within the flow, the respective input values can be accessed as $.request, $.config, and $.env.

Now, let’s do something with $.request by adding the Mustache flow-node to the graph, and configuring Data to be $.request, and the Template to be “{{data.message}}” (without quotes).

Now the flow is ready to be triggered by a Kafka Producer. However, let’s verify that the flow works as expected. Click on the debugger icon in the upper-right of the graph to reveal the debug panel. Change the “request” to have a JSON message with the value “banana” as we plan on this flow receiving a JSON input from the “Producer flow”.

Click Execute Flow. The flow editor will not show much, just a message, “Flow successfully executed with no response”. However, if you check the console window where you launched your API Builder project, you see a detailed debug log showing the execution, and that it handled the “banana” message as expected:

// Flow output

1618583651686 DEBUG [request-id: 38b6f1ca-5fac-4b88-aefe-4f428f2b1e17] Flow invoked by debugger: Consumer
1618583651687 DEBUG [request-id: 38b6f1ca-5fac-4b88-aefe-4f428f2b1e17] Waiting: Format string (mustache.1)
1618583651687 DEBUG [request-id: 38b6f1ca-5fac-4b88-aefe-4f428f2b1e17] Invoking: Format string (mustache.1)
1618583651688 DEBUG [request-id: 38b6f1ca-5fac-4b88-aefe-4f428f2b1e17]   finished flow-node Format string (mustache.1) route: []
1618583651688 DEBUG [request-id: 38b6f1ca-5fac-4b88-aefe-4f428f2b1e17]   Format string (mustache.1) writing "Banana" to ctx as: $.value
1618583651689 DEBUG [request-id: 38b6f1ca-5fac-4b88-aefe-4f428f2b1e17] Flow debugging completed: Consumer

After, you can click Apply to save the flow. If all goes well, the Status of the Kafka Consumer flow-trigger will change to enabled. If there are any errors, you may need to check connectivity.

Create a producer flow

Follow the instructions on Create a new flow to create a “Producer flow”. In this flow, you will want to drag the Kafka Producer from the Flow-Nodes > Flow-Triggers panel on the left, into the flow graph on the right. Configure the flow-node to have the properties as shown below.

Click on the debugger icon in the upper-right of the graph, do not change any values, and click Execute Flow, and check your console debug log, you should see:

Format string (mustache.1) writing “Hello!” to ctx as: $.value