05.kafka connect Distributed 模式环境安装配置
前言：
前段时间在调研通过PostgreSQL Logical Decoding Pluggins方式实时获取数据库表增、删、改的数据。打算采用PostgreSQL 10之后版本自带的逻辑解码插件pgoutput + debezium for PostgreSQL connector的方案，因此，首先学习了一下kafka connect的相关知识，为了更加深入的了解研读了一些kafka-0.10.0的connect的相关源码，kafka connect具体实现细节可以在之后的源码分析文章中解答。本篇文章首先对kafka connect distributed模式下环境搭建和如何提交运行connector做一下整体的介绍。
安装
{KAFKA_HOME}代表kafka的安装路径
第一：首先编辑分布式启动所需要的配置文件，{KAFKA_HOME}/config/connect-distributed.properties，主要配置项目如下图所示：
下面分别介绍每一项的含义
1. bootstrap.servers：kafka集群地址(写集群中某一个可用的broker的地址即可)
2. group.id：kafka connect集群的group id
3. key.converter、value.converter：把数据导入到kafka的某个topic时，topic中数据的key和value按照某种converter转化，默认是json格式
4. key.converter.schemas.enable、value.converter.schemas.enable：指定topic中数据的key和value是否包含schema信息，包含格式如下：
如果设定为true，json中就会包含schema属性，如果设定为false，json中就只有payload属性
5. internal.key.converter、internal.value.converter和上面第3项含义一样
6. internal.key.converter.schemas.enable、internal.value.converter.schemas.enable和上面第4项含义一样
第5项和第6项是用于kafka connect内部使用topic的配置（要运行kafka connect的分布式环境需要三个内置topic，如下所示）
7. offset.storage.topic：用于保存connector运行中的offset，当connector宕机时可以继续从某个offset开始运行
8. config.storage.topic：用于保存connector的配置信息（注意：此topic只能由一个partition）
9. status.storage.topic：用于保存connector和task的状态
10. offset.flush.interval.ms：保存connector运行中offset到topic的频率
第二：基础配置完成之后，我们就可以启动kafka connect的分布式环境了
命令：{KAFKA_HOME} /connect-distributed.sh ../config/connect-distributed.properties &
经过上面两步我们的kafka connect分布式环境已经运行起来了。
相关restful命令
1. Method：POST，URL：http://ip:port/connectors 提交connector
2. Method：GET，URL：http://ip:port/connectors 获取所有connector
3. Method：DELETE，URL：http://ip:port/connectors/{connector name} 删除指定的connector name的connector
4. Method：GET，URL：http://ip:port/connectors/{connector name}/status 获取指定connecto name的运行状态
5. Method：POST，URL：http://ip:port/connectors/{connector name}/restart 重启指定connector name的connector
6. Method：GET，URL：http://ip:port/connectors/{connector name}/tasks/{task id}/status 获取指定task的运行状态
7. Method：GET，URL：http://ip:port/connector-plugins/ 获取kafka connect环境中的所有可执行connector plugins
上面只是列举了一些常用的restful api，还有一些其它的api，感兴趣的小伙伴可以参考源码中的ConnectorsResource、ConnectorPluginsResource、RootResource这三个文件里面提供了所有的api，也可以参考官方文档。
