如何从kafka主题检索消息，并在融合超文本传输协议接收器连接器中使用相同的值？

Question

我正在使用融合的 http 接收器连接器从 kafka 主题读取消息并将其发送到endpoint。下面是 http 接收器连接器，

    {
      "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":"" 
      }
    }

主题中的消息具有以下 json 格式，

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

因此，使用超文本传输协议接收器连接器，我需要从主题消息中检索“endpointurl”和“凭据”的值，并在同一连接器中使用与“http.api.url”、“connection.user”和“connection.password”键的值相同的值。

所以最终的格式应该是这样的，

    {
      "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"
      }
    }

如何从主题消息中检索数据并在同一连接器中使用它？

Answer 1

我找到了答案。这可以通过创建自定义连接器来实现。下面是http接收器自定义连接器的参考链接。

使用专家：https://github.com/university-of-auckland/kafka-http-sink-connector

使用gradle：https://github.com/aiven/aiven-kafka-connect-http

执行步骤，

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.

执行插件的命令，

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

connect-http-sink.properties文件的内容如下所示:

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