增加 source & sink 端kafka

[listonejava]

TIS 开源平台中 没有实现 kafks端。请求实现。谢谢

[baisui1981]

传输使用的的协议为 json，
使用dataX 同步参照的例子：https://gitee.com/mjlfto/dataX/blob/master/kafkawriter/src/main/java/com/zcreate/datax/plugin/writer/kafkawriter/KafkaWriter.java
https://github.com/danny-plus/datax-kafkawriter/tree/master/src/main/java/ni/danny/datax/plugin/writer/kafkawriter

[baisui1981]

chunjun sink: https://github.com/DTStack/chunjun/tree/master/chunjun-connectors/chunjun-connector-kafka 对应的说明文档：https://dtstack.github.io/chunjun/documents/ChunJun%E8%BF%9E%E6%8E%A5%E5%99%A8@kafka@kafka-sink

[baisui1981]

查看airbyte 的kafka source端配置，主要有以下几项：

1. kafka source端的基本信息，如groupId
2. 通讯协议 protocol，有 plaintext，saslplantext，sasl ssl 三种
3. Subscription Method
4. MessageFormat，目前支持 json 和avro 两种
5. 等

具体配置可以查看 https://github.com/airbytehq/airbyte/blob/master/airbyte-integrations/connectors/source-kafka/src/main/resources/spec.json

根据spec.json 文件可以生成对应的 TIS plugin 脚手架代码：参考代码 plugins/tis-datax/tis-datax-kafka-plugin/src/main/java/com/qlangtech/tis/plugins/datax/kafka/reader/Conver2TISProps.java

kafka的sink端 配置基本如 source端，配置清单：https://github.com/airbytehq/airbyte/blob/master/airbyte-integrations/connectors/destination-kafka/src/main/resources/spec.json

[baisui1981]

kafka 发送数据样例：

{
  "id": "76d6d59f-5d08-41e8-a460-4ec3e03ad4d1",
  "tableName": "orderinfo",
  "occure_time": "1679274580528",
  "event": "insert" / "update" / "delete"
  "data": {
    "order_id": "aabbbdddd",
    "product_name": "aabbcc"
  }
}

可以参考debezium 数据格式：https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/table/formats/debezium/

[baisui1981]

目前已经实现了一个一个雏形，不足的是 在发送到kafka 中的update 消息目前之后 after状态的数据，这样用户没法根据before的状态判断数据是否已经更改

可以参考： https://github.com/qlangtech/flink/blob/f850f556f7f26f99636058481e57a251a9b654fb/flink-formats/flink-json/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory

内部有两种debeziumJson和CanalJson两种格式实现：

1. org.apache.flink.formats.json.debezium.DebeziumJsonFormatFactory
2. org.apache.flink.formats.json.canal.CanalJsonFormatFactory

[baisui1981]

已经测试通过：

采用debezium-json 格式：

{
	"before": null,
	"after": {
		"instance_id": "000102514520a464014",
		"order_id": "10000015361171282085747c1f92005a",
		"batch_msg": "",
		"type": 0,
		"ext": "{\"hitPrice\":0,\"isWait\":0,\"optionalType\":0}",
		"waitinginstance_id": "",
		"kind": 3,
		"parent_id": "",
		"pricemode": 1,
		"name": "test02",
		"makename": "",
		"taste": "",
		"spec_detail_name": "",
		"num": 1,
		"account_num": 1,
		"unit": "?",
		"account_unit": "",
		"price": 1,
		"member_price": 1,
		"fee": 1,
		"ratio": 100,
		"ratio_fee": 1,
		"ratio_cause": "",
		"status": 2,
		"kindmenu_id": "8000001564d710600164f9cd0c510172",
		"kindmenu_name": "??",
		"menu_id": "0",
		"memo": "",
		"is_ratio": 0,
		"entity_id": "80000015",
		"is_valid": 1,
		"create_time": 1535257651481,
		"op_time": 1535257670907,
		"last_ver": 3,
		"load_time": 1536117832,
		"modify_time": 1678704159,
		"draw_status": 0,
		"bookmenu_id": "",
		"make_id": "",
		"make_price": 0,
		"prodplan_id": "",
		"is_wait": 0,
		"specdetail_id": "",
		"specdetail_price": 0,
		"makeprice_mode": 1,
		"original_price": "1.0",
		"is_buynumber_changed": 1,
		"ratio_operator_id": "",
		"child_id": "",
		"kind_bookmenu_id": "",
		"specprice_mode": 2,
		"worker_id": "05eb689d72304a8d9064b20f58953d91",
		"is_backauth": 1,
		"service_fee_mode": 0,
		"service_fee": "0.0",
		"orign_id": "",
		"addition_price": 0,
		"has_addition": 0,
		"seat_id": ""
	},
	"op": "c",
	"source": {
		"table": "instancedetail"
	},
	"ts_ms": 1681615789046
}

使用canal-json 传输的消息例子：

{
	"data": [{
		"instance_id": "000102514520a464010",
		"order_id": "10000015361171282085747c1f92005a",
		"batch_msg": "",
		"type": 0,
		"ext": "{\"hitPrice\":0,\"isWait\":0,\"optionalType\":0}",
		"waitinginstance_id": "",
		"kind": 3,
		"parent_id": "",
		"pricemode": 1,
		"name": "test02",
		"makename": "",
		"taste": "",
		"spec_detail_name": "",
		"num": 1,
		"account_num": 1,
		"unit": "?",
		"account_unit": "",
		"price": 1,
		"member_price": 1,
		"fee": 1,
		"ratio": 100,
		"ratio_fee": 1,
		"ratio_cause": "",
		"status": 2,
		"kindmenu_id": "8000001564d710600164f9cd0c510172",
		"kindmenu_name": "美丽",
		"menu_id": "0",
		"memo": "",
		"is_ratio": 0,
		"entity_id": "80000015",
		"is_valid": 1,
		"create_time": 1535257651481,
		"op_time": 1535257670907,
		"last_ver": 3,
		"load_time": 1536117832,
		"modify_time": 1678704159,
		"draw_status": 0,
		"bookmenu_id": "",
		"make_id": "",
		"make_price": 0,
		"prodplan_id": "",
		"is_wait": 0,
		"specdetail_id": "",
		"specdetail_price": 0,
		"makeprice_mode": 1,
		"original_price": "1.0",
		"is_buynumber_changed": 1,
		"ratio_operator_id": "",
		"child_id": "",
		"kind_bookmenu_id": "",
		"specprice_mode": 2,
		"worker_id": "05eb689d72304a8d9064b20f58953d91",
		"is_backauth": 1,
		"service_fee_mode": 0,
		"service_fee": "0.0",
		"orign_id": "",
		"addition_price": 0,
		"has_addition": 0,
		"seat_id": ""
	}],
	"type": "INSERT",
	"table": "instancedetail",
	"ts": 1681616142062
}