使用Kafka Connect建立测试数据生成器

在最近的一些项目中，我使用Apache Kafka开发了一些数据管道。在性能测试方面，数据生成老是会在整个活动中引入一些样板代码，例如建立客户端实例，编写控制流以发送数据，根据业务逻辑随机化有效负载等等。

在测试设置期间，拥有一个处理全部繁重工做的框架会很好，所以只须要回答两个基本和基本的问题：

• 数据应该是什么样的？（架构）
• 要生成多少数据？（体积）

有了Kafka Connect，事实证实实现自定义源链接器可以实现这一目标。如下是用于生成测试数据的示例属性列表的快速概述。

topic.name = generated.events 
poll.size = 10 
poll.interval.ms = 5000 
message.template = {“status”：“foo”，“direction”：“up”} 
random.fields = status：foo | bar |巴兹，方向：向上|向下|向左|向右

这些属性是不言自明的。为了回答上述两个基本问题：message.template和random.fields控制模式，而poll.size和poll.interval.ms控制音量。

基于这些属性，我建立了一个名为“kafka-connect-datagen”（或简称“datagen”）的自定义源链接器，可在GitHub上得到。

在下一节中，我将简要介绍一些实现细节。

实现自定义链接器

Kafka Connect源链接器将数据从数据存储复制到Kafka，而接收器链接则相反。虽然是源链接器，但datagen不会连接到任何数据存储; 它从内部生成数据。其实施的其他部分根据是标准卡夫卡链接开发指南：它延伸SourceConnector和SourceTask，并实现了一些生命周期方法的钩。如下片断缩写自datagen。

 

 

如代码所示，Connector定义Task要运行的类型和要为其设置的配置Task，同时Task是执行自定义逻辑的工做单元。二者Connector和Task实例都在一个Worker 进程中运行。该汇合的文档详细介绍了这些概念。

除了实现这两个类以外，还有一个步骤在运行演示以前：ConfigDef为用户定义配置列表（）。以后，这些类能够打包为Connector插件。在全面实施能够在GitHub上找到。

在下一节中，我将演示如何将插件与dockerized本地群集设置一块儿使用。

快速入门演示

在本快速入门示例中，咱们使用docker-compose管理全部必需的服务，如ZooKeeper，Kafka和Kafka Connect。要显示全部这些服务，请运行docker-compose up -d，而后运行docker-compose ps以打印状态信息，以下所示。

Name State Ports 
----------------------------------------------- -------------------- 
quickstart_broker_1 Up 0.0.0.0:9092->9092/tcp 
quickstart_connect_1 Up 0.0.0.0:8083->8083/tcp,t ... 
quickstart_kafka- connect-ui_1 Up 0.0.0.0:8001->8000/tcp 
quickstart_kafka-rest-proxy_1 Up 0.0.0.0:8082->8082/tcp 
quickstart_kafka-topics-ui_1 Up 0.0.0.0:8000->8000/tcp 
quickstart_zookeeper_1 Up 0.0。 0.0：2181-> 2181 / tcp，...

Kafka和Kafka Connect将须要更长的时间才能开始。感谢Landoop Ltd，咱们有这些不错的UI工具：打开http：// localhost：8000查看Kafka主题UI，http：// localhost：8001查看Kafka Connect UI。您也能够运行docker-compose logs -f以查看日志。

因为一般Kafka Connect服务是最后一个完成启动的服务，咱们能够经过运行docker-compose logs -f connect来查看其日志，以查看以下的正常运行指标。

INFO使用config offset -1（org.apache.kafka.connect.runtime.distributed.DistributedHerder）
启动链接器和任务INFO完成启动链接器和任务（org.apache.kafka.connect.runtime.distributed.DistributedHerder）

当全部服务彻底启动时，是时候建立“datagen”链接器实例了。如下是用于此演示的配置示例。它基本上设置“datagen”任务，每5秒生成10条消息。每条消息都使用定义的JSON消息模板和一些随机字段。运行如下命令以实例化Connector和Task。

curl -X POST http：// localhost：8083 / connectors \ 
-H'Content-Type：application / json'\ 
-H'Eccept：application / json'\ 
-d @ connect.source.datagen.json

如今，在Kafka主题UI中，咱们可以看到以generated.events定义的速率发布到主题的随机JSON消息。

 

kafka-connect-datagen发布消息

要中止生成，咱们能够转到Kafka Connect UI并暂停或删除链接器。一样，咱们可使用以下所示的REST API来实现相同的结果。查看此Confluent文档以获取更多操做。

＃pause链接器（若是成功则为空响应）
curl -X PUT http：// localhost：8083 / connectors / connect.source.datagen / pause

#delete链接器（若是成功则为空响应）
curl -X DELETE http：// localhost：8083 / connectors / connect.source.datagen

总之，咱们可以利用Kafka Connect，这是一种现成的工具，能够很好地与Kafka集成，以最少的样板代码实现随机数据生成。自定义链接器插件 - kafka-connect-datagen - 具备高度可移植性，能够进一步扩展以支持集成测试和不一样消息格式等功能。