Sending Messages to Confluent Cloud topic using Logic App

With Logic Apps, we can create workflows that connect to various services and systems, allowing us to automate tasks and streamline the business operations. In this blog post, we will explore how to use Azure Logic Apps to send messages to Kafka Confluent topic.

Currently, there is no out-of-box Kafka Confluent connector in logic app. We found that Kafka Confluent provides REST API Confluent Cloud API Reference Documentation. This sample shows how to use HTTP action in workflow to call the Kafka Confluent API which produce record to topic.

Prerequisites

• Azure Account and access to Azure Portal.
• Access to Confluent Cloud. Confluent Cloud is a fully managed pay-as-you-go Kafka service. You can get a free trial here.

Setup the Kafka cluster and topic

If you are new to Confluent Kafka, you can check their tutorials: Quick Start for Confluent Cloud | Confluent Documentation.

• Create a new Kafka cluster on Confluent Cloud.
• Navigate to the cluster and click Cluster settings. Note the REST endpoint. We will use following endpoint in this example:

• Create a new Kafka Topic called "LAtest" using the default topic settings.

• Create a new API Key and Secret. Navigate to the cluster and from the left menu, select Cluster Overview -> API Keys. Click Create key and follow the prompts to create a Global access API key. Note down the value of key and secret.

To communicate with the REST API, we need to use this API key ID and corresponding secret to create the base64 encoded string in the authorization header that will be included in the REST calls. To learn more, see Authentication part which describes Cloud and Cluster API keys and base64 encoding.

Create Logic App workflow

To produce message to a topic, we need to provide JSON data and a base64-encoded API key and secret to the REST Produce endpoint: /kafka/v3/clusters/<cluster-id>/topics/<topic-name>/records. Below is a sample REST code snippet (Non-streaming mode):

curl \ 
-X POST \ 
-H "Content-Type: application/json" \ 
-H "Authorization: Basic <base64-encoded API key and secret>" \ 
https://xxx.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-mxpx52/topics/LAtest/records \ 
-d '{"value":{"type":"JSON","data":"Hello World!"}}'

In the Logic App workflow, we can add a "When a HTTP request is received trigger" to fire the workflow for test. Then we add a HTTP action with following configurations:

Run the workflow

We can send the body message to target topic successfully.

View the Messages tab of Latest topic on the Confluent Cloud UI: