How to retrieve message from kafka topic and use the same as a value in the confluent http sink connector?

user_6746719 0
Viewed 744

I am using confluent http sink connector to read message from kafka topic and send it to an endpoint. Below is the http sink connector,

    {
      "name": "HttpSink",
      "config": {
        "topics": "http-messages",
        "tasks.max": "1",
        "connector.class": "io.confluent.connect.http.HttpSinkConnector",
        "value.converter": "org.apache.kafka.connect.storage.StringConverter",
        "topics":"topic-name",
        "request.method":"POST",
        "behavior.on.null.values":"ignore",
        "behavior.on.error":"log",
        "confluent.topic.bootstrap.servers": "localhost:9092",
        "confluent.topic.replication.factor": "1",
        "reporter.bootstrap.servers": "localhost:9092",
        "reporter.result.topic.name": "success-responses",
        "reporter.result.topic.replication.factor": "1",
        "reporter.error.topic.name":"error-responses",
        "reporter.error.topic.replication.factor":"1",
        "http.api.url": "",
        "auth.type":"BASIC",
        "connection.user":"",
        "connection.password":"" 
      }
    }

The messages in the topic has following json format,

    {
       "endpoint url": "http://localhost:8080/api/messages",
       "rest method":"",
       "credentials": {"username":"username", "password":"password"},
       "payload": {}
    }

So using http sink connector, I need to retrieve the value of "endpoint url" and "credentials" from topic message and use the same as a value for "http.api.url", "connection.user" and "connection.password" key's in the same connector.

So final format should look like,

    {
      "name": "HttpSink",
      "config": {
        "topics": "http-messages",
        "tasks.max": "1",
        "connector.class": "io.confluent.connect.http.HttpSinkConnector",
        "value.converter": "org.apache.kafka.connect.storage.StringConverter",
        "topics":"topic",
        "request.method":"POST",
        "behavior.on.null.values":"ignore",
        "behavior.on.error":"log",
        "confluent.topic.bootstrap.servers": "localhost:9092",
        "confluent.topic.replication.factor": "1",
        "reporter.bootstrap.servers": "localhost:9092",
        "reporter.result.topic.name": "success-responses",
        "reporter.result.topic.replication.factor": "1",
        "reporter.error.topic.name":"error-responses",
        "reporter.error.topic.replication.factor":"1",
        "http.api.url": "http://localhost:8080/api/messages",
        "auth.type":"BASIC",
        "connection.user":"username",
        "connection.password":"password"
      }
    }

How to retrieve data from the topic message and use it in the same connector?

confluent-platform apache-kafka-connect apache-kafka
1 Answers

I found the answer. This can be achieved by creating custom connector. Find below are the reference link to http sink custom connector.

Using maven : https://github.com/university-of-auckland/kafka-http-sink-connector

Using gradle : https://github.com/aiven/aiven-kafka-connect-http

Steps to execute,

Step 1 : Clone the project and build
Step 2 : Copy the jar to the desired location. For example, you can create a directory named <path-to-kafka>/share/kafka/plugins then copy the connector plugin jar.
Step 3 : Add this to the plugin path in your Connect properties file. For example, plugin.path=/usr/local/share/kafka/plugins. Kafka Connect finds the plugins using its plugin path. A plugin path is a comma-separated list of directories defined in the Kafka Connect's worker configuration.
Step 4 : Start the Connect workers with that configuration. Connect will discover all connectors defined within those plugins.

Command to execute the plugin,

$CONFLUENT_HOME/bin/connect-standalone $CONFLUENT_HOME/etc/kafka/connect-standalone.properties $CONFLUENT_HOME/etc/kafka/connect-http-sink.properties

And the connect-http-sink.properties file content looks like,

name=HttpSink
http.api.url=http://localhost:8080/api/messages
request.methods=POST
topics=http-test-topic
connector.class=com.connector.HttpSinkConnector
user_67467190
Related