腾讯轻联 逻辑组件使用指南

一、简介

·Apache Kafka 连接器用于连接 Kafka 集群和腾讯云数据连接器集群，可消费 Kafka 集群的消息并作为集成流的 trigger，也可将集成流的消息生产发布到 Kafka 集群中。

·Apache Kafka 是一个分布式发布-订阅消息传递系统，kafka 有以下一些基本概念：

·Producer：消息生产者，即向 kafka broker 发布消息的客户端。

·Consumer：消息消费者，负责消费 Kafka 服务器上的消息。

·Topic：主题，用于建立 Producer 和 Consumer 之间的订阅关系。

·Partition：消息分区，一个 topic 可以分为多个 partition，每个 partition 是一个有序的队列，partition 中的每条消息都会被分配一个有序的 ID（offset）。

·Broker：一台 kafka 服务器即为一个 broker，一个集群由多个 broker 组成，一个 broker 可以容纳多个 topic。

·Consumer Group：消费者分组，用于归组同类消费者，每个 consumer 属于一个特定的 consumer group，多个消费者可以共同消息一个 Topic 下的消息，每个消费者消费其中的部分消息，这些消费者就组成一个分组，拥有同一个分组名称。

·Offset：消息在 partition 中的偏移量，每一条消息在 partition 都有唯一的偏移量，消息者可以指定偏移量来指定要消费的消息。

·Producer、Consumer、Broker 关系如下：

·Broker 保存消息数据，和 topic、partition 关系示意图如下：

·同一 topic 的各 partition 逻辑关系如下，kafka 仅保证同一 partition 下的消息有序。

二、连接器配置

通用配置标签页-连接配置通用配置标签页-消费者配置高级配置标签页-消费者组配置高级配置标签页-生产者配置

三、操作配置

Apache Kafka 连接器包含 Consumer 和 Publish 两种操作。

四、输入参数

None，Consumer 操作没有输入参数。

五、输出

Consumer 操作执行成功后，输出结果会保存在 message 消息体的 payload；执行失败后，错误信息会保存在 message 消息体的 error。

组件输出的 message 信息如下：

atrrbites 属性列表：

例如：执行成功后，message payload 值为 byte 数组，message attribute 值如下：

{ "topic": "test-topic", "partition": 1, "offset": 4}

执行失败后，message error 值如下：

{ "Code": "CORE:RUNTIME", "Description": "kafka: client has run out of available brokers to talk to (Is your cluster reachable?)"}

六、案例

1. 按照 Kafka 集群信息，填写 kafka 连接器配置。

2. 新建流，将 kafka consumer 作为 trigger 节点。

3. 添加其他业务逻辑，处理消费到的数据，例如：增加 amqp publish，将消费到的消息发布到 rabbitMQ。

4. 若配置正确，流发布后，可以看到 kafka 中的流被正确消费，且被发布到 rabbitMQ 中。

七、输入参数

通用配置

高级配置

八、输出

Publish 操作执行成功后，输出属性会保存在 message 消息体的 attribute；执行失败后，错误信息会保存在 message 消息体的 error。

组件输出的 message 信息如下：

atrrbites 属性列表：

例如：执行成功后，message attribute 值如下：

{ "topic": "test-topic", "partition": 1, "offset": 4}

执行失败后，message error 值如下：

{ "Code": "CORE:RUNTIME", "Description": "kafka: client has run out of available brokers to talk to (Is your cluster reachable?)"}

九、案例

1. 按照 Kafka 集群信息，填写 kafka 连接器配置。

2. 新建流，添加 trigger，例如：HTTP listener。

3. 在流中增加 kafka publish 操作，填写消息内容，将消息发布到 Kafka 集群中。

4. 发布流，触发流后，若连接参数及消息参数配置正确，消息将成功投递到所配置的 Kafka 集群的对应队列中。