You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by zh...@apache.org on 2022/08/25 08:28:39 UTC
[rocketmq-connect] branch master updated: [ISSUE #255]Standardize configuration and update documentation (#259)
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-connect.git
The following commit(s) were added to refs/heads/master by this push:
new 037aaf6 [ISSUE #255]Standardize configuration and update documentation (#259)
037aaf6 is described below
commit 037aaf617a4a1593db2cc2a150506ec0503c6912
Author: xiaoyi <su...@163.com>
AuthorDate: Thu Aug 25 16:28:33 2022 +0800
[ISSUE #255]Standardize configuration and update documentation (#259)
* Fix debezium demecial type conversion problem #190
* Upgrade rocketmq-replicator API to v0.1.3 #189
* Encountered change event for table databasename.tablename whose schema isn`t known to this connector #191
* Debezium mysql source connector delete event causes null pointer #196
* remove local config
* Debezium mysql source connector delete event causes null pointer #196
* Rocketmq replicator running null pointer #205
* Add pause and resume to the task, and optimize some rest APIs #238
* fixed
* fixed
* remove useless code
* Optimize WorkerSinkTask，add pause and resume feature, replace DefaultMQPullConsumer with DefaultLitePullConsumerImpl #238
* fixed
* fixed
* fixed
* fixed
* fixed
* custom MessageQueueListener #238
* fixed
* remove useless info
* optimize rest APIs #238
* add task rest api #238
* fixed
* fixed
* fixed
* fixed
* Upgrade JAVALIN 2.8.0 to 4.6.4
* fixed
* fixed
* fixed
* fixed
* fixed
* fixed
* test fixed bug
* Optimize rest api #238
* Fix state persistence logic
* format code
* fixed
* optimize method
* sink close auto commit
* standard config
* fixed
* fixed
* add config validate
* fixed
* fixed
* update readme
* add jackson
* fixed
* update config
* add transform config
* add debezium mysql config demo
* fixed get status bug
* format code
* fixed
* validate timestamp empty
* reformat code
* fix conflict
---
README.md | 302 +++++++++++++++------
.../rocketmq-connect-debezium-mysql/README.md | 28 ++
connectors/rocketmq-connect-jdbc/README.md | 73 ++---
distribution/conf/connect-file-sink.properties | 2 +-
distribution/conf/connect-file-source.properties | 2 +-
pom.xml | 5 +
.../connect/runtime/DistributedConnectStartup.java | 4 +-
.../connect/runtime/StandaloneConnectStartup.java | 4 +-
.../connect/runtime/config/ConnectorConfig.java | 127 +++++++++
.../runtime/config/RuntimeConfigDefine.java | 111 --------
.../runtime/config/SinkConnectorConfig.java | 55 +++-
.../runtime/config/SourceConnectorConfig.java | 16 ++
.../{ConnectConfig.java => WorkerConfig.java} | 262 +++++++++++-------
.../SourceTaskOffsetCommitter.java | 10 +-
.../runtime/connectorwrapper/TransformChain.java | 10 +-
.../connect/runtime/connectorwrapper/Worker.java | 30 +-
.../runtime/connectorwrapper/WorkerConnector.java | 11 +-
.../runtime/connectorwrapper/WorkerDirectTask.java | 4 +-
.../runtime/connectorwrapper/WorkerSinkTask.java | 17 +-
.../runtime/connectorwrapper/WorkerSourceTask.java | 16 +-
.../runtime/connectorwrapper/WorkerTask.java | 6 +-
.../controller/AbstractConnectController.java | 14 +-
.../controller/distributed/DistributedConfig.java | 4 +-
.../distributed/DistributedConnectController.java | 4 +-
.../runtime/controller/isolation/Plugin.java | 4 +-
.../controller/standalone/StandaloneConfig.java | 4 +-
.../runtime/errors/DeadLetterQueueConfig.java | 62 ++---
.../runtime/errors/DeadLetterQueueReporter.java | 9 +-
.../runtime/errors/ReporterManagerUtil.java | 4 +-
.../service/AbstractConfigManagementService.java | 34 ++-
.../runtime/service/ClusterManagementService.java | 6 +-
.../service/ClusterManagementServiceImpl.java | 8 +-
.../runtime/service/ConfigManagementService.java | 6 +-
.../service/ConfigManagementServiceImpl.java | 78 +++---
.../runtime/service/DefaultConnectorContext.java | 4 +-
.../runtime/service/PositionManagementService.java | 6 +-
.../service/PositionManagementServiceImpl.java | 6 +-
.../connect/runtime/service/RebalanceImpl.java | 4 +-
.../runtime/service/StateManagementService.java | 4 +-
.../service/StateManagementServiceImpl.java | 18 +-
.../memory/FilePositionManagementServiceImpl.java | 4 +-
.../memory/MemoryClusterManagementServiceImpl.java | 6 +-
.../memory/MemoryConfigManagementServiceImpl.java | 53 ++--
.../memory/MemoryStateManagementServiceImpl.java | 4 +-
...llocateConnAndTaskStrategyByConsistentHash.java | 6 +-
.../connect/runtime/stats/ConnectStatsManager.java | 6 +-
.../connect/runtime/utils/ConnectUtil.java | 37 +--
.../runtime/utils/datasync/BrokerBasedLog.java | 18 +-
.../connect/runtime/config/ConnectConfigTest.java | 9 +-
.../runtime/config/SinkConnectorConfigTest.java | 6 +-
.../SourceTaskOffsetCommitterTest.java | 6 +-
.../connectorwrapper/TransformChainTest.java | 4 +-
.../connectorwrapper/WorkerSinkTaskTest.java | 20 +-
.../connectorwrapper/WorkerSourceTaskTest.java | 15 +-
.../runtime/connectorwrapper/WorkerTest.java | 33 ++-
.../testimpl/TestPositionManageServiceImpl.java | 4 +-
.../DistributedConnectControllerTest.java | 6 +-
.../distributed/TestConfigManagementService.java | 4 +-
.../connect/runtime/rest/RestHandlerTest.java | 18 +-
.../service/ConfigManagementServiceImplTest.java | 12 +-
.../service/PositionManagementServiceImplTest.java | 6 +-
.../connect/runtime/utils/TransferUtilsTest.java | 14 +-
.../runtime/utils/datasync/BrokerBasedLogTest.java | 6 +-
.../rocketmq/connect/file/FilterTransform.java | 1 -
64 files changed, 1017 insertions(+), 655 deletions(-)
diff --git a/README.md b/README.md
index 2fb2e34..376af57 100644
--- a/README.md
+++ b/README.md
@@ -17,14 +17,15 @@ rocketmq-connect-sample的主要作用是从源文件中读取数据发送到Roc
2. 64bit JDK 1.8+;
3. Maven 3.2.x或以上版本;
4. 启动 [RocketMQ](https://rocketmq.apache.org/docs/quick-start/);
-5. 创建测试Topic
- sh ${ROCKETMQ_HOME}/bin/mqadmin updateTopic -t fileTopic -n localhost:9876 -c DefaultCluster -r 8 -w 8
+5. 创建测试Topic
+> sh ${ROCKETMQ_HOME}/bin/mqadmin updateTopic -t fileTopic -n localhost:9876 -c DefaultCluster -r 8 -w 8
-**tips** : ${ROCKETMQ_HOME} 位置说明
-bin-release.zip 版本：/rocketmq-all-4.9.4-bin-release
+**tips** : ${ROCKETMQ_HOME} 位置说明
-source-release.zip 版本：/rocketmq-all-4.9.4-source-release/distribution
+>bin-release.zip 版本：/rocketmq-all-4.9.4-bin-release
+>
+>source-release.zip 版本：/rocketmq-all-4.9.4-source-release/distribution
## 2.构建Connect
@@ -48,16 +49,15 @@ sh bin/connect-standalone.sh -c conf/connect-standalone.conf &
```
**tips**: 可修改 /bin/runconnect.sh 适当调整 JVM Parameters Configuration
-```
-JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m"
-```
+>JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m"
+
runtime启动成功：
-The standalone worker boot success.
+>The standalone worker boot success.
查看启动日志文件：
-tail -100f ~/logs/rocketmqconnect/connect_runtime.log
+>tail -100f ~/logs/rocketmqconnect/connect_runtime.log
ctrl + c 退出日志
@@ -69,48 +69,49 @@ touch test-source-file.txt
echo "Hello \r\nRocketMQ\r\n Connect" >> test-source-file.txt
-curl -X POST -H "Content-Type: application/json" http://127.0.0.1:8082/connectors/fileSourceConnector -d '{"connector-class":"org.apache.rocketmq.connect.file.FileSourceConnector","filename":"test-source-file.txt","connect-topicname":"fileTopic"}'
+curl -X POST -H "Content-Type: application/json" http://127.0.0.1:8082/connectors/fileSourceConnector -d '{"connector.class":"org.apache.rocketmq.connect.file.FileSourceConnector","filename":"test-source-file.txt","connect.topicname":"fileTopic"}'
```
看到以下日志说明 file source connector 启动成功了
-tail -100f ~/logs/rocketmqconnect/connect_runtime.log
-
-2019-07-16 11:18:39 INFO pool-7-thread-1 - **Source task start**, config:{"properties":{"source-record-...
+>tail -100f ~/logs/rocketmqconnect/connect_runtime.log
+>
+>2019-07-16 11:18:39 INFO pool-7-thread-1 - **Source task start**, config:{"properties":{"source-record-...
#### source connector配置说明
-| key | nullable | default | description |
-| ----------------------- | -------- | -------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------- |
-| connector-class | false | |实现 Connector接口的类名称（包含包名）|
-| filename | false | | 数据源文件名称 |
-| connect-topicname | false | |同步文件数据所需topic|
+| key | nullable | default | description |
+|-------------------| -------- | ---------------------|--------------------------|
+| connector.class | false | | 实现 Connector接口的类名称（包含包名） |
+| filename | false | | 数据源文件名称 |
+| connect.topicname | false | | 同步文件数据所需topic |
## 5.启动sink connector
```
-curl -X POST -H "Content-Type: application/json" http://127.0.0.1:8082/connectors/fileSinkConnector -d '{"connector-class":"org.apache.rocketmq.connect.file.FileSinkConnector","filename":"test-sink-file.txt","connect-topicname":"fileTopic"}'
+curl -X POST -H "Content-Type: application/json" http://127.0.0.1:8082/connectors/fileSinkConnector -d '{"connector.class":"org.apache.rocketmq.connect.file.FileSinkConnector","filename":"test-sink-file.txt","connect.topicnames":"fileTopic"}'
cat test-sink-file.txt
```
-tail -100f ~/logs/rocketmqconnect/connect_runtime.log
+
+> tail -100f ~/logs/rocketmqconnect/connect_runtime.log
看到以下日志说明file sink connector 启动成功了
-2019-07-16 11:24:58 INFO pool-7-thread-2 - **Sink task start**, config:{"properties":{"source-record-...
+> 2019-07-16 11:24:58 INFO pool-7-thread-2 - **Sink task start**, config:{"properties":{"source-record-...
如果 test-sink-file.txt 生成并且与 source-file.txt 内容一样，说明整个流程正常运行。
文件内容可能顺序不一样，这主要是因为RocketMQ发到不同queue时，接收不同queue消息顺序可能也不一致导致的，是正常的。
#### sink connector配置说明
-| key | nullable | default | description |
-| ----------------------- | -------- | ------- | -------------------------------------------------------------------------------------- |
-| connector-class | false | | 实现Connector接口的类名称（包含包名） |
-| filename | false | | sink拉去的数据保存到文件 |
-| connect-topicname | false | | sink需要处理数据消息topics |
+| key | nullable | default | description |
+|--------------------| -------- | ------- | -------------------------------------------------------------------------------------- |
+| connector.class | false | | 实现Connector接口的类名称（包含包名） |
+| filename | false | | sink拉去的数据保存到文件 |
+| connect.topicnames | false | | sink需要处理数据消息topics |
```
注：source/sink配置文件说明是以rocketmq-connect-sample为demo，不同source/sink connector配置有差异，请以具体sourc/sink connector 为准
@@ -129,7 +130,7 @@ curl http://127.0.0.1:8082/connectors/fileSourceConnector/stop
```
看到以下日志说明connector停止成功了
-**Source task stop**, config:{"properties":{"source-record-converter":"org.apache.rocketmq.connect.runtime.converter.JsonConverter","filename":"/home/zhoubo/IdeaProjects/my-new3-rocketmq-externals/rocketmq-connect/rocketmq-connect-runtime/source-file.txt","task-class":"org.apache.rocketmq.connect.file.FileSourceTask","topic":"fileTopic","connector-class":"org.apache.rocketmq.connect.file.FileSourceConnector","update-timestamp":"1564765189322"}}
+>**Source task stop**, config:{"properties":{"source-record-converter":"org.apache.rocketmq.connect.runtime.converter.JsonConverter","filename":"/home/zhoubo/IdeaProjects/my-new3-rocketmq-externals/rocketmq-connect/rocketmq-connect-runtime/source-file.txt","task-class":"org.apache.rocketmq.connect.file.FileSourceTask","topic":"fileTopic","connector-class":"org.apache.rocketmq.connect.file.FileSourceConnector","update-timestamp":"1564765189322"}}
## 7.停止Worker进程
@@ -137,19 +138,22 @@ curl http://127.0.0.1:8082/connectors/fileSourceConnector/stop
sh bin/connectshutdown.sh
```
-
## 8.日志目录
- ${user.home}/logs/rocketmqconnect
+ >${user.home}/logs/rocketmqconnect
## 9.配置文件
持久化配置文件默认目录 /tmp/storeRoot
-1. connectorConfig.json connector配置持久化文件
-2. position.json source connect数据处理进度持久化文件
-3. taskConfig.json task配置持久化文件
-4. offset.json sink connect数据消费进度持久化文件
+| key | description |
+|----------------------|---------------------------|
+| connectorConfig.json | connector配置持久化文件 |
+| position.json | source connect数据处理进度持久化文件 |
+| taskConfig.json | task配置持久化文件 |
+| offset.json | sink connect数据消费进度持久化文件 |
+| connectorStatus.json | connector 状态持久化文件 |
+| taskStatus.json | task 状态持久化文件 |
## 10.配置说明
@@ -182,77 +186,208 @@ pluginPaths=rocketmq-connect-sample/target/rocketmq-connect-sample-0.0.1-SNAPSHO
## 11.其它restful接口
-
-查看集群节点信息：
-
+### 集群信息
++ 查看集群节点信息：
```
curl -X GET http://(your worker ip):(port)/getClusterInfo
```
-查看集群中Connector和Task配置信息：
++ 重新加载Connector插件目录下的Connector包：
```
-curl -X GET http://(your worker ip):(port)/getConfigInfo
+curl -X GET http://(your worker ip):(port)/plugin/reload
```
-查看当前节点分配Connector和Task配置信息：
+### Connector/Task管理
++ 创建或更新connector（存在且配置不同会更新，不存在创建）
```
-curl -X GET http://(your worker ip):(port)/getAllocatedInfo
+curl -X GET http://(your worker ip):(port)/connectors/{connectorName}
```
-
-查看指定Connector配置信息：
-
++ Pause(暂停)指定的connector
+```
+curl -X GET http://(your worker ip):(port)/connectors/{connectorName}/pause
+```
++ Resume(重启)指定的connector
+```
+curl -X GET http://(your worker ip):(port)/connectors/{connectorName}/resume
+```
++ Pause(暂停)所有的connector
```
-curl -X GET http://(your worker ip):(port)/connectors/(connector name)/config
+curl -X GET http://(your worker ip):(port)/connectors/pause/all
```
-查看指定Connector状态：
++ Resume(重启)所有的connector
+```
+curl -X GET http://(your worker ip):(port)/connectors/resume/all
+```
++ 停止并删除指定的connector(谨慎使用)
+```
+curl -X GET http://(your worker ip):(port)/connectors/{connectorName}/stop
```
-curl -X GET http://(your worker ip):(port)/connectors/(connector name)/status
++ 停止并删除所有的connector(谨慎使用)
+```
+curl -X GET http://(your worker ip):(port)/connectors/stop/all
```
-停止所有Connector：
++ 列举集群中所有connector信息
+```
+curl -X GET http://(your worker ip):(port)/connectors/list
+```
++ 根据connectorName获取connector的配置信息
+```
+curl -X GET http://(your worker ip):(port)/connectors/{connectorName}/config
+```
++ 根据connectorName获取connector的状态
+```
+curl -X GET http://(your worker ip):(port)/connectors/{connectorName}/status
+```
++ 根据connectorName获取connector下所有task信息
+```
+curl -X GET http://(your worker ip):(port)/connectors/{connectorName}/tasks
+```
++ 根据connectorName和task id获取task状态
```
-curl -X GET http://(your worker ip):(port)/connectors/stopAll
+curl -X GET http://(your worker ip):(port)/connectors/{connectorName}/tasks/{task}/status
```
-重新加载Connector插件目录下的Connector包：
++ 获取当前worker分配的connector信息
+```
+curl -X GET http://(your worker ip):(port)/allocated/connectors
+```
++ 获取当前worker分配的task信息
```
-curl -X GET http://(your worker ip):(port)/plugin/reload
+curl -X GET http://(your worker ip):(port)/allocated/tasks
```
+
从内存删除Connector配置信息（谨慎使用）：
```
curl -X GET http://(your worker ip):(port)/connectors/(connector name)/delete
```
+## 12.Connector通用配置参数说明
+
+| key | nullable | default | description |
+|-----------------------------|----------|---------------------------------------------|-----------------------------------------------------------------------------|
+| connector.class | false | | 指定插件的class信息 |
+| max.tasks | false | 1 | connector下运行的task的数量，根据情况而定 |
+| value.converter | false | org.apache.rocketmq.connect.runtime.converter.record.StringConverter | ConnectRecord key的转换器 |
+| key.converter | false | org.apache.rocketmq.connect.runtime.converter.record.StringConverter | ConnectRecord value的转换器 |
+| transforms | true | | 配置的数据转换器，多个需要用 ","分割 |
+| errors.log.include.messages | true | false | 是否将导致故障的connector异常信息纪录在日志中 |
+| errors.log.enable | true | false | 如果为true，请将每个错误以及失败操作和问题记录的详细信息写入Connect应用程序日志。默认情况下，这是“false”，因此只报告不可容忍的错误。 |
+| errors.retry.timeout | true | 0 | 重试超时时间 |
+| errors.retry.delay.max.ms | true | 60000 | 重试延迟时间 |
+| errors.tolerance | true | ToleranceType.NONE | 在Connector操作期间容忍错误的行为。“NONE”是默认值，表示任何错误都将导致连接器任务立即失败；"ALL“更改行为以跳过有问题的记录。 |
+#### Transform配置案例[单个配置]
+```
+"transforms": "Replace",
+"transforms.Replace.field.pattern": "company",
+"transforms.Replace.field.replacement": "company02",
+"transforms.Replace.class": "org.apache.rocketmq.connect.transforms.PatternRename$Value",
+```
+#### Transform配置案例[多个配置],会按照配置顺序依次执行
+```
+"transforms": "Replace, Replace02",
+"transforms.Replace.field.pattern": "company",
+"transforms.Replace.field.replacement": "company02",
+"transforms.Replace.class": "org.apache.rocketmq.connect.transforms.PatternRename$Value",
+"transforms.Replace02.field.pattern": "company02",
+"transforms.Replace02.field.replacement": "company03",
+"transforms.Replace02.class": "org.apache.rocketmq.connect.transforms.PatternRename$Value",
+```
+> 1.transforms: 为固定配置,不可变
+>
+> 2.Replace, Replace02: 为配置名称，可自定义, 多个用","分割，若为多个，下面配置均需重复配置
+>
+> 3.transforms.${transform-name}.class: 用此配置来表示transform的class
+>
+> 4.transforms.${transform-name}.{config.key}: transform中使用的多个配置项；
+
+#### Transform接口
+```java
+
+package io.openmessaging.connector.api.component;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.data.ConnectRecord;
+
+public interface Transform<R extends ConnectRecord> extends Component {
+ default void validate(KeyValue config) {
+ }
+
+ R doTransform(R var1);
+}
-## 12.runtime配置参数说明
-
-| key | nullable | default | description |
-| ------------------------ | -------- | ------------------------------------------------------------ | ------------------------------------------------------------ |
-| workerId | false | DEFAULT_WORKER_1 | 集群节点唯一标识 |
-| namesrvAddr | false | loacalhost:9876 | RocketMQ Name Server地址列表，多个NameServer地址用分号隔开 |
-| httpPort | false | 8082 | runtime提供restful接口服务端口 |
-| pluginPaths | false | | source或者sink目录，启动runttime时加载 |
-| storePathRootDir | true | /tmp/connectorStore | 持久化文件保存目录 |
-| positionPersistInterval | true | 20s | source端持久化position数据间隔 |
-| offsetPersistInterval | true | 20s | sink端持久化offset数据间隔 |
-| configPersistInterval | true | 20s | 集群中配置信息持久化间隔 |
-| rmqProducerGroup | true | defaultProducerGroup | Producer组名，多个Producer如果属于一个应用，发送同样的消息，则应该将它们归为同一组 |
-| rmqConsumerGroup | true | defaultConsumerGroup | Consumer组名，多个Consumer如果属于一个应用，发送同样的消息，则应该将它们归为同一组 |
-| maxMessageSize | true | 4MB | RocketMQ最大消息大小 |
-| operationTimeout | true | 3s | Producer发送消息超时时间 |
-| rmqMaxRedeliveryTimes | true | | 最大重新消费次数 |
-| rmqMessageConsumeTimeout | true | 3s | Consumer超时时间 |
-| rmqMaxConsumeThreadNums | true | 32 | Consumer客户端最大线程数 |
-| rmqMinConsumeThreadNums | true | 1 | Consumer客户端最小线程数 |
-| allocTaskStrategy | true | org.apache.rocketmq.connect.<br>runtime.service.strategy.<br>DefaultAllocateConnAndTaskStrategy | 负载均衡策略类 |
+```
+> 1. 自定义Transform只需要实现以上接口即可，将自定义包按照到 [pluginPaths]指定的插件目录下面即可;
+>
+> 2. 平台已经实现的transform在 transforms module下面，欢迎大家定义和扩展更加丰富的transform库
+>> org.apache.rocketmq.connect.transforms.ChangeCase$Key
+>> org.apache.rocketmq.connect.transforms.ChangeCase$Value
+>> org.apache.rocketmq.connect.transforms.ExtractNestedField$Value
+>> org.apache.rocketmq.connect.transforms.ExtractNestedField$Key
+>> org.apache.rocketmq.connect.transforms.PatternRename$Key
+>> org.apache.rocketmq.connect.transforms.PatternRename$Value
+>> org.apache.rocketmq.connect.transforms.PatternFilter$Key
+>> org.apache.rocketmq.connect.transforms.PatternFilter$Value
+>> org.apache.rocketmq.connect.transforms.SetMaximumPrecision$Key
+>> org.apache.rocketmq.connect.transforms.SetMaximumPrecision$Value
+
+
+
+
+### Source Connector特殊配置
+| key | nullable | default | description |
+|-----------------------------|---------|---------|------------------------------------------------------|
+| connect.topicname | true | | 指定数据写入的topic，若不配置则直接取position中key为topic的值，若取不到则抛出异常 |
+
+### Sink Connector特殊配置
+| key | nullable | default | description |
+|---------------------------------------------------|----------|-----------------------|----------------------------|
+| connect.topicnames | false | | sink订阅的topic配置多个用","分割 |
+| task.group.id | true | connect-{connectName} | sink订阅topic的group id |
+| errors.deadletterqueue.topic.name | true | | 错误队列的topic配置 |
+| errors.deadletterqueue.read.queue.nums | true | 8 | 错误队列创建时指定读的队列数 |
+| errors.deadletterqueue.write.queue.nums | true | 8 | 错误队列创建时指定写的队列数 |
+| errors.deadletterqueue.context.properties.enable | true | false | 上报错误数据是否包含错误的详细的属性信息 |
+
+## 13.Worker配置参数说明
+
+> ❌ 表示该模式下不需要配置此选项
+> ✅ 表示该模式下配置生效
+
+| key | nullable | standalone | distributed | default | description |
+|-----------------------------|----------|------------|-------------|--------------------------------------------------------------------|---------------------------------------------------|
+| workerId | false | ✅ ️ | ✅ | DEFAULT_WORKER_1 | 集群节点唯一标识 |
+| namesrvAddr | false | ✅ | ✅ | loacalhost:9876 | RocketMQ Name Server地址列表，多个NameServer地址用分号隔开 |
+| httpPort | false | ✅ | ✅ | 8082 | runtime提供restful接口服务端口 |
+| pluginPaths | false | ✅ | ✅ | | source或者sink目录，启动runttime时加载 |
+| storePathRootDir | true | ✅ | ✅ | /tmp/connectorStore | 持久化文件保存目录 |
+| positionStoreTopic | true | ❌ | ✅ | connector-position-topic | source端position变更通知topic |
+| positionPersistInterval | true | ✅ | ✅ | 20s | source端持久化position数据间隔 |
+| configStoreTopic | true | ❌ | ✅ | connector-config-topic | 集群connector配置变更通知topic |
+| configPersistInterval | true | ❌ | ✅ | 20s | 集群中配置信息持久化间隔 |
+| connectStatusTopic | true | ❌ | ✅ | connect-status-topic | connector和task状态变更通知 |
+| statePersistInterval | true | ❌ | ✅ | 20s | connector及task状态持久化间隔 |
+| rmqProducerGroup | true | ✅ | ✅ | defaultProducerGroup | Producer组名，多个Producer如果属于一个应用，发送同样的消息，则应该将它们归为同一组 |
+| rmqConsumerGroup | true | ✅ | ✅ | defaultConsumerGroup | Consumer组名，多个Consumer如果属于一个应用，发送同样的消息，则应该将它们归为同一组 |
+| maxMessageSize | true | ✅ | ✅ | 4MB | RocketMQ最大消息大小 |
+| operationTimeout | true | ✅ | ✅ | 3s | Producer发送消息超时时间 |
+| rmqMaxRedeliveryTimes | true | ✅ | ✅ | | 最大重新消费次数 |
+| rmqMessageConsumeTimeout | true | ✅ | ✅ | 3s | Consumer超时时间 |
+| rmqMaxConsumeThreadNums | true | ✅ | ✅ | 32 | Consumer客户端最大线程数 |
+| rmqMinConsumeThreadNums | true | ✅ | ✅ | 1 | Consumer客户端最小线程数 |
+| allocTaskStrategy | true | ✅ | ✅ | org.apache.rocketmq.connect.<br>runtime.service.strategy.<br>DefaultAllocateConnAndTaskStrategy | 负载均衡策略类 |
+| offsetCommitTimeoutMsConfig | true | ✅ | ✅ | 5000L | source和sink offset提交超时时间 |
+| offsetCommitIntervalMsConfig| true | ✅ | ✅ | 60000L | source和sink offset提交间隔时间配置 |
+| keyConverter | true | ✅ | ✅ | org.apache.rocketmq.connect.runtime.converter.record.StringConverter | 集群配置默认 key 转换器 |
+| valueConverter | true | ✅ | ✅ | org.apache.rocketmq.connect.runtime.converter.record.StringConverter | 集群配置默认 Value 转换器 |
### allocTaskStrategy说明
@@ -271,7 +406,7 @@ org.apache.rocketmq.connect.runtime.service.strategy.AllocateConnAndTaskStrategy
[负载均衡](https://rocketmq-1.gitbook.io/rocketmq-connector/rocketmq-connect-1/rocketmq-runtime/fu-zai-jun-heng)
-## 13.runtime支持JVM参数说明
+## 14.runtime支持JVM参数说明
| key | nullable | default | description |
| ----------------------------------------------- | -------- | ------- | ----------------------- |
@@ -330,37 +465,38 @@ private static class MD5Hash implements HashFunction {
}
```
-##开发指南
-如何在IDE中启动Connect Worker ?
+### 开发指南
-###单机模式启动Connect Worker
+> 如何在IDE中启动Connect Worker ?
-Main Class配置
-org.apache.rocketmq.connect.runtime.StandaloneConnectStartup
+### 单机模式启动Connect Worker
+
+Main Class配置:
+>org.apache.rocketmq.connect.runtime.StandaloneConnectStartup
Program arguments配置
--c ${user path}/rocketmq-connect/distribution/conf/connect-standalone.conf
+> -c ${user path}/rocketmq-connect/distribution/conf/connect-standalone.conf
Environment variables配置
-CONNECT_HOME=${user path}/rocketmq-connect/distribution
+> CONNECT_HOME=${user path}/rocketmq-connect/distribution
###集群模式启动Connect Worker
Main Class配置
-org.apache.rocketmq.connect.runtime.DistributedConnectStartup
+>org.apache.rocketmq.connect.runtime.DistributedConnectStartup
Program arguments配置
--c ${user path}/rocketmq-connect/distribution/conf/connect-distributed.conf
+>-c ${user path}/rocketmq-connect/distribution/conf/connect-distributed.conf
Environment variables配置
-CONNECT_HOME=${user path}/rocketmq-connect/distribution
+>CONNECT_HOME=${user path}/rocketmq-connect/distribution
diff --git a/connectors/rocketmq-connect-debezium/rocketmq-connect-debezium-mysql/README.md b/connectors/rocketmq-connect-debezium/rocketmq-connect-debezium-mysql/README.md
new file mode 100644
index 0000000..7e1dce1
--- /dev/null
+++ b/connectors/rocketmq-connect-debezium/rocketmq-connect-debezium-mysql/README.md
@@ -0,0 +1,28 @@
+```
+{
+ "connector.class":"org.apache.rocketmq.connect.debezium.mysql.DebeziumMysqlConnector",
+ "max.tasks":"1",
+ "connect.topicname":"debezium-mysql-source",
+ "kafka.transforms": "Unwrap",
+ "kafka.transforms.Unwrap.delete.handling.mode": "none",
+ "kafka.transforms.Unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
+ "kafka.transforms.Unwrap.add.headers": "op,source.db,source.table",
+ "database.history.name.srv.addr": "localhost:9876",
+ "database.history.rocketmq.topic": "db-history-debezium-topic",
+ "database.history.store.only.monitored.tables.ddl": true,
+ "include.schema.changes": false,
+ "database.server.name": "server-1",
+ "database.port": 3306,
+ "database.hostname": “localhost”,
+ "database.user": “*******”,
+ "database.password": “**********”,
+ "database.connectionTimeZone": "Asia/Shanghai",
+ "table.include.list": "testdb.testTable",
+ "max.batch.size": 5,
+ "database.include.list": "testdb",
+ "snapshot.mode": "when_needed",
+ "key.converter":"org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
+ "value.converter":"org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter"
+}
+
+```
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-jdbc/README.md b/connectors/rocketmq-connect-jdbc/README.md
index d295290..ddd6b3e 100644
--- a/connectors/rocketmq-connect-jdbc/README.md
+++ b/connectors/rocketmq-connect-jdbc/README.md
@@ -17,18 +17,17 @@ mvn clean package -Dmaven.test.skip=true
```
POST http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-jdbc-source-connector-name}
{
- "connector-class":"org.apache.rocketmq.connect.jdbc.connector.JdbcSourceConnector",
- "max-task":"2",
- "connection.url":"jdbc:mysql://XXXXXXXXX:3306/{sourceDbName}",
+ "connector.class":"org.apache.rocketmq.connect.jdbc.connector.JdbcSourceConnector",
+ "max.tasks":"2",
+ "connection.url":"jdbc:mysql://XXXXXXXXX:3306",
"connection.user":"*****",
"connection.password":"*****",
"table.whitelist":"db.table",
"mode": "incrementing",
"incrementing.column.name":"id",
"timestamp.initial": -1,
- "connect-topicname":"targetTopic",
- "key-converter":"org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
- "value-converter":"org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter"
+ "key.converter":"org.apache.rocketmq.connect.runtime.converter.JsonConverter",
+ "value.converter":"org.apache.rocketmq.connect.runtime.converter.JsonConverter"
}
```
@@ -46,8 +45,8 @@ POST http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-jdbc-sink-conne
"pk.fields":"id",
"pk.mode":"record_value",
"insert.mode":"UPSERT",
- "key-converter":"org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
- "value-converter":"org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter"
+ "key.converter":"org.apache.rocketmq.connect.runtime.converter.JsonConverter",
+ "value.converter":"org.apache.rocketmq.connect.runtime.converter.JsonConverter"
}
```
@@ -63,21 +62,22 @@ http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-jdbc-connector-name}/
* **jdbc-source-connector 参数说明**
-| KEY | TYPE | Must be filled | Description| Example
-|------------------------|----|---------|---------------|------------------|
-|connection.url | String | YES | source端 jdbc连接 | jdbc:mysql://XXXXXXXXX:3306|
-|connection.user | String | YES | source端 DB 用户名 | root |
-|connection.password | String | YES | source端 DB 密码 | root |
-|connection.attempts | String | YES | source端 DB连接重试次数 | 3 |
-|connection.backoff.ms | Long | YES | |
-|poll.interval.ms | Long | YES |拉取间隔时间 | 3000ms |
-|batch.max.rows | Integer | NO |每次拉取数量 | 300 |
-|mode | Integer | NO |拉取模式 | bulk、timestamp、incrementing、timestamp+incrementing |
-|incrementing.column.name | Integer | NO |增量字段，常用ID | id |
-|timestamp.column.name | String | YES |时间增量字段 | modified_time |
-|table.whitelist | String | YES |需要扫描的表 | db.table,db.table01 |
-|max-task | Integer | YES |任务数量，最大不能大于表的数量 | 2 |
-|source-record-converter | Integer | YES |data转换器 | org.apache.rocketmq.connect.runtime.converter.JsonConverter |
+| KEY | TYPE | Must be filled | Description | Example
+|--------------------------|----|---------|------------------|-------------------------------------------------------------|
+| connection.url | String | YES | source端 jdbc连接 | jdbc:mysql://XXXXXXXXX:3306 |
+| connection.user | String | YES | source端 DB 用户名 | root |
+| connection.password | String | YES | source端 DB 密码 | root |
+| connection.attempts | String | YES | source端 DB连接重试次数 | 3 |
+| connection.backoff.ms | Long | YES | |
+| poll.interval.ms | Long | YES | 拉取间隔时间 | 3000ms |
+| batch.max.rows | Integer | NO | 每次拉取数量 | 300 |
+| mode | Integer | NO | 拉取模式 | bulk、timestamp、incrementing、timestamp+incrementing |
+| incrementing.column.name | Integer | NO | 增量字段，常用ID | id |
+| timestamp.column.name | String | YES | 时间增量字段 | modified_time |
+| table.whitelist | String | YES | 需要扫描的表 | db.table,db.table01 |
+| max.tasks | Integer | YES | 任务数量，最大不能大于表的数量 | 默认是1 |
+| value.converter | Integer | YES | value转换器 | org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter |
+| key.converter | Integer | YES | key转换器 | org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter |
```
注：1.source拉取的数据写入到以表名自动创建的topic中，如果需要写入特定的topic中则需要指定"connect-topicname" 参数
@@ -86,19 +86,20 @@ http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-jdbc-connector-name}/
* **jdbc-sink-connector 参数说明**
-| KEY | TYPE | Must be filled | Description| Example
-|------------------------|----|---------|---------------|------------------|
-|connection.url | String | YES | sink端 jdbc连接 | jdbc:mysql://XXXXXXXXX:3306|
-|connection.user | String | YES | sink端 DB 用户名 | root |
-|connection.password | String | YES | sink端 DB 密码 | root |
-|connection.attempts | String | NO | sink端 DB连接重试次数 | 3 |
-|connection.backoff.ms | Long | NO | |
-|connect-topicname | Long | YES |监听的topic | topic-name |
-|pk.fields | String | NO |写入侧主键配置，用于更新使用 | id |
-|pk.mode | String | NO |获取主键的模式 | none、record_value |
-|insert.mode | Integer | YES |写入模式 | UPDATE、UPSERT、INSERT |
-|max-task | Integer | NO |任务数量 | 2 |
-|source-record-converter | Integer | YES |data转换器 | org.apache.rocketmq.connect.runtime.converter.JsonConverter |
+| KEY | TYPE | Must be filled | Description| Example
+|-------------------------|----|---------|---------------|------------------|
+| connection.url | String | YES | sink端 jdbc连接 | jdbc:mysql://XXXXXXXXX:3306|
+| connection.user | String | YES | sink端 DB 用户名 | root |
+| connection.password | String | YES | sink端 DB 密码 | root |
+| connection.attempts | String | NO | sink端 DB连接重试次数 | 3 |
+| connection.backoff.ms | Long | NO | |
+| connect-topicname | Long | YES |监听的topic | topic-name |
+| pk.fields | String | NO |写入侧主键配置，用于更新使用 | id |
+| pk.mode | String | NO |获取主键的模式 | none、record_value |
+| insert.mode | Integer | YES |写入模式 | UPDATE、UPSERT、INSERT |
+| max.tasks | Integer | NO |任务数量 | 2 |
+| value.converter | Integer | YES | value转换器 | org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter |
+| key.converter | Integer | YES | key转换器 | org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter |
```
注: openMLDB maven包的引入：
diff --git a/distribution/conf/connect-file-sink.properties b/distribution/conf/connect-file-sink.properties
index 6efe7e5..dc0d191 100644
--- a/distribution/conf/connect-file-sink.properties
+++ b/distribution/conf/connect-file-sink.properties
@@ -16,4 +16,4 @@
name=FileSinkConnector
connector.class=org.apache.rocketmq.connect.file.FileSinkConnector
file=test-sink.txt
-connect-topicname=TopicTest
\ No newline at end of file
+connect.topicnames=TopicTest
\ No newline at end of file
diff --git a/distribution/conf/connect-file-source.properties b/distribution/conf/connect-file-source.properties
index 5b3fc71..139f48e 100644
--- a/distribution/conf/connect-file-source.properties
+++ b/distribution/conf/connect-file-source.properties
@@ -16,4 +16,4 @@
name=FileSourceConnector
connector.class=org.apache.rocketmq.connect.file.FileSourceConnector
file=test-source.txt
-connect-topicname=TopicTest
\ No newline at end of file
+connect.topicname=TopicTest
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index 59ac233..19f5e4d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -123,6 +123,11 @@
<artifactId>rocketmq-client</artifactId>
<version>${rocketmq.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>${jackson.version}</version>
+ </dependency>
<dependency>
<groupId>org.reflections</groupId>
<artifactId>reflections</artifactId>
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/DistributedConnectStartup.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/DistributedConnectStartup.java
index 4e468a3..53e644b 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/DistributedConnectStartup.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/DistributedConnectStartup.java
@@ -32,7 +32,7 @@ import org.apache.commons.cli.Options;
import org.apache.commons.cli.PosixParser;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.controller.distributed.DistributedConfig;
import org.apache.rocketmq.connect.runtime.controller.distributed.DistributedConnectController;
import org.apache.rocketmq.connect.runtime.service.ClusterManagementService;
@@ -110,7 +110,7 @@ public class DistributedConnectStartup {
}
if (null == connectConfig.getConnectHome()) {
- System.out.printf("Please set the %s variable in your environment to match the location of the Connect installation", ConnectConfig.CONNECT_HOME_ENV);
+ System.out.printf("Please set the %s variable in your environment to match the location of the Connect installation", WorkerConfig.CONNECT_HOME_ENV);
System.exit(-2);
}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/StandaloneConnectStartup.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/StandaloneConnectStartup.java
index 3b61f53..ad1d64a 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/StandaloneConnectStartup.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/StandaloneConnectStartup.java
@@ -25,7 +25,7 @@ import org.apache.commons.cli.Options;
import org.apache.commons.cli.PosixParser;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.controller.isolation.Plugin;
import org.apache.rocketmq.connect.runtime.controller.standalone.StandaloneConfig;
import org.apache.rocketmq.connect.runtime.controller.standalone.StandaloneConnectController;
@@ -109,7 +109,7 @@ public class StandaloneConnectStartup {
}
if (null == connectConfig.getConnectHome()) {
- System.out.printf("Please set the %s variable in your environment to match the location of the Connect installation", ConnectConfig.CONNECT_HOME_ENV);
+ System.out.printf("Please set the %s variable in your environment to match the location of the Connect installation", WorkerConfig.CONNECT_HOME_ENV);
System.exit(-2);
}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/ConnectorConfig.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/ConnectorConfig.java
new file mode 100644
index 0000000..fcc7ba2
--- /dev/null
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/ConnectorConfig.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.rocketmq.connect.runtime.config;
+
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.errors.ConnectException;
+import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
+import org.apache.rocketmq.connect.runtime.errors.ToleranceType;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Define keys for connector and task configs.
+ */
+public class ConnectorConfig {
+ protected ConnectKeyValue config;
+
+ public ConnectorConfig(ConnectKeyValue config) {
+ this.config = config;
+ validate();
+ }
+
+ /**
+ * original config
+ *
+ * @return
+ */
+ public Map<String, String> originalConfig() {
+ return config.getProperties();
+ }
+
+ /**
+ * The full class name of a specific connector implements.
+ */
+ public static final String CONNECTOR_CLASS = "connector.class";
+
+ public static final String TASK_CLASS = "task.class";
+
+ public static final String TASK_ID = "task.id";
+
+ public static final String MAX_TASK = "max.tasks";
+ public static final int TASKS_MAX_DEFAULT = 1;
+
+ public static final String TASK_TYPE = "task.type";
+
+ public static final String CONNECTOR_DIRECT_ENABLE = "connector.direct.enable";
+ public static final String SOURCE_TASK_CLASS = "source.task.class";
+ public static final String SINK_TASK_CLASS = "sink.task.class";
+
+
+ /**
+ * Last updated time of the configuration.
+ */
+ public static final String UPDATE_TIMESTAMP = "update.timestamp";
+
+ /**
+ * The full class name of record converter. Which is used to parse {@link ConnectRecord} to/from byte[].
+ */
+ public static final String VALUE_CONVERTER = "value.converter";
+ public static final String KEY_CONVERTER = "key.converter";
+
+ public static final String TRANSFORMS = "transforms";
+
+ public static final String CONNECT_TIMESTAMP = "connect.timestamp";
+
+ public static final String HASH_FUNC = "consistentHashFunc";
+
+ public static final String VIRTUAL_NODE = "virtualNode";
+
+ public static final String ERRORS_LOG_INCLUDE_MESSAGES_CONFIG = "errors.log.include.messages";
+ public static final boolean ERRORS_LOG_INCLUDE_MESSAGES_DEFAULT = false;
+
+ public static final String ERRORS_LOG_ENABLE_CONFIG = "errors.log.enable";
+ public static final boolean ERRORS_LOG_ENABLE_DEFAULT = false;
+
+ // The default is 0, which means no retries will be attempted. Use -1 for infinite retries.
+ public static final String ERRORS_RETRY_TIMEOUT_CONFIG = "errors.retry.timeout";
+ public static final int ERRORS_RETRY_TIMEOUT_DEFAULT = 0;
+
+ public static final String ERRORS_RETRY_MAX_DELAY_CONFIG = "errors.retry.delay.max.ms";
+ public static final int ERRORS_RETRY_MAX_DELAY_DEFAULT = 60000;
+
+ public static final String ERRORS_TOLERANCE_CONFIG = "errors.tolerance";
+ public static final ToleranceType ERRORS_TOLERANCE_DEFAULT = ToleranceType.NONE;
+
+ /**
+ * The required key for all configurations.
+ */
+ public static final Set<String> REQUEST_CONFIG = new HashSet<String>() {
+ {
+ add(CONNECTOR_CLASS);
+ }
+ };
+
+ /**
+ * Maximum allowed message size in bytes, the default value is 4M.
+ */
+ public static final int MAX_MESSAGE_SIZE = Integer.parseInt(System.getProperty("rocketmq.runtime.max.message.size", "4194304"));
+
+
+ /**
+ * validate
+ */
+ public void validate() {
+ if (!config.containsKey(CONNECTOR_CLASS)) {
+ throw new ConnectException("Config connector.class cannot be empty");
+ }
+ }
+}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/RuntimeConfigDefine.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/RuntimeConfigDefine.java
deleted file mode 100644
index 7139596..0000000
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/RuntimeConfigDefine.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.rocketmq.connect.runtime.config;
-
-import io.openmessaging.connector.api.data.ConnectRecord;
-import java.util.HashSet;
-import java.util.Set;
-
-/**
- * Define keys for connector and task configs.
- */
-public class RuntimeConfigDefine {
-
- /**
- * The full class name of a specific connector implements.
- */
- public static final String CONNECTOR_CLASS = "connector-class";
-
- public static final String CONNECTOR_DIRECT_ENABLE = "connector-direct-enable";
-
- public static final String TASK_CLASS = "task-class";
-
- public static final String TASK_ID = "task-id";
-
- public static final String TASK_GROUP_ID = "task-group-id";
-
- public static final String TASK_TYPE = "task-type";
-
- public static final String SOURCE_TASK_CLASS = "source-task-class";
-
- public static final String SINK_TASK_CLASS = "sink-task-class";
-
- public static final String MAX_TASK = "max-task";
-
- public static final String CONNECTOR_ID = "connector-id";
-
- /**
- * Last updated time of the configuration.
- */
- public static final String UPDATE_TIMESTAMP = "update-timestamp";
-
- /**
- * The full class name of record converter. Which is used to parse {@link ConnectRecord} to/from byte[].
- */
- public static final String SOURCE_RECORD_CONVERTER = "value-converter";
- public static final String SOURCE_RECORD_CONVERTER_DEFAULT = "org.apache.rocketmq.connect.runtime.converter.record.StringConverter";
- public static final String SOURCE_RECORD_KEY_CONVERTER = "key-converter";
- public static final String SOURCE_RECORD_KEY_CONVERTER_DEFAULT = "org.apache.rocketmq.connect.runtime.converter.record.StringConverter";
-
- public static final String NAMESRV_ADDR = "namesrv-addr";
-
- public static final String RMQ_PRODUCER_GROUP = "rmq-producer-group";
-
- public static final String RMQ_CONSUMNER_GROUP = "rmq-consumer-group";
-
- public static final String OPERATION_TIMEOUT = "operation-timeout";
-
- public static final String HASH_FUNC = "consistentHashFunc";
-
- public static final String VIRTUAL_NODE = "virtualNode";
-
- public static final String CONNECT_SHARDINGKEY = "connect-shardingkey";
-
- public static final String CONNECT_TOPICNAME = "connect-topicname";
-
- public static final String CONNECT_TOPICNAMES = "connect-topicnames";
-
- public static final String CONNECT_SOURCE_PARTITION = "connect-source-partition";
-
- public static final String CONNECT_SOURCE_POSITION = "connect-source-position";
-
- public static final String CONNECT_ENTRYTYPE = "connect-entrytype";
-
- public static final String CONNECT_TIMESTAMP = "connect-timestamp";
-
- public static final String CONNECT_SCHEMA = "connect-schema";
-
- public static final String TRANSFORMS = "transforms";
-
- /**
- * The required key for all configurations.
- */
- public static final Set<String> REQUEST_CONFIG = new HashSet<String>() {
- {
- add(CONNECTOR_CLASS);
- add(CONNECT_TOPICNAME);
- }
- };
-
- /**
- * Maximum allowed message size in bytes, the default value is 4M.
- */
- public static final int MAX_MESSAGE_SIZE = Integer.parseInt(System.getProperty("rocketmq.runtime.max.message.size", "4194304"));
-
-}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/SinkConnectorConfig.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/SinkConnectorConfig.java
index b513078..26f16e3 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/SinkConnectorConfig.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/SinkConnectorConfig.java
@@ -19,18 +19,56 @@
package org.apache.rocketmq.connect.runtime.config;
import com.google.common.base.Splitter;
+
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+
+import io.openmessaging.connector.api.errors.ConnectException;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
+import org.apache.rocketmq.connect.runtime.errors.DeadLetterQueueConfig;
+
+/**
+ * sink connector config
+ */
+public class SinkConnectorConfig extends ConnectorConfig {
+
+ public static final String SEMICOLON = ";";
+
+
+ public static final String CONNECT_TOPICNAMES = "connect.topicnames";
+
+ public static final String TASK_GROUP_ID = "task.group.id";
+
+
+ /**
+ * dead letter queue config
+ */
+ public static final String DLQ_PREFIX = "errors.deadletterqueue.";
+ public static final String DLQ_TOPIC_NAME_CONFIG = DLQ_PREFIX + "topic.name";
-public class SinkConnectorConfig extends ConnectConfig {
+ public static final String DLQ_TOPIC_READ_QUEUE_NUMS = DLQ_PREFIX + "read.queue.nums";
+ public static final short DLQ_TOPIC_READ_QUEUE_NUMS_DEFAULT = 8;
- public static Set<String> parseTopicList(ConnectKeyValue taskConfig) {
- String messageQueueStr = taskConfig.getString(RuntimeConfigDefine.CONNECT_TOPICNAME);
+ public static final String DLQ_TOPIC_WRITE_QUEUE_NUMS = DLQ_PREFIX + "write.queue.nums";
+ public static final short DLQ_TOPIC_WRITE_QUEUE_NUMS_DEFAULT = 8;
+
+ public static final String DLQ_CONTEXT_PROPERTIES_ENABLE_CONFIG = DLQ_PREFIX + "context.properties.enable";
+ public static final boolean DLQ_CONTEXT_PROPERTIES_ENABLE_DEFAULT = false;
+
+ public SinkConnectorConfig(ConnectKeyValue config) {
+ super(config);
+ }
+
+ public DeadLetterQueueConfig parseDeadLetterQueueConfig() {
+ return new DeadLetterQueueConfig(super.config);
+ }
+
+ public Set<String> parseTopicList() {
+ String messageQueueStr = this.config.getString(SinkConnectorConfig.CONNECT_TOPICNAMES);
if (StringUtils.isBlank(messageQueueStr)) {
return null;
}
@@ -45,4 +83,15 @@ public class SinkConnectorConfig extends ConnectConfig {
}
return new MessageQueue(messageQueueStrList.get(0), messageQueueStrList.get(1), Integer.valueOf(messageQueueStrList.get(2)));
}
+
+ /**
+ * validate
+ */
+ @Override
+ public void validate() {
+ super.validate();
+ if (!config.containsKey(CONNECT_TOPICNAMES)) {
+ throw new ConnectException("Config connect.topicnames cannot be empty");
+ }
+ }
}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/SourceConnectorConfig.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/SourceConnectorConfig.java
new file mode 100644
index 0000000..7fc12ec
--- /dev/null
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/SourceConnectorConfig.java
@@ -0,0 +1,16 @@
+package org.apache.rocketmq.connect.runtime.config;
+
+import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
+
+/**
+ * source connector config
+ */
+public class SourceConnectorConfig extends ConnectorConfig {
+
+ public static final String CONNECT_TOPICNAME = "connect.topicname";
+
+ public SourceConnectorConfig(ConnectKeyValue config) {
+ super(config);
+ }
+
+}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/ConnectConfig.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/WorkerConfig.java
similarity index 78%
rename from rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/ConnectConfig.java
rename to rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/WorkerConfig.java
index b4d63e9..69bcacf 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/ConnectConfig.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/WorkerConfig.java
@@ -28,16 +28,11 @@ import static org.apache.rocketmq.connect.runtime.common.LoggerName.ROCKETMQ_RUN
/**
* Configurations for runtime.
*/
-public class ConnectConfig {
+public class WorkerConfig {
private static final Logger log = LoggerFactory.getLogger(ROCKETMQ_RUNTIME);
-
public static final String CONNECT_HOME_PROPERTY = "connect.home.dir";
-
public static final String CONNECT_HOME_ENV = "CONNECT_HOME";
-
- public static final String COMMA = ",";
-
- public static final String SEMICOLON = ";";
+ private String connectHome = System.getProperty(CONNECT_HOME_PROPERTY, System.getenv(CONNECT_HOME_ENV));
/**
* The unique ID of each worker instance in the cluster
@@ -45,55 +40,95 @@ public class ConnectConfig {
private String workerId = "DefaultWorker";
/**
- * Storage directory for file store.
+ * Group listening for cluster changes
*/
- private String storePathRootDir = System.getProperty("user.home") + File.separator + "connectorStore";
-
- private String connectHome = System.getProperty(CONNECT_HOME_PROPERTY, System.getenv(CONNECT_HOME_ENV));
+ private String connectClusterId = "DefaultConnectCluster";
+ /**
+ * config example:
+ * namesrvAddr = localhost:9876
+ */
private String namesrvAddr = System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY, System.getenv(MixAll.NAMESRV_ADDR_ENV));
- private String rmqProducerGroup = "connector-producer-group";
-
- private int maxMessageSize;
-
- private int operationTimeout = 3000;
-
- private String rmqConsumerGroup = "connector-consumer-group";
+ /**
+ * Http port for REST API.
+ */
+ private int httpPort = 8082;
- private int rmqMaxRedeliveryTimes;
+ /**
+ * plugin paths config;
+ * Multiple use ',' split
+ * config example:
+ * pluginPaths = /tmp/plugins/,/app/plugins/
+ */
+ private String pluginPaths;
- private int rmqMessageConsumeTimeout = 3000;
+ /**
+ * rocketmq access control config;
+ * config example:
+ * aclEnable = true
+ * accessKey = 12345
+ * secretKey = 11111
+ */
+ private boolean aclEnable = false;
+ private String accessKey;
+ private String secretKey;
- private int rmqMaxConsumeThreadNums = 32;
+ /**
+ * auto create group enable config:
+ * config example:
+ * autoCreateGroupEnable = false
+ */
+ private boolean autoCreateGroupEnable = false;
- private int rmqMinConsumeThreadNums = 1;
+ /**
+ * Configure cluster converter
+ */
+ private String keyConverter = "org.apache.rocketmq.connect.runtime.converter.record.StringConverter";
+ private String valueConverter = "org.apache.rocketmq.connect.runtime.converter.record.StringConverter";
- /** Task start timeout mills, default 3 minute */
- private long maxStartTimeoutMills = 1000 * 60 * 3;
+ /**
+ * Storage directory for file store.
+ * config example:
+ * storePathRootDir = /tmp/storeRoot
+ */
+ private String storePathRootDir = System.getProperty("user.home") + File.separator + "connectorStore";
- // task stop timeout mills, default 1 minute
- private long maxStopTimeoutMills = 1000 * 60;
+ private int maxMessageSize;
- // offset commit timeout
- private long offsetCommitTimeoutMsConfig = 1000 * 30;
+ private int operationTimeout = 3000;
- private int brokerSuspendMaxTimeMillis = 300;
/**
* Default topic to send/consume online or offline message.
*/
private String clusterStoreTopic = "connector-cluster-topic";
+ /**
+ * Task allocated strategy
+ * config example with default:
+ * allocTaskStrategy
+ */
+ private String allocTaskStrategy = "org.apache.rocketmq.connect.runtime.service.strategy.DefaultAllocateConnAndTaskStrategy";
+
/**
* Default topic to send/consume config change message.
*/
private String configStoreTopic = "connector-config-topic";
+ /**
+ * Connector configuration persistence interval.
+ */
+ private int configPersistInterval = 20 * 1000;
+
/**
* Default topic to send/consume position change message.
*/
private String positionStoreTopic = "connector-position-topic";
+ /**
+ * Source task position persistence interval.
+ */
+ private int positionPersistInterval = 20 * 1000;
/**
* Default topic to send/consume offset change message.
@@ -101,54 +136,77 @@ public class ConnectConfig {
private String offsetStoreTopic = "connector-offset-topic";
/**
- * Default topic to send/consume offset change message.
+ * Sink task offset persistence interval.
*/
- private String connectStatusTopic = "connect-status-topic";
+ private int offsetPersistInterval = 20 * 1000;
/**
- * Http port for REST API.
+ * Default topic to send/consume state change message.
*/
- private int httpPort = 8082;
+ private String connectStatusTopic = "connect-status-topic";
/**
- * Source task position persistence interval.
+ * Connector state persistence interval.
*/
- private int positionPersistInterval = 20 * 1000;
+ private int statePersistInterval = 20 * 1000;
+
/**
- * Sink task offset persistence interval.
+ * RocketMq store topic producer group
+ * config example with default:
+ * rmqProducerGroup = connector-producer-group;
*/
- private int offsetPersistInterval = 20 * 1000;
-
+ private String rmqProducerGroup = "connector-producer-group";
/**
- * Connector configuration persistence interval.
+ * RocketMq store topic consumer group
+ * config example with default:
+ * rmqConsumerGroup = connector-consumer-group;
*/
- private int configPersistInterval = 20 * 1000;
-
- private String pluginPaths;
+ private String rmqConsumerGroup = "connector-consumer-group";
- private String connectClusterId = "DefaultConnectCluster";
+ /**
+ * RocketMq admin group
+ * config example with default:
+ * adminExtGroup = connector-admin-group;
+ */
+ private String adminExtGroup = "connector-admin-group";
- private String allocTaskStrategy = "org.apache.rocketmq.connect.runtime.service.strategy.DefaultAllocateConnAndTaskStrategy";
+ private int rmqMaxRedeliveryTimes;
- private boolean aclEnable = false;
+ private int rmqMessageConsumeTimeout = 3000;
+ private int rmqMaxConsumeThreadNums = 32;
+ private int rmqMinConsumeThreadNums = 1;
+ private int brokerSuspendMaxTimeMillis = 300;
- private String accessKey;
- private String secretKey;
+ /**
+ * Task start timeout mills, default 3 minute
+ * config example by default:
+ * maxStartTimeoutMills = 1000 * 60 * 3
+ */
+ private long maxStartTimeoutMills = 1000 * 60 * 3;
- private boolean autoCreateGroupEnable = false;
+ /**
+ * task stop timeout mills, default 1 minute
+ * config example by default:
+ * maxStopTimeoutMills = 1000 * 60
+ */
+ private long maxStopTimeoutMills = 1000 * 60;
- private String clusterName;
- private String adminExtGroup = "connector-admin-group";
+ /**
+ * offset commit timeout(ms)
+ * config example with default
+ * offsetCommitTimeoutMsConfig = 5000L
+ */
+ private long offsetCommitTimeoutMsConfig = 5000L;
/**
* offset commit interval ms
+ * config example with default
+ * offsetCommitIntervalMsConfig = 60000L
*/
- private long offsetCommitIntervalMs = 1000L * 30;
-
-
+ private long offsetCommitIntervalMsConfig = 60000L;
public String getWorkerId() {
@@ -367,14 +425,6 @@ public class ConnectConfig {
this.autoCreateGroupEnable = autoCreateGroupEnable;
}
- public String getClusterName() {
- return clusterName;
- }
-
- public void setClusterName(String clusterName) {
- this.clusterName = clusterName;
- }
-
public String getAdminExtGroup() {
return adminExtGroup;
}
@@ -392,12 +442,12 @@ public class ConnectConfig {
}
- public long getOffsetCommitIntervalMs() {
- return offsetCommitIntervalMs;
+ public long getOffsetCommitIntervalMsConfig() {
+ return offsetCommitIntervalMsConfig;
}
- public void setOffsetCommitIntervalMs(long offsetCommitIntervalMs) {
- this.offsetCommitIntervalMs = offsetCommitIntervalMs;
+ public void setOffsetCommitIntervalMsConfig(long offsetCommitIntervalMsConfig) {
+ this.offsetCommitIntervalMsConfig = offsetCommitIntervalMsConfig;
}
public long getMaxStartTimeoutMills() {
@@ -432,44 +482,74 @@ public class ConnectConfig {
this.brokerSuspendMaxTimeMillis = brokerSuspendMaxTimeMillis;
}
+ public boolean isAclEnable() {
+ return aclEnable;
+ }
+
+ public String getKeyConverter() {
+ return keyConverter;
+ }
+
+ public void setKeyConverter(String keyConverter) {
+ this.keyConverter = keyConverter;
+ }
+
+ public String getValueConverter() {
+ return valueConverter;
+ }
+
+ public void setValueConverter(String valueConverter) {
+ this.valueConverter = valueConverter;
+ }
+
+ public int getStatePersistInterval() {
+ return statePersistInterval;
+ }
+
+ public void setStatePersistInterval(int statePersistInterval) {
+ this.statePersistInterval = statePersistInterval;
+ }
+
@Override
public String toString() {
- return "ConnectConfig{" +
- "workerId='" + workerId + '\'' +
- ", storePathRootDir='" + storePathRootDir + '\'' +
- ", connectHome='" + connectHome + '\'' +
+ return "WorkerConfig{" +
+ "connectHome='" + connectHome + '\'' +
+ ", workerId='" + workerId + '\'' +
+ ", connectClusterId='" + connectClusterId + '\'' +
", namesrvAddr='" + namesrvAddr + '\'' +
- ", rmqProducerGroup='" + rmqProducerGroup + '\'' +
+ ", httpPort=" + httpPort +
+ ", pluginPaths='" + pluginPaths + '\'' +
+ ", aclEnable=" + aclEnable +
+ ", accessKey='" + accessKey + '\'' +
+ ", secretKey='" + secretKey + '\'' +
+ ", autoCreateGroupEnable=" + autoCreateGroupEnable +
+ ", keyConverter='" + keyConverter + '\'' +
+ ", valueConverter='" + valueConverter + '\'' +
+ ", storePathRootDir='" + storePathRootDir + '\'' +
", maxMessageSize=" + maxMessageSize +
", operationTimeout=" + operationTimeout +
+ ", clusterStoreTopic='" + clusterStoreTopic + '\'' +
+ ", allocTaskStrategy='" + allocTaskStrategy + '\'' +
+ ", configStoreTopic='" + configStoreTopic + '\'' +
+ ", configPersistInterval=" + configPersistInterval +
+ ", positionStoreTopic='" + positionStoreTopic + '\'' +
+ ", positionPersistInterval=" + positionPersistInterval +
+ ", offsetStoreTopic='" + offsetStoreTopic + '\'' +
+ ", offsetPersistInterval=" + offsetPersistInterval +
+ ", connectStatusTopic='" + connectStatusTopic + '\'' +
+ ", statePersistInterval=" + statePersistInterval +
+ ", rmqProducerGroup='" + rmqProducerGroup + '\'' +
", rmqConsumerGroup='" + rmqConsumerGroup + '\'' +
+ ", adminExtGroup='" + adminExtGroup + '\'' +
", rmqMaxRedeliveryTimes=" + rmqMaxRedeliveryTimes +
", rmqMessageConsumeTimeout=" + rmqMessageConsumeTimeout +
", rmqMaxConsumeThreadNums=" + rmqMaxConsumeThreadNums +
", rmqMinConsumeThreadNums=" + rmqMinConsumeThreadNums +
+ ", brokerSuspendMaxTimeMillis=" + brokerSuspendMaxTimeMillis +
", maxStartTimeoutMills=" + maxStartTimeoutMills +
", maxStopTimeoutMills=" + maxStopTimeoutMills +
", offsetCommitTimeoutMsConfig=" + offsetCommitTimeoutMsConfig +
- ", brokerSuspendMaxTimeMillis=" + brokerSuspendMaxTimeMillis +
- ", clusterStoreTopic='" + clusterStoreTopic + '\'' +
- ", configStoreTopic='" + configStoreTopic + '\'' +
- ", positionStoreTopic='" + positionStoreTopic + '\'' +
- ", offsetStoreTopic='" + offsetStoreTopic + '\'' +
- ", connectStatusTopic='" + connectStatusTopic + '\'' +
- ", httpPort=" + httpPort +
- ", positionPersistInterval=" + positionPersistInterval +
- ", offsetPersistInterval=" + offsetPersistInterval +
- ", configPersistInterval=" + configPersistInterval +
- ", pluginPaths='" + pluginPaths + '\'' +
- ", connectClusterId='" + connectClusterId + '\'' +
- ", allocTaskStrategy='" + allocTaskStrategy + '\'' +
- ", aclEnable=" + aclEnable +
- ", accessKey='" + accessKey + '\'' +
- ", secretKey='" + secretKey + '\'' +
- ", autoCreateGroupEnable=" + autoCreateGroupEnable +
- ", clusterName='" + clusterName + '\'' +
- ", adminExtGroup='" + adminExtGroup + '\'' +
- ", offsetCommitIntervalMs=" + offsetCommitIntervalMs +
+ ", offsetCommitIntervalMsConfig=" + offsetCommitIntervalMsConfig +
'}';
}
-}
+}
\ No newline at end of file
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/SourceTaskOffsetCommitter.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/SourceTaskOffsetCommitter.java
index b069417..2eb7faf 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/SourceTaskOffsetCommitter.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/SourceTaskOffsetCommitter.java
@@ -19,7 +19,7 @@ package org.apache.rocketmq.connect.runtime.connectorwrapper;
import io.openmessaging.connector.api.errors.ConnectException;
import org.apache.rocketmq.common.utils.ThreadUtils;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.utils.ConnectorTaskId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,12 +39,12 @@ import java.util.concurrent.TimeUnit;
class SourceTaskOffsetCommitter {
private static final Logger log = LoggerFactory.getLogger(SourceTaskOffsetCommitter.class);
- private final ConnectConfig config;
+ private final WorkerConfig config;
private final ScheduledExecutorService commitExecutorService;
private final ConcurrentMap<ConnectorTaskId, ScheduledFuture<?>> committers;
- SourceTaskOffsetCommitter(ConnectConfig config,
+ SourceTaskOffsetCommitter(WorkerConfig config,
ScheduledExecutorService commitExecutorService,
ConcurrentMap<ConnectorTaskId, ScheduledFuture<?>> committers) {
this.config = config;
@@ -52,7 +52,7 @@ class SourceTaskOffsetCommitter {
this.committers = committers;
}
- public SourceTaskOffsetCommitter(ConnectConfig config) {
+ public SourceTaskOffsetCommitter(WorkerConfig config) {
this(config, Executors.newSingleThreadScheduledExecutor(ThreadUtils.newGenericThreadFactory(
SourceTaskOffsetCommitter.class.getSimpleName(), false)),
new ConcurrentHashMap<>());
@@ -70,7 +70,7 @@ class SourceTaskOffsetCommitter {
}
public void schedule(final ConnectorTaskId id, final WorkerSourceTask workerTask) {
- long commitIntervalMs = config.getOffsetCommitIntervalMs();
+ long commitIntervalMs = config.getOffsetCommitIntervalMsConfig();
ScheduledFuture<?> commitFuture = commitExecutorService.scheduleWithFixedDelay(() -> {
commit(workerTask);
}, commitIntervalMs, commitIntervalMs, TimeUnit.MILLISECONDS);
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/TransformChain.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/TransformChain.java
index c6e46e2..b487d02 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/TransformChain.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/TransformChain.java
@@ -27,7 +27,7 @@ import io.openmessaging.internal.DefaultKeyValue;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.apache.rocketmq.connect.runtime.controller.isolation.Plugin;
import org.apache.rocketmq.connect.runtime.errors.ErrorReporter;
import org.apache.rocketmq.connect.runtime.errors.RetryWithToleranceOperator;
@@ -56,7 +56,7 @@ public class TransformChain<R extends ConnectRecord> implements AutoCloseable {
private static final String COMMA = ",";
- private static final String PREFIX = RuntimeConfigDefine.TRANSFORMS + "-";
+ private static final String PREFIX = ConnectorConfig.TRANSFORMS + ".";
private RetryWithToleranceOperator retryWithToleranceOperator;
@@ -75,7 +75,7 @@ public class TransformChain<R extends ConnectRecord> implements AutoCloseable {
}
private void init() {
- String transformsStr = config.getString(RuntimeConfigDefine.TRANSFORMS);
+ String transformsStr = config.getString(ConnectorConfig.TRANSFORMS);
if (StringUtils.isBlank(transformsStr)) {
log.warn("no transforms config, {}", JSON.toJSONString(config));
return;
@@ -86,7 +86,7 @@ public class TransformChain<R extends ConnectRecord> implements AutoCloseable {
return;
}
transformList.stream().forEach(transformStr -> {
- String transformClassKey = PREFIX + transformStr + "-class";
+ String transformClassKey = PREFIX + transformStr + ".class";
String transformClass = config.getString(transformClassKey);
try {
Transform transform = newTransform(transformClass);
@@ -94,7 +94,7 @@ public class TransformChain<R extends ConnectRecord> implements AutoCloseable {
Set<String> configKeys = config.keySet();
for (String key : configKeys) {
if (key.startsWith(PREFIX + transformStr) && !key.equals(transformClassKey)) {
- String originKey = key.replace(PREFIX + transformStr + "-", "");
+ String originKey = key.replace(PREFIX + transformStr + ".", "");
transformConfig.put(originKey, config.getString(key));
}
}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java
index 07b5351..bcd889b 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java
@@ -32,8 +32,8 @@ import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.status.WrapperStatusListener;
import org.apache.rocketmq.connect.runtime.controller.AbstractConnectController;
import org.apache.rocketmq.connect.runtime.controller.isolation.Plugin;
@@ -112,7 +112,7 @@ public class Worker {
private final PositionManagementService positionManagementService;
private Optional<SourceTaskOffsetCommitter> sourceTaskOffsetCommitter;
- private final ConnectConfig workerConfig;
+ private final WorkerConfig workerConfig;
private final Plugin plugin;
/**
@@ -129,7 +129,7 @@ public class Worker {
private final ConfigManagementService configManagementService;
- public Worker(ConnectConfig workerConfig,
+ public Worker(WorkerConfig workerConfig,
PositionManagementService positionManagementService,
ConfigManagementService configManagementService,
Plugin plugin,
@@ -359,7 +359,7 @@ public class Worker {
ClassLoader savedLoader = plugin.currentThreadLoader();
try {
ConnectKeyValue keyValue = newConnectors.get(connectorName);
- String connectorClass = keyValue.getString(RuntimeConfigDefine.CONNECTOR_CLASS);
+ String connectorClass = keyValue.getString(ConnectorConfig.CONNECTOR_CLASS);
ClassLoader connectorLoader = plugin.delegatingLoader().pluginClassLoader(connectorClass);
savedLoader = Plugin.compareAndSwapLoaders(connectorLoader);
@@ -782,9 +782,9 @@ public class Worker {
private void startTask(Map<String, List<ConnectKeyValue>> newTasks) throws Exception {
for (String connectorName : newTasks.keySet()) {
for (ConnectKeyValue keyValue : newTasks.get(connectorName)) {
- int taskId = keyValue.getInt(RuntimeConfigDefine.TASK_ID);
+ int taskId = keyValue.getInt(ConnectorConfig.TASK_ID);
ConnectorTaskId id = new ConnectorTaskId(connectorName, taskId);
- String taskType = keyValue.getString(RuntimeConfigDefine.TASK_TYPE);
+ String taskType = keyValue.getString(ConnectorConfig.TASK_TYPE);
if (TaskType.DIRECT.name().equalsIgnoreCase(taskType)) {
createDirectTask(id, keyValue);
continue;
@@ -792,27 +792,27 @@ public class Worker {
ClassLoader savedLoader = plugin.currentThreadLoader();
try {
- String connType = keyValue.getString(RuntimeConfigDefine.CONNECTOR_CLASS);
+ String connType = keyValue.getString(ConnectorConfig.CONNECTOR_CLASS);
ClassLoader connectorLoader = plugin.delegatingLoader().connectorLoader(connType);
savedLoader = Plugin.compareAndSwapLoaders(connectorLoader);
// new task
- final Class<? extends Task> taskClass = plugin.currentThreadLoader().loadClass(keyValue.getString(RuntimeConfigDefine.TASK_CLASS)).asSubclass(Task.class);
+ final Class<? extends Task> taskClass = plugin.currentThreadLoader().loadClass(keyValue.getString(ConnectorConfig.TASK_CLASS)).asSubclass(Task.class);
final Task task = plugin.newTask(taskClass);
/**
* create key/value converter
*/
- RecordConverter valueConverter = plugin.newConverter(keyValue, RuntimeConfigDefine.SOURCE_RECORD_CONVERTER, Plugin.ClassLoaderUsage.CURRENT_CLASSLOADER);
- RecordConverter keyConverter = plugin.newConverter(keyValue, RuntimeConfigDefine.SOURCE_RECORD_KEY_CONVERTER, Plugin.ClassLoaderUsage.CURRENT_CLASSLOADER);
+ RecordConverter valueConverter = plugin.newConverter(keyValue, ConnectorConfig.VALUE_CONVERTER, workerConfig.getValueConverter(), Plugin.ClassLoaderUsage.CURRENT_CLASSLOADER);
+ RecordConverter keyConverter = plugin.newConverter(keyValue, ConnectorConfig.KEY_CONVERTER, workerConfig.getKeyConverter(), Plugin.ClassLoaderUsage.CURRENT_CLASSLOADER);
if (keyConverter == null) {
- keyConverter = plugin.newConverter(keyValue, RuntimeConfigDefine.SOURCE_RECORD_KEY_CONVERTER, Plugin.ClassLoaderUsage.PLUGINS);
+ keyConverter = plugin.newConverter(keyValue, ConnectorConfig.KEY_CONVERTER, workerConfig.getValueConverter(), Plugin.ClassLoaderUsage.PLUGINS);
log.info("Set up the key converter {} for task {} using the worker config", keyConverter.getClass(), id);
} else {
log.info("Set up the key converter {} for task {} using the connector config", keyConverter.getClass(), id);
}
if (valueConverter == null) {
- valueConverter = plugin.newConverter(keyValue, RuntimeConfigDefine.SOURCE_RECORD_CONVERTER, Plugin.ClassLoaderUsage.PLUGINS);
+ valueConverter = plugin.newConverter(keyValue, ConnectorConfig.VALUE_CONVERTER, workerConfig.getKeyConverter(), Plugin.ClassLoaderUsage.PLUGINS);
log.info("Set up the value converter {} for task {} using the worker config", valueConverter.getClass(), id);
} else {
log.info("Set up the value converter {} for task {} using the connector config", valueConverter.getClass(), id);
@@ -884,10 +884,10 @@ public class Worker {
}
private void createDirectTask(ConnectorTaskId id, ConnectKeyValue keyValue) throws Exception {
- String sourceTaskClass = keyValue.getString(RuntimeConfigDefine.SOURCE_TASK_CLASS);
+ String sourceTaskClass = keyValue.getString(ConnectorConfig.SOURCE_TASK_CLASS);
Task sourceTask = getTask(sourceTaskClass);
- String sinkTaskClass = keyValue.getString(RuntimeConfigDefine.SINK_TASK_CLASS);
+ String sinkTaskClass = keyValue.getString(ConnectorConfig.SINK_TASK_CLASS);
Task sinkTask = getTask(sinkTaskClass);
TransformChain<ConnectRecord> transformChain = new TransformChain<>(keyValue, plugin);
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerConnector.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerConnector.java
index 82bae4e..0ed4871 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerConnector.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerConnector.java
@@ -38,7 +38,7 @@ import java.util.concurrent.atomic.AtomicReference;
/**
* A wrapper of {@link Connector} for runtime.
*/
-public class WorkerConnector implements Runnable {
+public class WorkerConnector implements Runnable {
private static final Logger log = LoggerFactory.getLogger(WorkerConnector.class);
private static final String THREAD_NAME_PREFIX = "connector-thread-";
@@ -129,7 +129,7 @@ public class WorkerConnector implements Runnable {
@Override
public void run() {
- try {
+ try {
ClassLoader savedLoader = Plugin.compareAndSwapLoaders(classloader);
String savedName = Thread.currentThread().getName();
try {
@@ -162,7 +162,8 @@ public class WorkerConnector implements Runnable {
}
try {
wait();
- } catch (InterruptedException e) {}
+ } catch (InterruptedException e) {
+ }
}
}
doShutdown();
@@ -346,6 +347,7 @@ public class WorkerConnector implements Runnable {
/**
* connector name
+ *
* @return
*/
public String getConnectorName() {
@@ -354,6 +356,7 @@ public class WorkerConnector implements Runnable {
/**
* connector config
+ *
* @return
*/
public ConnectKeyValue getKeyValue() {
@@ -376,7 +379,7 @@ public class WorkerConnector implements Runnable {
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append("connectorName:" + connectorName)
- .append("\nConfigs:" + JSON.toJSONString(keyValue));
+ .append("\nConfigs:" + JSON.toJSONString(keyValue));
return sb.toString();
}
}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerDirectTask.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerDirectTask.java
index f4e24b5..efbeaf4 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerDirectTask.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerDirectTask.java
@@ -29,7 +29,7 @@ import io.openmessaging.connector.api.data.RecordPartition;
import io.openmessaging.connector.api.storage.OffsetStorageReader;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.status.WrapperStatusListener;
import org.apache.rocketmq.connect.runtime.errors.RetryWithToleranceOperator;
import org.apache.rocketmq.connect.runtime.service.PositionManagementService;
@@ -65,7 +65,7 @@ public class WorkerDirectTask extends WorkerSourceTask {
private SinkTask sinkTask;
private final OffsetStorageReader positionStorageReader;
- public WorkerDirectTask(ConnectConfig workerConfig,
+ public WorkerDirectTask(WorkerConfig workerConfig,
ConnectorTaskId id,
SourceTask sourceTask,
ClassLoader classLoader,
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSinkTask.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSinkTask.java
index 809e517..14dc9cb 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSinkTask.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSinkTask.java
@@ -39,8 +39,8 @@ import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.apache.rocketmq.connect.runtime.config.SinkConnectorConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.status.WrapperStatusListener;
import org.apache.rocketmq.connect.runtime.errors.ErrorReporter;
@@ -148,7 +148,7 @@ public class WorkerSinkTask extends WorkerTask {
}
};
- public WorkerSinkTask(ConnectConfig workerConfig,
+ public WorkerSinkTask(WorkerConfig workerConfig,
ConnectorTaskId id,
SinkTask sinkTask,
ClassLoader classLoader,
@@ -181,7 +181,7 @@ public class WorkerSinkTask extends WorkerTask {
this.messageBatch = new ArrayList<>();
// commit
- this.nextCommit = System.currentTimeMillis() + workerConfig.getOffsetCommitIntervalMs();
+ this.nextCommit = System.currentTimeMillis() + workerConfig.getOffsetCommitIntervalMsConfig();
this.committing = false;
this.commitSeqno = 0;
this.commitStarted = -1;
@@ -191,7 +191,7 @@ public class WorkerSinkTask extends WorkerTask {
protected void iteration() {
- final long offsetCommitIntervalMs = workerConfig.getOffsetCommitIntervalMs();
+ final long offsetCommitIntervalMs = workerConfig.getOffsetCommitIntervalMsConfig();
long now = System.currentTimeMillis();
// check committing
if (!committing && now >= nextCommit) {
@@ -448,7 +448,6 @@ public class WorkerSinkTask extends WorkerTask {
return msgs;
}
-
@Override
public void close() {
sinkTask.stop();
@@ -507,8 +506,8 @@ public class WorkerSinkTask extends WorkerTask {
private ConnectRecord convertMessages(MessageExt message) {
Map<String, String> properties = message.getProperties();
// timestamp
- String connectTimestamp = properties.get(RuntimeConfigDefine.CONNECT_TIMESTAMP);
- Long timestamp = StringUtils.isNotEmpty(connectTimestamp) ? Long.valueOf(connectTimestamp) : null;
+ String connectTimestamp = properties.get(ConnectorConfig.CONNECT_TIMESTAMP);
+ Long timestamp = StringUtils.isNotEmpty(connectTimestamp) ? Long.valueOf(connectTimestamp) : message.getBornTimestamp();
// partition and offset
RecordPartition recordPartition = ConnectUtil.convertToRecordPartition(message.getTopic(), message.getBrokerName(), message.getQueueId());
@@ -566,7 +565,7 @@ public class WorkerSinkTask extends WorkerTask {
*/
@Override
protected void initializeAndStart() {
- Set<String> topics = SinkConnectorConfig.parseTopicList(taskConfig);
+ Set<String> topics = new SinkConnectorConfig(taskConfig).parseTopicList();
if (org.apache.commons.collections4.CollectionUtils.isEmpty(topics)) {
throw new ConnectException("Sink connector topics config can be null, please check sink connector config info");
}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTask.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTask.java
index d480c02..28f2bdb 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTask.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTask.java
@@ -41,8 +41,9 @@ import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.SourceConnectorConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.status.WrapperStatusListener;
import org.apache.rocketmq.connect.runtime.errors.ErrorReporter;
import org.apache.rocketmq.connect.runtime.errors.RetryWithToleranceOperator;
@@ -132,7 +133,7 @@ public class WorkerSourceTask extends WorkerTask {
WHITE_KEY_SET.add(MessageConst.PROPERTY_TAGS);
}
- public WorkerSourceTask(ConnectConfig workerConfig,
+ public WorkerSourceTask(WorkerConfig workerConfig,
ConnectorTaskId id,
SourceTask sourceTask,
ClassLoader classLoader,
@@ -371,8 +372,8 @@ public class WorkerSourceTask extends WorkerTask {
byte[] value = retryWithToleranceOperator.execute(() -> valueConverter.fromConnectData(topic, record.getSchema(), record.getData()),
ErrorReporter.Stage.CONVERTER, valueConverter.getClass());
- if (value.length > RuntimeConfigDefine.MAX_MESSAGE_SIZE) {
- log.error("Send record, message size is greater than {} bytes, record: {}", RuntimeConfigDefine.MAX_MESSAGE_SIZE, JSON.toJSONString(record));
+ if (value.length > ConnectorConfig.MAX_MESSAGE_SIZE) {
+ log.error("Send record, message size is greater than {} bytes, record: {}", ConnectorConfig.MAX_MESSAGE_SIZE, JSON.toJSONString(record));
}
if (key != null) {
sourceMessage.setKeys(Base64Util.base64Encode(key));
@@ -393,7 +394,7 @@ public class WorkerSourceTask extends WorkerTask {
* @return
*/
private String maybeCreateAndGetTopic(ConnectRecord record) {
- String topic = taskConfig.getString(RuntimeConfigDefine.CONNECT_TOPICNAME);
+ String topic = taskConfig.getString(SourceConnectorConfig.CONNECT_TOPICNAME);
if (StringUtils.isBlank(topic)) {
RecordPosition recordPosition = record.getPosition();
if (null == recordPosition) {
@@ -433,6 +434,9 @@ public class WorkerSourceTask extends WorkerTask {
log.info("extension keySet null.");
return;
}
+ if (sourceDataEntry.getTimestamp() != null) {
+ MessageAccessor.putProperty(sourceMessage, ConnectorConfig.CONNECT_TIMESTAMP, sourceDataEntry.getTimestamp().toString());
+ }
for (String key : keySet) {
if (WHITE_KEY_SET.contains(key)) {
MessageAccessor.putProperty(sourceMessage, key, extensionKeyValues.getString(key));
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerTask.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerTask.java
index d60e23a..9dace23 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerTask.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerTask.java
@@ -18,7 +18,7 @@ package org.apache.rocketmq.connect.runtime.connectorwrapper;
import io.openmessaging.connector.api.data.ConnectRecord;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.status.TaskStatus;
import org.apache.rocketmq.connect.runtime.controller.isolation.Plugin;
import org.apache.rocketmq.connect.runtime.errors.RetryWithToleranceOperator;
@@ -39,7 +39,7 @@ public abstract class WorkerTask implements Runnable {
private static final Logger log = LoggerFactory.getLogger(WorkerTask.class);
private static final String THREAD_NAME_PREFIX = "task-thread-";
- protected final ConnectConfig workerConfig;
+ protected final WorkerConfig workerConfig;
protected final ConnectorTaskId id;
protected final ClassLoader loader;
@@ -61,7 +61,7 @@ public abstract class WorkerTask implements Runnable {
private final CountDownLatch shutdownLatch = new CountDownLatch(1);
- public WorkerTask(ConnectConfig workerConfig, ConnectorTaskId id, ClassLoader loader, ConnectKeyValue taskConfig, RetryWithToleranceOperator retryWithToleranceOperator, TransformChain<ConnectRecord> transformChain, AtomicReference<WorkerState> workerState, TaskStatus.Listener taskListener) {
+ public WorkerTask(WorkerConfig workerConfig, ConnectorTaskId id, ClassLoader loader, ConnectKeyValue taskConfig, RetryWithToleranceOperator retryWithToleranceOperator, TransformChain<ConnectRecord> transformChain, AtomicReference<WorkerState> workerState, TaskStatus.Listener taskListener) {
this.workerConfig = workerConfig;
this.id = id;
this.loader = loader;
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/AbstractConnectController.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/AbstractConnectController.java
index 69c8342..7b1f982 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/AbstractConnectController.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/AbstractConnectController.java
@@ -20,8 +20,8 @@ package org.apache.rocketmq.connect.runtime.controller;
import io.openmessaging.connector.api.errors.ConnectException;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.Worker;
import org.apache.rocketmq.connect.runtime.connectorwrapper.status.ConnectorStatus;
import org.apache.rocketmq.connect.runtime.connectorwrapper.status.TaskStatus;
@@ -59,7 +59,7 @@ public abstract class AbstractConnectController implements ConnectController {
/**
* Configuration of current runtime.
*/
- protected final ConnectConfig connectConfig;
+ protected final WorkerConfig connectConfig;
/**
* All the configurations of current running connectors and tasks in cluster.
@@ -107,7 +107,7 @@ public abstract class AbstractConnectController implements ConnectController {
*/
public AbstractConnectController(
Plugin plugin,
- ConnectConfig connectConfig,
+ WorkerConfig connectConfig,
ClusterManagementService clusterManagementService,
ConfigManagementService configManagementService,
PositionManagementService positionManagementService,
@@ -164,7 +164,7 @@ public abstract class AbstractConnectController implements ConnectController {
}
- public ConnectConfig getConnectConfig() {
+ public WorkerConfig getConnectConfig() {
return connectConfig;
}
@@ -286,7 +286,7 @@ public abstract class AbstractConnectController implements ConnectController {
Collections.sort(taskStates);
Map<String, String> conf = rawConfig(connName);
return new ConnectorStateInfo(connName, connectorState, taskStates,
- conf == null ? ConnectorType.UNKNOWN : connectorTypeForClass(conf.get(RuntimeConfigDefine.CONNECTOR_CLASS)));
+ conf == null ? ConnectorType.UNKNOWN : connectorTypeForClass(conf.get(ConnectorConfig.CONNECTOR_CLASS)));
}
@@ -309,7 +309,7 @@ public abstract class AbstractConnectController implements ConnectController {
connector,
config,
configState.tasks(connector),
- connectorTypeForClass(config.get(RuntimeConfigDefine.CONNECTOR_CLASS))
+ connectorTypeForClass(config.get(ConnectorConfig.CONNECTOR_CLASS))
);
}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/distributed/DistributedConfig.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/distributed/DistributedConfig.java
index 4784b21..9cdedba 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/distributed/DistributedConfig.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/distributed/DistributedConfig.java
@@ -17,10 +17,10 @@
package org.apache.rocketmq.connect.runtime.controller.distributed;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
/**
* distributed worker config
*/
-public class DistributedConfig extends ConnectConfig {
+public class DistributedConfig extends WorkerConfig {
}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/distributed/DistributedConnectController.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/distributed/DistributedConnectController.java
index 1872161..a93a5c2 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/distributed/DistributedConnectController.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/distributed/DistributedConnectController.java
@@ -93,7 +93,7 @@ public class DistributedConnectController extends AbstractConnectController {
}, 1000, this.connectConfig.getPositionPersistInterval(), TimeUnit.MILLISECONDS);
- // Persist position information of source tasks.
+ // Persist state information of connector
scheduledExecutorService.scheduleAtFixedRate(() -> {
try {
@@ -102,7 +102,7 @@ public class DistributedConnectController extends AbstractConnectController {
log.error("schedule persist position error.", e);
}
- }, 1000, this.connectConfig.getPositionPersistInterval(), TimeUnit.MILLISECONDS);
+ }, 1000, this.connectConfig.getStatePersistInterval(), TimeUnit.MILLISECONDS);
}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/isolation/Plugin.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/isolation/Plugin.java
index cc698a8..a4511d3 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/isolation/Plugin.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/isolation/Plugin.java
@@ -159,9 +159,9 @@ public class Plugin {
return newPlugin(taskClass);
}
- public RecordConverter newConverter(ConnectKeyValue config, String classPropertyName, ClassLoaderUsage classLoaderUsage) {
+ public RecordConverter newConverter(ConnectKeyValue config, String classPropertyName, String defaultConverter, ClassLoaderUsage classLoaderUsage) {
if (!config.containsKey(classPropertyName)) {
- return null;
+ config.put(classPropertyName, defaultConverter);
}
Class<? extends RecordConverter> klass = null;
switch (classLoaderUsage) {
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/standalone/StandaloneConfig.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/standalone/StandaloneConfig.java
index 967c72e..5f353c6 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/standalone/StandaloneConfig.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/standalone/StandaloneConfig.java
@@ -16,11 +16,11 @@
*/
package org.apache.rocketmq.connect.runtime.controller.standalone;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
/**
* worker standalone config
*
*/
-public class StandaloneConfig extends ConnectConfig {
+public class StandaloneConfig extends WorkerConfig {
}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/errors/DeadLetterQueueConfig.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/errors/DeadLetterQueueConfig.java
index 0a27494..e157e0c 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/errors/DeadLetterQueueConfig.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/errors/DeadLetterQueueConfig.java
@@ -19,42 +19,28 @@ package org.apache.rocketmq.connect.runtime.errors;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
+import static org.apache.rocketmq.connect.runtime.config.ConnectorConfig.ERRORS_LOG_ENABLE_CONFIG;
+import static org.apache.rocketmq.connect.runtime.config.ConnectorConfig.ERRORS_LOG_ENABLE_DEFAULT;
+import static org.apache.rocketmq.connect.runtime.config.ConnectorConfig.ERRORS_LOG_INCLUDE_MESSAGES_CONFIG;
+import static org.apache.rocketmq.connect.runtime.config.ConnectorConfig.ERRORS_LOG_INCLUDE_MESSAGES_DEFAULT;
+import static org.apache.rocketmq.connect.runtime.config.ConnectorConfig.ERRORS_RETRY_MAX_DELAY_CONFIG;
+import static org.apache.rocketmq.connect.runtime.config.ConnectorConfig.ERRORS_RETRY_MAX_DELAY_DEFAULT;
+import static org.apache.rocketmq.connect.runtime.config.ConnectorConfig.ERRORS_RETRY_TIMEOUT_CONFIG;
+import static org.apache.rocketmq.connect.runtime.config.ConnectorConfig.ERRORS_RETRY_TIMEOUT_DEFAULT;
+import static org.apache.rocketmq.connect.runtime.config.ConnectorConfig.ERRORS_TOLERANCE_CONFIG;
+import static org.apache.rocketmq.connect.runtime.config.ConnectorConfig.ERRORS_TOLERANCE_DEFAULT;
+import static org.apache.rocketmq.connect.runtime.config.SinkConnectorConfig.DLQ_CONTEXT_PROPERTIES_ENABLE_CONFIG;
+import static org.apache.rocketmq.connect.runtime.config.SinkConnectorConfig.DLQ_CONTEXT_PROPERTIES_ENABLE_DEFAULT;
+import static org.apache.rocketmq.connect.runtime.config.SinkConnectorConfig.DLQ_TOPIC_NAME_CONFIG;
+import static org.apache.rocketmq.connect.runtime.config.SinkConnectorConfig.DLQ_TOPIC_READ_QUEUE_NUMS;
+import static org.apache.rocketmq.connect.runtime.config.SinkConnectorConfig.DLQ_TOPIC_READ_QUEUE_NUMS_DEFAULT;
+import static org.apache.rocketmq.connect.runtime.config.SinkConnectorConfig.DLQ_TOPIC_WRITE_QUEUE_NUMS;
+import static org.apache.rocketmq.connect.runtime.config.SinkConnectorConfig.DLQ_TOPIC_WRITE_QUEUE_NUMS_DEFAULT;
+
/**
* dead letter queue config
*/
public class DeadLetterQueueConfig {
- /**
- * dead letter queue config
- */
- public static final String DLQ_PREFIX = "errors.deadletterqueue.";
- public static final String DLQ_TOPIC_NAME_CONFIG = DLQ_PREFIX + "topic.name";
-
-
- public static final String DLQ_TOPIC_READ_QUEUE_NUMS = DLQ_PREFIX + "read.queue.nums";
- public static final short DLQ_TOPIC_READ_QUEUE_NUMS_DEFAULT = 16;
-
- public static final String DLQ_TOPIC_WRITE_QUEUE_NUMS = DLQ_PREFIX + "write.queue.nums";
- public static final short DLQ_TOPIC_WRITE_QUEUE_NUMS_DEFAULT = 16;
-
- public static final String DLQ_CONTEXT_HEADERS_ENABLE_CONFIG = DLQ_PREFIX + "context.headers.enable";
- public static final boolean DLQ_CONTEXT_HEADERS_ENABLE_DEFAULT = false;
-
- public static final String ERRORS_LOG_INCLUDE_MESSAGES_CONFIG = "errors.log.include.messages";
- public static final boolean ERRORS_LOG_INCLUDE_MESSAGES_DEFAULT = false;
-
- public static final String ERRORS_LOG_ENABLE_CONFIG = "errors.log.enable";
- public static final boolean ERRORS_LOG_ENABLE_DEFAULT = false;
-
-
- public static final String ERRORS_RETRY_TIMEOUT_CONFIG = "errors.retry.timeout";
- public static final int ERRORS_RETRY_TIMEOUT_DEFAULT = 0;
-
- public static final String ERRORS_RETRY_MAX_DELAY_CONFIG = "errors.retry.delay.max.ms";
- public static final String ERRORS_RETRY_MAX_DELAY_DISPLAY = "Maximum Delay Between Retries for Errors";
- public static final int ERRORS_RETRY_MAX_DELAY_DEFAULT = 60000;
-
- public static final String ERRORS_TOLERANCE_CONFIG = "errors.tolerance";
- public static final ToleranceType ERRORS_TOLERANCE_DEFAULT = ToleranceType.NONE;
private ConnectKeyValue config;
@@ -84,8 +70,8 @@ public class DeadLetterQueueConfig {
* @return
*/
public Boolean isDlqContextHeadersEnabled() {
- return config.getProperties().containsKey(DLQ_CONTEXT_HEADERS_ENABLE_CONFIG) ?
- Boolean.getBoolean(config.getProperties().get(DLQ_CONTEXT_HEADERS_ENABLE_CONFIG)) : DLQ_CONTEXT_HEADERS_ENABLE_DEFAULT;
+ return config.getProperties().containsKey(DLQ_CONTEXT_PROPERTIES_ENABLE_CONFIG) ?
+ Boolean.valueOf(config.getProperties().get(DLQ_CONTEXT_PROPERTIES_ENABLE_CONFIG)) : DLQ_CONTEXT_PROPERTIES_ENABLE_DEFAULT;
}
@@ -115,7 +101,7 @@ public class DeadLetterQueueConfig {
*/
public boolean includeRecordDetailsInErrorLog() {
return config.getProperties().containsKey(ERRORS_LOG_INCLUDE_MESSAGES_CONFIG) ?
- Boolean.getBoolean(config.getProperties().get(ERRORS_LOG_INCLUDE_MESSAGES_CONFIG)) : ERRORS_LOG_INCLUDE_MESSAGES_DEFAULT;
+ Boolean.valueOf(config.getProperties().get(ERRORS_LOG_INCLUDE_MESSAGES_CONFIG)) : ERRORS_LOG_INCLUDE_MESSAGES_DEFAULT;
}
@@ -128,16 +114,16 @@ public class DeadLetterQueueConfig {
private boolean enableErrorLog() {
return config.getProperties().containsKey(ERRORS_LOG_ENABLE_CONFIG) ?
- Boolean.getBoolean(config.getProperties().get(ERRORS_LOG_ENABLE_CONFIG)) : ERRORS_LOG_ENABLE_DEFAULT;
+ Boolean.valueOf(config.getProperties().get(ERRORS_LOG_ENABLE_CONFIG)) : ERRORS_LOG_ENABLE_DEFAULT;
}
public long errorRetryTimeout() {
- return config.getLong(ERRORS_RETRY_TIMEOUT_CONFIG);
+ return config.getLong(ERRORS_RETRY_TIMEOUT_CONFIG, ERRORS_RETRY_TIMEOUT_DEFAULT);
}
public long errorMaxDelayInMillis() {
- return config.getLong(ERRORS_RETRY_MAX_DELAY_CONFIG);
+ return config.getLong(ERRORS_RETRY_MAX_DELAY_CONFIG, ERRORS_RETRY_MAX_DELAY_DEFAULT);
}
public ToleranceType errorToleranceType() {
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/errors/DeadLetterQueueReporter.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/errors/DeadLetterQueueReporter.java
index cf8967b..b4f0b91 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/errors/DeadLetterQueueReporter.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/errors/DeadLetterQueueReporter.java
@@ -27,7 +27,7 @@ import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.utils.ConnectUtil;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
@@ -84,11 +84,6 @@ public class DeadLetterQueueReporter implements ErrorReporter {
private String connectorName;
- public void workerId(String workerId) {
- this.workerId = workerId;
- }
-
-
/**
* build reporter
*
@@ -99,7 +94,7 @@ public class DeadLetterQueueReporter implements ErrorReporter {
*/
public static DeadLetterQueueReporter build(String connectorName,
ConnectKeyValue sinkConfig,
- ConnectConfig workerConfig) {
+ WorkerConfig workerConfig) {
DeadLetterQueueConfig deadLetterQueueConfig = new DeadLetterQueueConfig(sinkConfig);
String dlqTopic = deadLetterQueueConfig.dlqTopicName();
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/errors/ReporterManagerUtil.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/errors/ReporterManagerUtil.java
index 2da9122..a144911 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/errors/ReporterManagerUtil.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/errors/ReporterManagerUtil.java
@@ -19,7 +19,7 @@ package org.apache.rocketmq.connect.runtime.errors;
import io.openmessaging.connector.api.data.RecordConverter;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import java.util.ArrayList;
import java.util.List;
@@ -73,7 +73,7 @@ public class ReporterManagerUtil {
*/
public static List<ErrorReporter> sinkTaskReporters(String connectName,
ConnectKeyValue connConfig,
- ConnectConfig workerConfig) {
+ WorkerConfig workerConfig) {
// ensure reporter order
ArrayList<ErrorReporter> reporters = new ArrayList<>();
LogReporter logReporter = new LogReporter(connectName, connConfig);
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/AbstractConfigManagementService.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/AbstractConfigManagementService.java
index 3e6f166..fe23d68 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/AbstractConfigManagementService.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/AbstractConfigManagementService.java
@@ -20,7 +20,10 @@ package org.apache.rocketmq.connect.runtime.service;
import io.openmessaging.KeyValue;
import io.openmessaging.connector.api.component.connector.Connector;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
+import org.apache.rocketmq.connect.runtime.config.SinkConnectorConfig;
+import org.apache.rocketmq.connect.runtime.config.SourceConnectorConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.TargetState;
import org.apache.rocketmq.connect.runtime.connectorwrapper.Worker;
import org.apache.rocketmq.connect.runtime.controller.isolation.Plugin;
@@ -42,6 +45,8 @@ import java.util.concurrent.ConcurrentHashMap;
*/
public abstract class AbstractConfigManagementService implements ConfigManagementService {
+ protected WorkerConfig workerConfig;
+
protected Plugin plugin;
/**
@@ -57,9 +62,9 @@ public abstract class AbstractConfigManagementService implements ConfigManagemen
@Override
public void recomputeTaskConfigs(String connectorName, Connector connector, Long currentTimestamp, ConnectKeyValue configs) {
- int maxTask = configs.getInt(RuntimeConfigDefine.MAX_TASK, 1);
+ int maxTask = configs.getInt(ConnectorConfig.MAX_TASK, ConnectorConfig.TASKS_MAX_DEFAULT);
ConnectKeyValue connectConfig = connectorKeyValueStore.get(connectorName);
- boolean directEnable = Boolean.parseBoolean(connectConfig.getString(RuntimeConfigDefine.CONNECTOR_DIRECT_ENABLE));
+ boolean directEnable = Boolean.parseBoolean(connectConfig.getString(ConnectorConfig.CONNECTOR_DIRECT_ENABLE, "false"));
List<KeyValue> taskConfigs = connector.taskConfigs(maxTask);
List<ConnectKeyValue> converterdConfigs = new ArrayList<>();
int taskId = 0;
@@ -69,20 +74,19 @@ public abstract class AbstractConfigManagementService implements ConfigManagemen
newKeyValue.put(key, keyValue.getString(key));
}
if (directEnable) {
- newKeyValue.put(RuntimeConfigDefine.TASK_TYPE, Worker.TaskType.DIRECT.name());
- newKeyValue.put(RuntimeConfigDefine.SOURCE_TASK_CLASS, connectConfig.getString(RuntimeConfigDefine.SOURCE_TASK_CLASS));
- newKeyValue.put(RuntimeConfigDefine.SINK_TASK_CLASS, connectConfig.getString(RuntimeConfigDefine.SINK_TASK_CLASS));
+ newKeyValue.put(ConnectorConfig.TASK_TYPE, Worker.TaskType.DIRECT.name());
+ newKeyValue.put(ConnectorConfig.SOURCE_TASK_CLASS, connectConfig.getString(ConnectorConfig.SOURCE_TASK_CLASS));
+ newKeyValue.put(ConnectorConfig.SINK_TASK_CLASS, connectConfig.getString(ConnectorConfig.SINK_TASK_CLASS));
}
// put task id
- newKeyValue.put(RuntimeConfigDefine.TASK_ID, taskId);
- newKeyValue.put(RuntimeConfigDefine.TASK_CLASS, connector.taskClass().getName());
- newKeyValue.put(RuntimeConfigDefine.UPDATE_TIMESTAMP, currentTimestamp);
-
- newKeyValue.put(RuntimeConfigDefine.CONNECT_TOPICNAME, configs.getString(RuntimeConfigDefine.CONNECT_TOPICNAME));
- newKeyValue.put(RuntimeConfigDefine.CONNECT_TOPICNAMES, configs.getString(RuntimeConfigDefine.CONNECT_TOPICNAMES));
+ newKeyValue.put(ConnectorConfig.TASK_ID, taskId);
+ newKeyValue.put(ConnectorConfig.TASK_CLASS, connector.taskClass().getName());
+ newKeyValue.put(ConnectorConfig.UPDATE_TIMESTAMP, currentTimestamp);
+ newKeyValue.put(SourceConnectorConfig.CONNECT_TOPICNAME, configs.getString(SourceConnectorConfig.CONNECT_TOPICNAME));
+ newKeyValue.put(SinkConnectorConfig.CONNECT_TOPICNAMES, configs.getString(SinkConnectorConfig.CONNECT_TOPICNAMES));
Set<String> connectConfigKeySet = configs.keySet();
for (String connectConfigKey : connectConfigKeySet) {
- if (connectConfigKey.startsWith(RuntimeConfigDefine.TRANSFORMS)) {
+ if (connectConfigKey.startsWith(ConnectorConfig.TRANSFORMS)) {
newKeyValue.put(connectConfigKey, configs.getString(connectConfigKey));
}
}
@@ -97,7 +101,7 @@ public abstract class AbstractConfigManagementService implements ConfigManagemen
@NotNull
protected Connector loadConnector(ConnectKeyValue configs) {
- String connectorClass = configs.getString(RuntimeConfigDefine.CONNECTOR_CLASS);
+ String connectorClass = configs.getString(ConnectorConfig.CONNECTOR_CLASS);
Connector connector = plugin.newConnector(connectorClass);
connector.validate(configs);
connector.start(configs);
@@ -115,7 +119,7 @@ public abstract class AbstractConfigManagementService implements ConfigManagemen
taskKeyValueStore.getKVMap().forEach((connectorName, taskConfigs) -> {
connectorTaskCounts.put(connectorName, taskConfigs.size());
taskConfigs.forEach(taskConfig -> {
- ConnectorTaskId id = new ConnectorTaskId(connectorName, taskConfig.getInt(RuntimeConfigDefine.TASK_ID));
+ ConnectorTaskId id = new ConnectorTaskId(connectorName, taskConfig.getInt(ConnectorConfig.TASK_ID));
connectorTaskConfigs.put(id, taskConfig.getProperties());
});
});
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ClusterManagementService.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ClusterManagementService.java
index 346d5d0..68ec86f 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ClusterManagementService.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ClusterManagementService.java
@@ -18,14 +18,14 @@
package org.apache.rocketmq.connect.runtime.service;
import java.util.List;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
/**
* Interface for cluster management.
*/
public interface ClusterManagementService {
- void initialize(ConnectConfig connectConfig);
+ void initialize(WorkerConfig connectConfig);
/**
* Start the cluster manager.
@@ -42,7 +42,7 @@ public interface ClusterManagementService {
*
* @param config can be DistributedConfig or StandaloneConfig
*/
- default void configure(ConnectConfig config) {
+ default void configure(WorkerConfig config) {
}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ClusterManagementServiceImpl.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ClusterManagementServiceImpl.java
index 2d97737..2710432 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ClusterManagementServiceImpl.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ClusterManagementServiceImpl.java
@@ -28,7 +28,7 @@ import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.header.NotifyConsumerIdsChangedRequestHeader;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.utils.ConnectUtil;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
@@ -45,7 +45,7 @@ public class ClusterManagementServiceImpl implements ClusterManagementService {
/**
* Configs of current worker.
*/
- private ConnectConfig connectConfig;
+ private WorkerConfig connectConfig;
/**
* Used for worker discovery
@@ -60,7 +60,7 @@ public class ClusterManagementServiceImpl implements ClusterManagementService {
*
* @param connectConfig
*/
- private void prepare(ConnectConfig connectConfig) {
+ private void prepare(WorkerConfig connectConfig) {
String consumerGroup = this.defaultMQPullConsumer.getConsumerGroup();
Set<String> consumerGroupSet = ConnectUtil.fetchAllConsumerGroupList(connectConfig);
if (!consumerGroupSet.contains(consumerGroup)) {
@@ -76,7 +76,7 @@ public class ClusterManagementServiceImpl implements ClusterManagementService {
}
- @Override public void initialize(ConnectConfig connectConfig) {
+ @Override public void initialize(WorkerConfig connectConfig) {
this.connectConfig = connectConfig;
this.workerStatusListeners = new HashSet<>();
this.defaultMQPullConsumer = ConnectUtil.initDefaultMQPullConsumer(connectConfig);
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementService.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementService.java
index fe78452..e123957 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementService.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementService.java
@@ -19,7 +19,7 @@ package org.apache.rocketmq.connect.runtime.service;
import io.openmessaging.connector.api.component.connector.Connector;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.controller.isolation.Plugin;
import org.apache.rocketmq.connect.runtime.store.ClusterConfigState;
@@ -48,7 +48,7 @@ public interface ConfigManagementService {
*
* @param config can be DistributedConfig or StandaloneConfig
*/
- default void configure(ConnectConfig config) {
+ default void configure(WorkerConfig config) {
}
@@ -111,7 +111,7 @@ public interface ConfigManagementService {
*/
void registerListener(ConnectorConfigUpdateListener listener);
- void initialize(ConnectConfig connectConfig, Plugin plugin);
+ void initialize(WorkerConfig connectConfig, Plugin plugin);
ClusterConfigState snapshot();
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementServiceImpl.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementServiceImpl.java
index b2b4cf0..dcc8981 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementServiceImpl.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementServiceImpl.java
@@ -18,13 +18,17 @@
package org.apache.rocketmq.connect.runtime.service;
import io.openmessaging.connector.api.component.connector.Connector;
+import io.openmessaging.connector.api.component.task.sink.SinkConnector;
+import io.openmessaging.connector.api.component.task.source.SourceConnector;
import io.openmessaging.connector.api.errors.ConnectException;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.connect.runtime.common.ConnAndTaskConfigs;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.SinkConnectorConfig;
+import org.apache.rocketmq.connect.runtime.config.SourceConnectorConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.TargetState;
import org.apache.rocketmq.connect.runtime.controller.isolation.Plugin;
import org.apache.rocketmq.connect.runtime.converter.ConnectKeyValueConverter;
@@ -48,6 +52,8 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import static org.apache.rocketmq.connect.runtime.config.ConnectorConfig.CONNECTOR_CLASS;
+
public class ConfigManagementServiceImpl extends AbstractConfigManagementService {
private static final Logger log = LoggerFactory.getLogger(LoggerName.ROCKETMQ_RUNTIME);
@@ -88,24 +94,25 @@ public class ConfigManagementServiceImpl extends AbstractConfigManagementService
private final String configManagePrefix = "ConfigManage";
@Override
- public void initialize(ConnectConfig connectConfig, Plugin plugin) {
+ public void initialize(WorkerConfig workerConfig, Plugin plugin) {
this.connectorConfigUpdateListener = new HashSet<>();
- this.dataSynchronizer = new BrokerBasedLog<>(connectConfig,
- connectConfig.getConfigStoreTopic(),
- ConnectUtil.createGroupName(configManagePrefix, connectConfig.getWorkerId()),
+ this.dataSynchronizer = new BrokerBasedLog<>(workerConfig,
+ workerConfig.getConfigStoreTopic(),
+ ConnectUtil.createGroupName(configManagePrefix, workerConfig.getWorkerId()),
new ConfigChangeCallback(),
new JsonConverter(),
new ConnectKeyValueConverter());
this.connectorKeyValueStore = new FileBaseKeyValueStore<>(
- FilePathConfigUtil.getConnectorConfigPath(connectConfig.getStorePathRootDir()),
+ FilePathConfigUtil.getConnectorConfigPath(workerConfig.getStorePathRootDir()),
new JsonConverter(),
new JsonConverter(ConnectKeyValue.class));
this.taskKeyValueStore = new FileBaseKeyValueStore<>(
- FilePathConfigUtil.getTaskConfigPath(connectConfig.getStorePathRootDir()),
+ FilePathConfigUtil.getTaskConfigPath(workerConfig.getStorePathRootDir()),
new JsonConverter(),
new ListConverter(ConnectKeyValue.class));
this.plugin = plugin;
- this.prepare(connectConfig);
+ this.prepare(workerConfig);
+ this.workerConfig = workerConfig;
}
@@ -114,7 +121,7 @@ public class ConfigManagementServiceImpl extends AbstractConfigManagementService
*
* @param connectConfig
*/
- private void prepare(ConnectConfig connectConfig) {
+ private void prepare(WorkerConfig connectConfig) {
String configStoreTopic = connectConfig.getConfigStoreTopic();
if (!ConnectUtil.isTopicExist(connectConfig, configStoreTopic)) {
log.info("try to create config store topic: {}!", configStoreTopic);
@@ -147,25 +154,32 @@ public class ConfigManagementServiceImpl extends AbstractConfigManagementService
@Override
public String putConnectorConfig(String connectorName, ConnectKeyValue configs) {
- ConnectKeyValue exist = connectorKeyValueStore.get(connectorName);
- // update version
- if (null != exist) {
- Long updateTimestamp = exist.getLong(RuntimeConfigDefine.UPDATE_TIMESTAMP);
+ // check request config
+ for (String requireConfig : ConnectorConfig.REQUEST_CONFIG) {
+ if (!configs.containsKey(requireConfig)) {
+ throw new ConnectException("Request config key: " + requireConfig);
+ }
+ }
+ // update timestamp
+ ConnectKeyValue oldConfig = connectorKeyValueStore.get(connectorName);
+ if (null != oldConfig) {
+ Long updateTimestamp = oldConfig.getLong(ConnectorConfig.UPDATE_TIMESTAMP);
if (null != updateTimestamp) {
- configs.put(RuntimeConfigDefine.UPDATE_TIMESTAMP, updateTimestamp);
+ configs.put(ConnectorConfig.UPDATE_TIMESTAMP, updateTimestamp);
}
}
-
- if (configs.equals(exist)) {
+ if (configs.equals(oldConfig)) {
throw new ConnectException("Connector with same config already exist.");
}
-
- for (String requireConfig : RuntimeConfigDefine.REQUEST_CONFIG) {
- if (!configs.containsKey(requireConfig)) {
- throw new ConnectException("Request config key: " + requireConfig);
- }
+ // validate config
+ Connector connector = plugin.newConnector(configs.getString(CONNECTOR_CLASS));
+ if (connector instanceof SourceConnector) {
+ new SourceConnectorConfig(configs).validate();
+ } else if (connector instanceof SinkConnector) {
+ new SinkConnectorConfig(configs).validate();
}
- configs.put(RuntimeConfigDefine.UPDATE_TIMESTAMP, System.currentTimeMillis());
+ connector.validate(configs);
+ configs.put(ConnectorConfig.UPDATE_TIMESTAMP, System.currentTimeMillis());
configs.setTargetState(TargetState.STARTED);
dataSynchronizer.send(CONNECTOR_KEY(connectorName), configs);
return connectorName;
@@ -231,7 +245,7 @@ public class ConfigManagementServiceImpl extends AbstractConfigManagementService
newConfig.setProperties(new HashMap<>(config.getProperties()));
newConfig.setTargetState(config.getTargetState());
// update version by timestamp
- newConfig.put(RuntimeConfigDefine.UPDATE_TIMESTAMP, System.currentTimeMillis());
+ newConfig.put(ConnectorConfig.UPDATE_TIMESTAMP, System.currentTimeMillis());
return newConfig;
}
@@ -290,7 +304,7 @@ public class ConfigManagementServiceImpl extends AbstractConfigManagementService
});
taskKeyValueStore.getKVMap().forEach((connectName, taskConfigs) -> {
taskConfigs.forEach(taskConfig -> {
- ConnectorTaskId taskId = new ConnectorTaskId(connectName, taskConfig.getInt(RuntimeConfigDefine.TASK_ID));
+ ConnectorTaskId taskId = new ConnectorTaskId(connectName, taskConfig.getInt(ConnectorConfig.TASK_ID));
dataSynchronizer.send(TASK_KEY(taskId), taskConfig);
});
});
@@ -339,8 +353,8 @@ public class ConfigManagementServiceImpl extends AbstractConfigManagementService
ConnectKeyValue oldConfig = connectorKeyValueStore.get(connectorName);
// config update
if (!value.equals(oldConfig)) {
- Long oldUpdateTime = oldConfig.getLong(RuntimeConfigDefine.UPDATE_TIMESTAMP);
- Long newUpdateTime = value.getLong(RuntimeConfigDefine.UPDATE_TIMESTAMP);
+ Long oldUpdateTime = oldConfig.getLong(ConnectorConfig.UPDATE_TIMESTAMP);
+ Long newUpdateTime = value.getLong(ConnectorConfig.UPDATE_TIMESTAMP);
if (newUpdateTime > oldUpdateTime) {
// remove
connectorKeyValueStore.remove(connectorName);
@@ -389,8 +403,8 @@ public class ConfigManagementServiceImpl extends AbstractConfigManagementService
ConnectKeyValue oldConfig = connectorKeyValueStore.get(connectorName);
// config update
if (!value.equals(oldConfig)) {
- Long oldUpdateTime = oldConfig.getLong(RuntimeConfigDefine.UPDATE_TIMESTAMP);
- Long newUpdateTime = value.getLong(RuntimeConfigDefine.UPDATE_TIMESTAMP);
+ Long oldUpdateTime = oldConfig.getLong(ConnectorConfig.UPDATE_TIMESTAMP);
+ Long newUpdateTime = value.getLong(ConnectorConfig.UPDATE_TIMESTAMP);
if (newUpdateTime > oldUpdateTime) {
// remove
oldConfig.setTargetState(value.getTargetState());
@@ -425,13 +439,13 @@ public class ConfigManagementServiceImpl extends AbstractConfigManagementService
private boolean mergeConnectConfig(String connectName, ConnectKeyValue connectKeyValue) {
if (!connectorKeyValueStore.containsKey(connectName)) {
connectorKeyValueStore.put(connectName, connectKeyValue);
- recomputeTaskConfigs(connectName, loadConnector(connectKeyValue), connectKeyValue.getLong(RuntimeConfigDefine.UPDATE_TIMESTAMP), connectKeyValue);
+ recomputeTaskConfigs(connectName, loadConnector(connectKeyValue), connectKeyValue.getLong(ConnectorConfig.UPDATE_TIMESTAMP), connectKeyValue);
}
ConnectKeyValue oldConfig = connectorKeyValueStore.get(connectName);
// config update
if (!connectKeyValue.equals(oldConfig)) {
- Long oldUpdateTime = oldConfig.getLong(RuntimeConfigDefine.UPDATE_TIMESTAMP);
- Long newUpdateTime = connectKeyValue.getLong(RuntimeConfigDefine.UPDATE_TIMESTAMP);
+ Long oldUpdateTime = oldConfig.getLong(ConnectorConfig.UPDATE_TIMESTAMP);
+ Long newUpdateTime = connectKeyValue.getLong(ConnectorConfig.UPDATE_TIMESTAMP);
if (newUpdateTime > oldUpdateTime) {
connectorKeyValueStore.put(connectName, connectKeyValue);
recomputeTaskConfigs(connectName, loadConnector(connectKeyValue), newUpdateTime, connectKeyValue);
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/DefaultConnectorContext.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/DefaultConnectorContext.java
index d6254cd..98a63a7 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/DefaultConnectorContext.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/DefaultConnectorContext.java
@@ -24,7 +24,7 @@ import java.util.Set;
import org.apache.rocketmq.connect.runtime.controller.AbstractConnectController;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.WorkerConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,7 +54,7 @@ public class DefaultConnectorContext implements ConnectorContext {
Connector connector = currentConnector.getConnector();
ConnectKeyValue connectKeyValue = controller.getConfigManagementService().getConnectorConfigs().get(connectorName);
long currentTimeMillis = System.currentTimeMillis();
- connectKeyValue.put(RuntimeConfigDefine.UPDATE_TIMESTAMP, currentTimeMillis);
+ connectKeyValue.put(ConnectorConfig.UPDATE_TIMESTAMP, currentTimeMillis);
controller.getConfigManagementService().recomputeTaskConfigs(connectorName, connector, currentTimeMillis, connectKeyValue);
log.info("Connector {} recompute taskConfigs success.", connectorName);
} else {
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/PositionManagementService.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/PositionManagementService.java
index 3d1a4b0..9a78474 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/PositionManagementService.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/PositionManagementService.java
@@ -21,7 +21,7 @@ package org.apache.rocketmq.connect.runtime.service;
import io.openmessaging.connector.api.data.RecordOffset;
import java.util.List;
import java.util.Map;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.store.ExtendRecordPartition;
/**
@@ -44,7 +44,7 @@ public interface PositionManagementService {
*
* @param config can be DistributedConfig or StandaloneConfig
*/
- default void configure(ConnectConfig config) {
+ default void configure(WorkerConfig config) {
}
@@ -93,7 +93,7 @@ public interface PositionManagementService {
*/
void registerListener(PositionUpdateListener listener);
- void initialize(ConnectConfig connectConfig);
+ void initialize(WorkerConfig connectConfig);
interface PositionUpdateListener {
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/PositionManagementServiceImpl.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/PositionManagementServiceImpl.java
index c372d68..e2fc963 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/PositionManagementServiceImpl.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/PositionManagementServiceImpl.java
@@ -28,7 +28,7 @@ import java.util.stream.Collectors;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.converter.JsonConverter;
import org.apache.rocketmq.connect.runtime.converter.RecordOffsetConverter;
import org.apache.rocketmq.connect.runtime.converter.RecordPartitionConverter;
@@ -77,7 +77,7 @@ public class PositionManagementServiceImpl implements PositionManagementService
*
* @param connectConfig
*/
- private void prepare(ConnectConfig connectConfig) {
+ private void prepare(WorkerConfig connectConfig) {
String positionStoreTopic = connectConfig.getPositionStoreTopic();
if (!ConnectUtil.isTopicExist(connectConfig, positionStoreTopic)) {
log.info("try to create position store topic: {}!", positionStoreTopic);
@@ -163,7 +163,7 @@ public class PositionManagementServiceImpl implements PositionManagementService
this.positionUpdateListener.add(listener);
}
- @Override public void initialize(ConnectConfig connectConfig) {
+ @Override public void initialize(WorkerConfig connectConfig) {
this.positionStore = new FileBaseKeyValueStore<>(FilePathConfigUtil.getPositionPath(connectConfig.getStorePathRootDir()),
new RecordPartitionConverter(),
new RecordOffsetConverter());
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/RebalanceImpl.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/RebalanceImpl.java
index ff49726..4fc9915 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/RebalanceImpl.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/RebalanceImpl.java
@@ -25,7 +25,7 @@ import org.apache.rocketmq.connect.runtime.controller.AbstractConnectController;
import org.apache.rocketmq.connect.runtime.common.ConnAndTaskConfigs;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.Worker;
import org.apache.rocketmq.connect.runtime.service.memory.MemoryClusterManagementServiceImpl;
import org.apache.rocketmq.connect.runtime.service.strategy.AllocateConnAndTaskStrategy;
@@ -77,7 +77,7 @@ public class RebalanceImpl {
public void checkClusterStoreTopic() {
if (!clusterManagementService.hasClusterStoreTopic()) {
- ConnectConfig connectConfig = this.connectController.getConnectConfig();
+ WorkerConfig connectConfig = this.connectController.getConnectConfig();
String clusterStoreTopic = this.connectController.getConnectConfig().getClusterStoreTopic();
log.info("cluster store topic not exist, try to create it!");
TopicConfig topicConfig = new TopicConfig(clusterStoreTopic, 1, 1, 6);
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/StateManagementService.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/StateManagementService.java
index 39c2508..7d5d645 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/StateManagementService.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/StateManagementService.java
@@ -16,7 +16,7 @@
*/
package org.apache.rocketmq.connect.runtime.service;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.status.ConnectorStatus;
import org.apache.rocketmq.connect.runtime.connectorwrapper.status.TaskStatus;
import org.apache.rocketmq.connect.runtime.utils.ConnectorTaskId;
@@ -31,7 +31,7 @@ public interface StateManagementService {
* initialize cb config
* @param connectConfig
*/
- void initialize(ConnectConfig connectConfig);
+ void initialize(WorkerConfig connectConfig);
/**
* Start dependent services (if needed)
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/StateManagementServiceImpl.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/StateManagementServiceImpl.java
index a934678..9c20f7e 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/StateManagementServiceImpl.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/StateManagementServiceImpl.java
@@ -20,7 +20,7 @@ import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.connect.runtime.common.ConnAndTaskStatus;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.status.AbstractStatus;
import org.apache.rocketmq.connect.runtime.connectorwrapper.status.ConnectorStatus;
import org.apache.rocketmq.connect.runtime.connectorwrapper.status.TaskStatus;
@@ -79,7 +79,7 @@ public class StateManagementServiceImpl implements StateManagementService {
*
* @param connectConfig
*/
- private void prepare(ConnectConfig connectConfig) {
+ private void prepare(WorkerConfig connectConfig) {
String connectStatusTopic = connectConfig.getConnectStatusTopic();
if (!ConnectUtil.isTopicExist(connectConfig, connectStatusTopic)) {
log.info("try to create status topic: {}!", connectStatusTopic);
@@ -94,7 +94,7 @@ public class StateManagementServiceImpl implements StateManagementService {
* @param config
*/
@Override
- public void initialize(ConnectConfig config) {
+ public void initialize(WorkerConfig config) {
this.dataSynchronizer = new BrokerBasedLog(config,
config.getConnectStatusTopic(),
ConnectUtil.createGroupName(statusManagePrefix, config.getWorkerId()),
@@ -270,7 +270,11 @@ public class StateManagementServiceImpl implements StateManagementService {
*/
@Override
public TaskStatus get(ConnectorTaskId id) {
- return connAndTaskStatus.getTasks().get(id.connector(), id.task()).get();
+ ConnAndTaskStatus.CacheEntry<TaskStatus> cacheEntry = connAndTaskStatus.getTasks().get(id.connector(), id.task());
+ if (cacheEntry == null) {
+ return null;
+ }
+ return cacheEntry.get();
}
/**
@@ -281,7 +285,11 @@ public class StateManagementServiceImpl implements StateManagementService {
*/
@Override
public ConnectorStatus get(String connector) {
- return connAndTaskStatus.getConnectors().get(connector).get();
+ ConnAndTaskStatus.CacheEntry<ConnectorStatus> cacheEntry = connAndTaskStatus.getConnectors().get(connector);
+ if (cacheEntry == null) {
+ return null;
+ }
+ return cacheEntry.get();
}
/**
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/FilePositionManagementServiceImpl.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/FilePositionManagementServiceImpl.java
index 056773b..024728c 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/FilePositionManagementServiceImpl.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/FilePositionManagementServiceImpl.java
@@ -20,7 +20,7 @@ import io.openmessaging.connector.api.data.RecordOffset;
import io.openmessaging.connector.api.errors.ConnectException;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.converter.RecordOffsetConverter;
import org.apache.rocketmq.connect.runtime.converter.RecordPartitionConverter;
import org.apache.rocketmq.connect.runtime.service.PositionManagementService;
@@ -61,7 +61,7 @@ public class FilePositionManagementServiceImpl implements PositionManagementServ
}
- @Override public void initialize(ConnectConfig connectConfig) {
+ @Override public void initialize(WorkerConfig connectConfig) {
this.positionStore = new FileBaseKeyValueStore<>(FilePathConfigUtil.getPositionPath(connectConfig.getStorePathRootDir()),
new RecordPartitionConverter(),
new RecordOffsetConverter());
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/MemoryClusterManagementServiceImpl.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/MemoryClusterManagementServiceImpl.java
index 8add05b..a9f24f8 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/MemoryClusterManagementServiceImpl.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/MemoryClusterManagementServiceImpl.java
@@ -17,7 +17,7 @@
package org.apache.rocketmq.connect.runtime.service.memory;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.controller.standalone.StandaloneConfig;
import org.apache.rocketmq.connect.runtime.service.ClusterManagementService;
@@ -36,7 +36,7 @@ public class MemoryClusterManagementServiceImpl implements ClusterManagementServ
}
- @Override public void initialize(ConnectConfig connectConfig) {
+ @Override public void initialize(WorkerConfig connectConfig) {
this.configure(connectConfig);
}
@@ -46,7 +46,7 @@ public class MemoryClusterManagementServiceImpl implements ClusterManagementServ
* @param config can be DistributedConfig or StandaloneConfig
*/
@Override
- public void configure(ConnectConfig config) {
+ public void configure(WorkerConfig config) {
this.config = (StandaloneConfig) config;
}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/MemoryConfigManagementServiceImpl.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/MemoryConfigManagementServiceImpl.java
index 515c714..a159190 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/MemoryConfigManagementServiceImpl.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/MemoryConfigManagementServiceImpl.java
@@ -18,11 +18,15 @@ package org.apache.rocketmq.connect.runtime.service.memory;
import io.openmessaging.connector.api.component.connector.Connector;
+import io.openmessaging.connector.api.component.task.sink.SinkConnector;
+import io.openmessaging.connector.api.component.task.source.SourceConnector;
import io.openmessaging.connector.api.errors.ConnectException;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.SinkConnectorConfig;
+import org.apache.rocketmq.connect.runtime.config.SourceConnectorConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.TargetState;
import org.apache.rocketmq.connect.runtime.controller.isolation.Plugin;
import org.apache.rocketmq.connect.runtime.service.AbstractConfigManagementService;
@@ -49,10 +53,11 @@ public class MemoryConfigManagementServiceImpl extends AbstractConfigManagementS
}
@Override
- public void initialize(ConnectConfig connectConfig, Plugin plugin) {
+ public void initialize(WorkerConfig workerConfig, Plugin plugin) {
this.connectorKeyValueStore = new MemoryBasedKeyValueStore<>();
this.taskKeyValueStore = new MemoryBasedKeyValueStore<>();
this.plugin = plugin;
+ this.workerConfig = workerConfig;
}
@Override
@@ -80,26 +85,35 @@ public class MemoryConfigManagementServiceImpl extends AbstractConfigManagementS
@Override
public String putConnectorConfig(String connectorName, ConnectKeyValue configs) {
- ConnectKeyValue exist = connectorKeyValueStore.get(connectorName);
- if (null != exist) {
- Long updateTimestamp = exist.getLong(RuntimeConfigDefine.UPDATE_TIMESTAMP);
- if (null != updateTimestamp) {
- configs.put(RuntimeConfigDefine.UPDATE_TIMESTAMP, updateTimestamp);
- }
- }
- if (configs.equals(exist)) {
- throw new ConnectException("Connector with same config already exist.");
- }
- Long currentTimestamp = System.currentTimeMillis();
- configs.put(RuntimeConfigDefine.UPDATE_TIMESTAMP, currentTimestamp);
- for (String requireConfig : RuntimeConfigDefine.REQUEST_CONFIG) {
+ // check request config
+ for (String requireConfig : ConnectorConfig.REQUEST_CONFIG) {
if (!configs.containsKey(requireConfig)) {
throw new ConnectException("Request config key: " + requireConfig);
}
}
+ // current timestamp
+ Long currentTimestamp = System.currentTimeMillis();
+ ConnectKeyValue oldConfig = connectorKeyValueStore.get(connectorName);
+ if (null != oldConfig) {
+ Long updateTimestamp = oldConfig.getLong(ConnectorConfig.UPDATE_TIMESTAMP);
+ if (null != updateTimestamp) {
+ configs.put(ConnectorConfig.UPDATE_TIMESTAMP, currentTimestamp);
+ }
+ }
+ if (configs.equals(oldConfig)) {
+ throw new ConnectException("Connector with same config already exist.");
+ }
+ // validate config
Connector connector = super.loadConnector(configs);
+ if (connector instanceof SourceConnector) {
+ new SourceConnectorConfig(configs).validate();
+ } else if (connector instanceof SinkConnector) {
+ new SinkConnectorConfig(configs).validate();
+ }
configs.setTargetState(TargetState.STARTED);
+ // update cache
+
connectorKeyValueStore.put(connectorName, configs);
recomputeTaskConfigs(connectorName, connector, currentTimestamp, configs);
return connectorName;
@@ -129,7 +143,7 @@ public class MemoryConfigManagementServiceImpl extends AbstractConfigManagementS
throw new ConnectException("Connector [" + connectorName + "] does not exist");
}
ConnectKeyValue config = connectorKeyValueStore.get(connectorName);
- config.put(RuntimeConfigDefine.UPDATE_TIMESTAMP, System.currentTimeMillis());
+ config.put(ConnectorConfig.UPDATE_TIMESTAMP, System.currentTimeMillis());
config.setTargetState(TargetState.PAUSED);
connectorKeyValueStore.put(connectorName, config);
triggerListener();
@@ -146,12 +160,15 @@ public class MemoryConfigManagementServiceImpl extends AbstractConfigManagementS
throw new ConnectException("Connector [" + connectorName + "] does not exist");
}
ConnectKeyValue config = connectorKeyValueStore.get(connectorName);
- config.put(RuntimeConfigDefine.UPDATE_TIMESTAMP, System.currentTimeMillis());
+ config.put(ConnectorConfig.UPDATE_TIMESTAMP, System.currentTimeMillis());
config.setTargetState(TargetState.STARTED);
+
connectorKeyValueStore.put(connectorName, config);
triggerListener();
}
+
+
@Override
public Map<String, List<ConnectKeyValue>> getTaskConfigs() {
return taskKeyValueStore.getKVMap();
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/MemoryStateManagementServiceImpl.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/MemoryStateManagementServiceImpl.java
index 93efbcb..254096d 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/MemoryStateManagementServiceImpl.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/MemoryStateManagementServiceImpl.java
@@ -17,7 +17,7 @@
package org.apache.rocketmq.connect.runtime.service.memory;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.status.ConnectorStatus;
import org.apache.rocketmq.connect.runtime.connectorwrapper.status.TaskStatus;
import org.apache.rocketmq.connect.runtime.service.StagingMode;
@@ -49,7 +49,7 @@ public class MemoryStateManagementServiceImpl implements StateManagementService
* @param config
*/
@Override
- public void initialize(ConnectConfig config) {
+ public void initialize(WorkerConfig config) {
this.tasks = new Table<>();
this.connectors = new ConcurrentHashMap<>();
}
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/strategy/AllocateConnAndTaskStrategyByConsistentHash.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/strategy/AllocateConnAndTaskStrategyByConsistentHash.java
index 956d048..bb0398e 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/strategy/AllocateConnAndTaskStrategyByConsistentHash.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/strategy/AllocateConnAndTaskStrategyByConsistentHash.java
@@ -28,7 +28,7 @@ import org.apache.rocketmq.common.consistenthash.Node;
import org.apache.rocketmq.connect.runtime.common.ConnAndTaskConfigs;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,8 +39,8 @@ public class AllocateConnAndTaskStrategyByConsistentHash implements AllocateConn
private HashFunction hashFunc;
public AllocateConnAndTaskStrategyByConsistentHash() {
- virtualNodes = Integer.parseInt(System.getProperty(RuntimeConfigDefine.VIRTUAL_NODE, "0"));
- String funcName = System.getProperty(RuntimeConfigDefine.HASH_FUNC);
+ virtualNodes = Integer.parseInt(System.getProperty(ConnectorConfig.VIRTUAL_NODE, "0"));
+ String funcName = System.getProperty(ConnectorConfig.HASH_FUNC);
if (StringUtils.isNoneEmpty(funcName)) {
try {
hashFunc = (HashFunction) Class.forName(funcName).newInstance();
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/stats/ConnectStatsManager.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/stats/ConnectStatsManager.java
index c02b0d8..4e555c5 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/stats/ConnectStatsManager.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/stats/ConnectStatsManager.java
@@ -25,7 +25,7 @@ import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.stats.StatsItem;
import org.apache.rocketmq.common.stats.StatsItemSet;
import org.apache.rocketmq.common.utils.ThreadUtils;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
@@ -89,9 +89,9 @@ public class ConnectStatsManager {
private final HashMap<String, StatsItemSet> statsTable = new HashMap<String, StatsItemSet>();
private final String worker;
- private ConnectConfig connectConfig;
+ private WorkerConfig connectConfig;
- public ConnectStatsManager(ConnectConfig connectConfig) {
+ public ConnectStatsManager(WorkerConfig connectConfig) {
this.connectConfig = connectConfig;
this.worker = connectConfig.getWorkerId();
init();
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/utils/ConnectUtil.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/utils/ConnectUtil.java
index f0684d5..42100ba 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/utils/ConnectUtil.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/utils/ConnectUtil.java
@@ -38,8 +38,9 @@ import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.SinkConnectorConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.apache.rocketmq.connect.runtime.service.strategy.AllocateConnAndTaskStrategy;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingUtil;
@@ -94,7 +95,7 @@ public class ConnectUtil {
return new StringBuffer(prefix).append("-").append(UUID.randomUUID().toString()).toString();
}
- public static AllocateConnAndTaskStrategy initAllocateConnAndTaskStrategy(ConnectConfig connectConfig) {
+ public static AllocateConnAndTaskStrategy initAllocateConnAndTaskStrategy(WorkerConfig connectConfig) {
try {
return (AllocateConnAndTaskStrategy) Thread.currentThread().getContextClassLoader().loadClass(connectConfig.getAllocTaskStrategy()).newInstance();
} catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) {
@@ -102,7 +103,7 @@ public class ConnectUtil {
}
}
- public static DefaultMQProducer initDefaultMQProducer(ConnectConfig connectConfig) {
+ public static DefaultMQProducer initDefaultMQProducer(WorkerConfig connectConfig) {
RPCHook rpcHook = null;
if (connectConfig.getAclEnable()) {
rpcHook = new AclClientRPCHook(new SessionCredentials(connectConfig.getAccessKey(), connectConfig.getSecretKey()));
@@ -112,12 +113,12 @@ public class ConnectUtil {
producer.setInstanceName(createUniqInstance(connectConfig.getNamesrvAddr()));
producer.setProducerGroup(connectConfig.getRmqProducerGroup());
producer.setSendMsgTimeout(connectConfig.getOperationTimeout());
- producer.setMaxMessageSize(RuntimeConfigDefine.MAX_MESSAGE_SIZE);
+ producer.setMaxMessageSize(ConnectorConfig.MAX_MESSAGE_SIZE);
producer.setLanguage(LanguageCode.JAVA);
return producer;
}
- public static DefaultMQPullConsumer initDefaultMQPullConsumer(ConnectConfig connectConfig) {
+ public static DefaultMQPullConsumer initDefaultMQPullConsumer(WorkerConfig connectConfig) {
RPCHook rpcHook = null;
if (connectConfig.getAclEnable()) {
rpcHook = new AclClientRPCHook(new SessionCredentials(connectConfig.getAccessKey(), connectConfig.getSecretKey()));
@@ -128,12 +129,12 @@ public class ConnectUtil {
consumer.setConsumerGroup(connectConfig.getRmqConsumerGroup());
consumer.setMaxReconsumeTimes(connectConfig.getRmqMaxRedeliveryTimes());
consumer.setBrokerSuspendMaxTimeMillis(connectConfig.getBrokerSuspendMaxTimeMillis());
- consumer.setConsumerPullTimeoutMillis((long) connectConfig.getRmqMessageConsumeTimeout());
+ consumer.setConsumerPullTimeoutMillis(connectConfig.getRmqMessageConsumeTimeout());
consumer.setLanguage(LanguageCode.JAVA);
return consumer;
}
- public static DefaultMQPushConsumer initDefaultMQPushConsumer(ConnectConfig connectConfig) {
+ public static DefaultMQPushConsumer initDefaultMQPushConsumer(WorkerConfig connectConfig) {
RPCHook rpcHook = null;
if (connectConfig.getAclEnable()) {
rpcHook = new AclClientRPCHook(new SessionCredentials(connectConfig.getAccessKey(), connectConfig.getSecretKey()));
@@ -143,14 +144,14 @@ public class ConnectUtil {
consumer.setInstanceName(createUniqInstance(connectConfig.getNamesrvAddr()));
consumer.setConsumerGroup(createGroupName(connectConfig.getRmqConsumerGroup()));
consumer.setMaxReconsumeTimes(connectConfig.getRmqMaxRedeliveryTimes());
- consumer.setConsumeTimeout((long) connectConfig.getRmqMessageConsumeTimeout());
+ consumer.setConsumeTimeout(connectConfig.getRmqMessageConsumeTimeout());
consumer.setConsumeThreadMin(connectConfig.getRmqMinConsumeThreadNums());
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
consumer.setLanguage(LanguageCode.JAVA);
return consumer;
}
- public static DefaultMQAdminExt startMQAdminTool(ConnectConfig connectConfig) throws MQClientException {
+ public static DefaultMQAdminExt startMQAdminTool(WorkerConfig connectConfig) throws MQClientException {
RPCHook rpcHook = null;
if (connectConfig.getAclEnable()) {
rpcHook = new AclClientRPCHook(new SessionCredentials(connectConfig.getAccessKey(), connectConfig.getSecretKey()));
@@ -163,7 +164,7 @@ public class ConnectUtil {
return defaultMQAdminExt;
}
- public static void createTopic(ConnectConfig connectConfig, TopicConfig topicConfig) {
+ public static void createTopic(WorkerConfig connectConfig, TopicConfig topicConfig) {
DefaultMQAdminExt defaultMQAdminExt = null;
try {
defaultMQAdminExt = startMQAdminTool(connectConfig);
@@ -185,7 +186,7 @@ public class ConnectUtil {
}
}
- public static boolean isTopicExist(ConnectConfig connectConfig, String topic) {
+ public static boolean isTopicExist(WorkerConfig connectConfig, String topic) {
DefaultMQAdminExt defaultMQAdminExt = null;
boolean foundTopicRouteInfo = false;
try {
@@ -206,7 +207,7 @@ public class ConnectUtil {
return foundTopicRouteInfo;
}
- public static Set<String> fetchAllConsumerGroupList(ConnectConfig connectConfig) {
+ public static Set<String> fetchAllConsumerGroupList(WorkerConfig connectConfig) {
Set<String> consumerGroupSet = Sets.newHashSet();
DefaultMQAdminExt defaultMQAdminExt = null;
try {
@@ -226,7 +227,7 @@ public class ConnectUtil {
return consumerGroupSet;
}
- public static String createSubGroup(ConnectConfig connectConfig, String subGroup) {
+ public static String createSubGroup(WorkerConfig connectConfig, String subGroup) {
DefaultMQAdminExt defaultMQAdminExt = null;
try {
defaultMQAdminExt = startMQAdminTool(connectConfig);
@@ -311,9 +312,9 @@ public class ConnectUtil {
* @return
* @throws MQClientException
*/
- public static DefaultLitePullConsumer initDefaultLitePullConsumer(ConnectConfig connectConfig, ConnectorTaskId id, ConnectKeyValue keyValue, boolean autoCommit) throws MQClientException {
+ public static DefaultLitePullConsumer initDefaultLitePullConsumer(WorkerConfig connectConfig, ConnectorTaskId id, ConnectKeyValue keyValue, boolean autoCommit) throws MQClientException {
DefaultLitePullConsumer consumer = null;
- String groupId = keyValue.getString(RuntimeConfigDefine.TASK_GROUP_ID);
+ String groupId = keyValue.getString(SinkConnectorConfig.TASK_GROUP_ID);
if (StringUtils.isBlank(groupId)) {
groupId = SYS_TASK_CG_PREFIX + id.connector();
}
@@ -336,7 +337,7 @@ public class ConnectUtil {
return new AclClientRPCHook(new SessionCredentials(accessKey, secretKey));
}
- public static DefaultMQPullConsumer initDefaultMQPullConsumer(ConnectConfig connectConfig, ConnectorTaskId id, ConnectKeyValue keyValue) {
+ public static DefaultMQPullConsumer initDefaultMQPullConsumer(WorkerConfig connectConfig, ConnectorTaskId id, ConnectKeyValue keyValue) {
RPCHook rpcHook = null;
if (connectConfig.getAclEnable()) {
rpcHook = new AclClientRPCHook(new SessionCredentials(connectConfig.getAccessKey(), connectConfig.getSecretKey()));
@@ -355,7 +356,7 @@ public class ConnectUtil {
return consumer;
}
- public static DefaultMQPullConsumer initDefaultMQPullConsumer(ConnectConfig connectConfig, String connectorName, ConnectKeyValue keyValue) {
+ public static DefaultMQPullConsumer initDefaultMQPullConsumer(WorkerConfig connectConfig, String connectorName, ConnectKeyValue keyValue) {
RPCHook rpcHook = null;
if (connectConfig.getAclEnable()) {
rpcHook = new AclClientRPCHook(new SessionCredentials(connectConfig.getAccessKey(), connectConfig.getSecretKey()));
diff --git a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/utils/datasync/BrokerBasedLog.java b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/utils/datasync/BrokerBasedLog.java
index bc5ed0c..fee1459 100644
--- a/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/utils/datasync/BrokerBasedLog.java
+++ b/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/utils/datasync/BrokerBasedLog.java
@@ -33,13 +33,13 @@ import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.connect.runtime.common.LoggerName;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.utils.Callback;
import org.apache.rocketmq.connect.runtime.utils.ConnectUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine.MAX_MESSAGE_SIZE;
+import static org.apache.rocketmq.connect.runtime.config.ConnectorConfig.MAX_MESSAGE_SIZE;
/**
* A Broker base data synchronizer, synchronize data between workers.
@@ -81,12 +81,12 @@ public class BrokerBasedLog<K, V> implements DataSynchronizer<K, V> {
*/
private Converter valueConverter;
- public BrokerBasedLog(ConnectConfig connectConfig,
- String topicName,
- String workId,
- DataSynchronizerCallback<K, V> dataSynchronizerCallback,
- Converter keyConverter,
- Converter valueConverter) {
+ public BrokerBasedLog(WorkerConfig connectConfig,
+ String topicName,
+ String workId,
+ DataSynchronizerCallback<K, V> dataSynchronizerCallback,
+ Converter keyConverter,
+ Converter valueConverter) {
this.topicName = topicName;
this.dataSynchronizerCallback = dataSynchronizerCallback;
@@ -104,7 +104,7 @@ public class BrokerBasedLog<K, V> implements DataSynchronizer<K, V> {
*
* @param connectConfig
*/
- private void prepare(ConnectConfig connectConfig) {
+ private void prepare(WorkerConfig connectConfig) {
if (connectConfig.isAutoCreateGroupEnable()) {
ConnectUtil.createSubGroup(connectConfig, consumer.getConsumerGroup());
}
diff --git a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/config/ConnectConfigTest.java b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/config/ConnectConfigTest.java
index 8486179..6da21bd 100644
--- a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/config/ConnectConfigTest.java
+++ b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/config/ConnectConfigTest.java
@@ -25,7 +25,7 @@ public class ConnectConfigTest {
@Test
public void testConnectConfigAttribute() {
- ConnectConfig connectConfig = new ConnectConfig();
+ WorkerConfig connectConfig = new WorkerConfig();
connectConfig.setHttpPort(8081);
assertThat(connectConfig.getHttpPort()).isEqualTo(8081);
@@ -104,17 +104,14 @@ public class ConnectConfigTest {
connectConfig.setAutoCreateGroupEnable(true);
assertThat(true == connectConfig.isAutoCreateGroupEnable());
- connectConfig.setClusterName("default_cluster");
- assertThat("default_cluster".equals(connectConfig.getClusterName()));
-
connectConfig.setAdminExtGroup("admin_group");
assertThat("admin_group".equals(connectConfig.getAdminExtGroup()));
connectConfig.setConnectHome("/usr/connect");
assertThat("/usr/connect".equals(connectConfig.getConnectHome()));
- connectConfig.setOffsetCommitIntervalMs(100);
- assertThat(100 == connectConfig.getOffsetCommitIntervalMs());
+ connectConfig.setOffsetCommitIntervalMsConfig(100);
+ assertThat(100 == connectConfig.getOffsetCommitIntervalMsConfig());
connectConfig.setMaxStartTimeoutMills(100);
assertThat(100 == connectConfig.getMaxStartTimeoutMills());
diff --git a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/config/SinkConnectorConfigTest.java b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/config/SinkConnectorConfigTest.java
index 4517be9..1ed009d 100644
--- a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/config/SinkConnectorConfigTest.java
+++ b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/config/SinkConnectorConfigTest.java
@@ -29,12 +29,12 @@ public class SinkConnectorConfigTest {
@Test
public void parseTopicListTest() {
// connect-topicname is null
- final Set<String> result1 = SinkConnectorConfig.parseTopicList(connectKeyValue);
+ final Set<String> result1 = new SinkConnectorConfig(connectKeyValue).parseTopicList();
assert result1 == null;
// connect-topicname is not null
- connectKeyValue.put(RuntimeConfigDefine.CONNECT_TOPICNAME, "test");
- final Set<String> result2 = SinkConnectorConfig.parseTopicList(connectKeyValue);
+ connectKeyValue.put(SinkConnectorConfig.CONNECT_TOPICNAMES, "test");
+ final Set<String> result2 = new SinkConnectorConfig(connectKeyValue).parseTopicList();
assert !result2.isEmpty();
}
diff --git a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/SourceTaskOffsetCommitterTest.java b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/SourceTaskOffsetCommitterTest.java
index 805a1d0..76ea9e1 100644
--- a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/SourceTaskOffsetCommitterTest.java
+++ b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/SourceTaskOffsetCommitterTest.java
@@ -23,7 +23,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.utils.ConnectorTaskId;
import org.junit.After;
import org.junit.Before;
@@ -37,7 +37,7 @@ public class SourceTaskOffsetCommitterTest {
private SourceTaskOffsetCommitter sourceTaskOffsetCommitter;
- private ConnectConfig connectConfig = new ConnectConfig();
+ private WorkerConfig connectConfig = new WorkerConfig();
private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);
@@ -51,7 +51,7 @@ public class SourceTaskOffsetCommitterTest {
@Before
public void before() {
- connectConfig.setOffsetCommitIntervalMs(100);
+ connectConfig.setOffsetCommitIntervalMsConfig(100);
sourceTaskOffsetCommitter = new SourceTaskOffsetCommitter(connectConfig);
sourceTaskOffsetCommitter = new SourceTaskOffsetCommitter(connectConfig, scheduledExecutorService, committers);
ScheduledFuture<?> commitFuture = scheduledExecutorService.scheduleWithFixedDelay(() -> {
diff --git a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/TransformChainTest.java b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/TransformChainTest.java
index 8c6f2a2..f258a01 100644
--- a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/TransformChainTest.java
+++ b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/TransformChainTest.java
@@ -20,7 +20,7 @@ package org.apache.rocketmq.connect.runtime.connectorwrapper;
import io.openmessaging.KeyValue;
import io.openmessaging.connector.api.data.ConnectRecord;
import io.openmessaging.internal.DefaultKeyValue;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.apache.rocketmq.connect.runtime.controller.isolation.Plugin;
import org.junit.After;
import org.junit.Before;
@@ -44,7 +44,7 @@ public class TransformChainTest {
@Before
public void before() {
- keyValue.put(RuntimeConfigDefine.TRANSFORMS, "testTransform");
+ keyValue.put(ConnectorConfig.TRANSFORMS, "testTransform");
keyValue.put("transforms-testTransform-class", "org.apache.rocketmq.connect.runtime.connectorwrapper.TestTransform");
connectRecord.setExtensions(keyValue);
transformChain = new TransformChain(keyValue, plugin);
diff --git a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSinkTaskTest.java b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSinkTaskTest.java
index df4d51e..d50c2af 100644
--- a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSinkTaskTest.java
+++ b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSinkTaskTest.java
@@ -26,16 +26,20 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
-import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
+
+import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.SinkConnectorConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
+import org.apache.rocketmq.connect.runtime.connectorwrapper.status.WrapperStatusListener;
import org.apache.rocketmq.connect.runtime.connectorwrapper.testimpl.TestConverter;
import org.apache.rocketmq.connect.runtime.connectorwrapper.testimpl.TestSinkTask;
import org.apache.rocketmq.connect.runtime.controller.isolation.Plugin;
import org.apache.rocketmq.connect.runtime.errors.RetryWithToleranceOperator;
import org.apache.rocketmq.connect.runtime.errors.ToleranceType;
import org.apache.rocketmq.connect.runtime.errors.WorkerErrorRecordReporter;
+import org.apache.rocketmq.connect.runtime.service.StateManagementServiceImpl;
import org.apache.rocketmq.connect.runtime.stats.ConnectStatsManager;
import org.apache.rocketmq.connect.runtime.stats.ConnectStatsService;
import org.apache.rocketmq.connect.runtime.utils.ConnectorTaskId;
@@ -53,7 +57,7 @@ public class WorkerSinkTaskTest {
private WorkerSinkTask workerSinkTask;
- private ConnectConfig connectConfig = new ConnectConfig();
+ private WorkerConfig connectConfig = new WorkerConfig();
private ConnectorTaskId connectorTaskId = new ConnectorTaskId("testConnector", 1);
@@ -63,7 +67,7 @@ public class WorkerSinkTaskTest {
private RecordConverter recordConverter = new TestConverter();
- private DefaultMQPullConsumer defaultMQPullConsumer = new DefaultMQPullConsumer();
+ private DefaultLitePullConsumer defaultMQPullConsumer = new DefaultLitePullConsumer();
private AtomicReference<WorkerState> workerState = new AtomicReference<>(WorkerState.STARTED);
@@ -86,14 +90,14 @@ public class WorkerSinkTaskTest {
@Before
public void before() {
- connectKeyValue.put(RuntimeConfigDefine.CONNECT_TOPICNAME, "TEST_TOPIC");
- keyValue.put(RuntimeConfigDefine.TRANSFORMS, "testTransform");
+ connectKeyValue.put(SinkConnectorConfig.CONNECT_TOPICNAMES, "TEST_TOPIC");
+ keyValue.put(ConnectorConfig.TRANSFORMS, "testTransform");
keyValue.put("transforms-testTransform-class", "org.apache.rocketmq.connect.runtime.connectorwrapper.TestTransform");
transformChain = new TransformChain<>(keyValue, plugin);
workerErrorRecordReporter = new WorkerErrorRecordReporter(retryWithToleranceOperator, recordConverter);
workerSinkTask = new WorkerSinkTask(connectConfig, connectorTaskId, sinkTask, WorkerSinkTaskTest.class.getClassLoader(), connectKeyValue,
recordConverter, recordConverter, defaultMQPullConsumer, workerState, connectStatsManager, connectStatsService,
- transformChain, retryWithToleranceOperator, workerErrorRecordReporter);
+ transformChain, retryWithToleranceOperator, workerErrorRecordReporter, new WrapperStatusListener(new StateManagementServiceImpl(),"workId"));
}
@After
diff --git a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTaskTest.java b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTaskTest.java
index d24f754..63c2831 100644
--- a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTaskTest.java
+++ b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTaskTest.java
@@ -30,8 +30,9 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.SourceConnectorConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.testimpl.TestConverter;
import org.apache.rocketmq.connect.runtime.connectorwrapper.testimpl.TestPositionManageServiceImpl;
import org.apache.rocketmq.connect.runtime.connectorwrapper.testimpl.TestSourceTask;
@@ -56,7 +57,7 @@ public class WorkerSourceTaskTest {
private WorkerSourceTask workerSourceTask;
- private ConnectConfig connectConfig;
+ private WorkerConfig connectConfig;
private ConnectorTaskId connectorTaskId = new ConnectorTaskId("testConnector" ,1);
@@ -90,16 +91,16 @@ public class WorkerSourceTaskTest {
@Before
public void before() throws MQClientException, InterruptedException {
- connectConfig = new ConnectConfig();
+ connectConfig = new WorkerConfig();
connectConfig.setNamesrvAddr("127.0.0.1:9876");
connectStatsManager = new ConnectStatsManager(connectConfig);
- connectKeyValue.put(RuntimeConfigDefine.CONNECT_TOPICNAME, "TEST_TOPIC");
- keyValue.put(RuntimeConfigDefine.TRANSFORMS, "testTransform");
+ connectKeyValue.put(SourceConnectorConfig.CONNECT_TOPICNAME, "TEST_TOPIC");
+ keyValue.put(ConnectorConfig.TRANSFORMS, "testTransform");
keyValue.put("transforms-testTransform-class", "org.apache.rocketmq.connect.runtime.connectorwrapper.TestTransform");
transformChain = new TransformChain<>(keyValue, plugin);
workerSourceTask = new WorkerSourceTask(connectConfig, connectorTaskId, sourceTask, this.getClass().getClassLoader(),
connectKeyValue, positionManagementService, recordConverter, recordConverter, defaultMQProducer, workerState,
- connectStatsManager, connectStatsService, transformChain, retryWithToleranceOperator);
+ connectStatsManager, connectStatsService, transformChain, retryWithToleranceOperator,null);
NameServerMocker.startByDefaultConf(9876, 10911);
ServerResponseMocker.startServer(10911, "Hello World".getBytes(StandardCharsets.UTF_8));
}
diff --git a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerTest.java b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerTest.java
index 8573f13..61775c9 100644
--- a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerTest.java
+++ b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerTest.java
@@ -27,15 +27,14 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.connect.runtime.connectorwrapper.testimpl.TestSinkTask;
import org.apache.rocketmq.connect.runtime.controller.distributed.DistributedConnectController;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.testimpl.TestConnector;
import org.apache.rocketmq.connect.runtime.connectorwrapper.testimpl.TestConverter;
import org.apache.rocketmq.connect.runtime.connectorwrapper.testimpl.TestPositionManageServiceImpl;
@@ -78,7 +77,7 @@ public class WorkerTest {
@Mock
private DefaultMQProducer producer;
- private ConnectConfig connectConfig;
+ private WorkerConfig connectConfig;
private Worker worker;
@@ -110,7 +109,7 @@ public class WorkerTest {
when(plugin.currentThreadLoader()).thenReturn(this.getClass().getClassLoader());
when(plugin.newConnector(any())).thenReturn(new TestConnector());
when(delegatingClassLoader.pluginClassLoader(any())).thenReturn(pluginClassLoader);
- connectConfig = new ConnectConfig();
+ connectConfig = new WorkerConfig();
connectConfig.setHttpPort(8081);
connectConfig.setStorePathRootDir(System.getProperty("user.home") + File.separator + "testConnectorStore");
connectConfig.setNamesrvAddr("localhost:9876");
@@ -135,7 +134,7 @@ public class WorkerTest {
// create retry operator
RetryWithToleranceOperator retryWithToleranceOperator = ReporterManagerUtil.createRetryWithToleranceOperator(connectKeyValue);
retryWithToleranceOperator.reporters(ReporterManagerUtil.sourceTaskReporters("TEST-CONN-" + i, connectKeyValue));
- final WorkerSourceTask task = new WorkerSourceTask(new ConnectConfig(),
+ final WorkerSourceTask task = new WorkerSourceTask(new WorkerConfig(),
new ConnectorTaskId("TEST-CONN-" + i, i),
new TestSourceTask(),
null,
@@ -147,7 +146,7 @@ public class WorkerTest {
new AtomicReference(WorkerState.STARTED),
connectStatsManager, connectStatsService,
transformChain,
- retryWithToleranceOperator, null));
+ retryWithToleranceOperator, null);
runnables.add(task);
}
worker.setWorkingTasks(runnables);
@@ -170,7 +169,7 @@ public class WorkerTest {
ConnectKeyValue connectKeyValue = new ConnectKeyValue();
connectKeyValue.getProperties().put("key1", "TEST-CONN-" + i + "1");
connectKeyValue.getProperties().put("key2", "TEST-CONN-" + i + "2");
- connectKeyValue.getProperties().put(RuntimeConfigDefine.CONNECTOR_CLASS, TestConnector.class.getName());
+ connectKeyValue.getProperties().put(ConnectorConfig.CONNECTOR_CLASS, TestConnector.class.getName());
connectorConfigs.put("TEST-CONN-" + i, connectKeyValue);
}
@@ -192,17 +191,17 @@ public class WorkerTest {
connectKeyValue.getProperties().put("key2", "TEST-CONN-" + i + "2");
if (i == 1) {
// start direct task
- connectKeyValue.getProperties().put(RuntimeConfigDefine.TASK_TYPE, Worker.TaskType.DIRECT.name());
- connectKeyValue.getProperties().put(RuntimeConfigDefine.SOURCE_TASK_CLASS, TestSourceTask.class.getName());
- connectKeyValue.getProperties().put(RuntimeConfigDefine.SINK_TASK_CLASS, TestSinkTask.class.getName());
+ connectKeyValue.getProperties().put(ConnectorConfig.TASK_TYPE, Worker.TaskType.DIRECT.name());
+ connectKeyValue.getProperties().put(ConnectorConfig.SOURCE_TASK_CLASS, TestSourceTask.class.getName());
+ connectKeyValue.getProperties().put(ConnectorConfig.SINK_TASK_CLASS, TestSinkTask.class.getName());
} else {
- connectKeyValue.getProperties().put(RuntimeConfigDefine.TASK_TYPE, Worker.TaskType.SOURCE.name());
+ connectKeyValue.getProperties().put(ConnectorConfig.TASK_TYPE, Worker.TaskType.SOURCE.name());
}
- connectKeyValue.getProperties().put(RuntimeConfigDefine.TASK_CLASS, TestSourceTask.class.getName());
- connectKeyValue.getProperties().put(RuntimeConfigDefine.SOURCE_RECORD_CONVERTER, TestConverter.class.getName());
- connectKeyValue.getProperties().put(RuntimeConfigDefine.NAMESRV_ADDR, "127.0.0.1:9876");
- connectKeyValue.getProperties().put(RuntimeConfigDefine.RMQ_PRODUCER_GROUP, UUID.randomUUID().toString());
- connectKeyValue.getProperties().put(RuntimeConfigDefine.OPERATION_TIMEOUT, "3000");
+ connectKeyValue.getProperties().put(ConnectorConfig.TASK_CLASS, TestSourceTask.class.getName());
+ connectKeyValue.getProperties().put(ConnectorConfig.VALUE_CONVERTER, TestConverter.class.getName());
+// connectKeyValue.getProperties().put(RuntimeConfigDefine.NAMESRV_ADDR, "127.0.0.1:9876");
+// connectKeyValue.getProperties().put(RuntimeConfigDefine.RMQ_PRODUCER_GROUP, UUID.randomUUID().toString());
+// connectKeyValue.getProperties().put(RuntimeConfigDefine.OPERATION_TIMEOUT, "3000");
connectKeyValues.add(connectKeyValue);
taskConfigs.put("TEST-CONN-" + i, connectKeyValues);
}
diff --git a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/testimpl/TestPositionManageServiceImpl.java b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/testimpl/TestPositionManageServiceImpl.java
index ecc9e25..b1c9717 100644
--- a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/testimpl/TestPositionManageServiceImpl.java
+++ b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/testimpl/TestPositionManageServiceImpl.java
@@ -20,7 +20,7 @@ package org.apache.rocketmq.connect.runtime.connectorwrapper.testimpl;
import io.openmessaging.connector.api.data.RecordOffset;
import java.util.List;
import java.util.Map;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.service.PositionManagementService;
import org.apache.rocketmq.connect.runtime.service.StagingMode;
import org.apache.rocketmq.connect.runtime.store.ExtendRecordPartition;
@@ -76,7 +76,7 @@ public class TestPositionManageServiceImpl implements PositionManagementService
}
- @Override public void initialize(ConnectConfig connectConfig) {
+ @Override public void initialize(WorkerConfig connectConfig) {
}
diff --git a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/controller/distributed/DistributedConnectControllerTest.java b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/controller/distributed/DistributedConnectControllerTest.java
index 2a203e4..0ca330a 100644
--- a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/controller/distributed/DistributedConnectControllerTest.java
+++ b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/controller/distributed/DistributedConnectControllerTest.java
@@ -18,8 +18,8 @@
package org.apache.rocketmq.connect.runtime.controller.distributed;
import java.nio.charset.StandardCharsets;
-import java.util.concurrent.TimeUnit;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.NameServerMocker;
import org.apache.rocketmq.connect.runtime.connectorwrapper.ServerResponseMocker;
import org.apache.rocketmq.connect.runtime.connectorwrapper.testimpl.TestPositionManageServiceImpl;
@@ -52,7 +52,7 @@ public class DistributedConnectControllerTest {
private PositionManagementService positionManagementService = new TestPositionManageServiceImpl();
- private ConnectConfig connectConfig = new ConnectConfig();
+ private WorkerConfig connectConfig = new WorkerConfig();
private ServerResponseMocker nameServerMocker;
diff --git a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/controller/distributed/TestConfigManagementService.java b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/controller/distributed/TestConfigManagementService.java
index 7a62144..b9c77ce 100644
--- a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/controller/distributed/TestConfigManagementService.java
+++ b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/controller/distributed/TestConfigManagementService.java
@@ -21,7 +21,7 @@ import io.openmessaging.connector.api.component.connector.Connector;
import java.util.List;
import java.util.Map;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.controller.isolation.Plugin;
import org.apache.rocketmq.connect.runtime.service.ConfigManagementService;
import org.apache.rocketmq.connect.runtime.service.StagingMode;
@@ -68,7 +68,7 @@ public class TestConfigManagementService implements ConfigManagementService {
}
- @Override public void initialize(ConnectConfig connectConfig, Plugin plugin) {
+ @Override public void initialize(WorkerConfig connectConfig, Plugin plugin) {
}
diff --git a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/rest/RestHandlerTest.java b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/rest/RestHandlerTest.java
index aaee4b2..85bd3ac 100644
--- a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/rest/RestHandlerTest.java
+++ b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/rest/RestHandlerTest.java
@@ -42,8 +42,8 @@ import org.apache.http.util.EntityUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.connect.runtime.controller.distributed.DistributedConnectController;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.apache.rocketmq.connect.runtime.connectorwrapper.TransformChain;
import org.apache.rocketmq.connect.runtime.connectorwrapper.Worker;
import org.apache.rocketmq.connect.runtime.connectorwrapper.WorkerConnector;
@@ -91,7 +91,7 @@ public class RestHandlerTest {
private RestHandler restHandler;
@Mock
- private ConnectConfig connectConfig;
+ private WorkerConfig connectConfig;
@Mock
private SourceTask sourceTask;
@@ -154,12 +154,12 @@ public class RestHandlerTest {
String connectName = "testConnector";
ConnectKeyValue connectKeyValue = new ConnectKeyValue();
- connectKeyValue.put(RuntimeConfigDefine.CONNECTOR_CLASS, "org.apache.rocketmq.connect.runtime.service.TestConnector");
- connectKeyValue.put(RuntimeConfigDefine.SOURCE_RECORD_CONVERTER, "source-record-converter");
+ connectKeyValue.put(ConnectorConfig.CONNECTOR_CLASS, "org.apache.rocketmq.connect.runtime.service.TestConnector");
+ connectKeyValue.put(ConnectorConfig.VALUE_CONVERTER, "source-record-converter");
ConnectKeyValue connectKeyValue1 = new ConnectKeyValue();
- connectKeyValue1.put(RuntimeConfigDefine.CONNECTOR_CLASS, "org.apache.rocketmq.connect.runtime.service.TestConnector");
- connectKeyValue1.put(RuntimeConfigDefine.SOURCE_RECORD_CONVERTER, "source-record-converter1");
+ connectKeyValue1.put(ConnectorConfig.CONNECTOR_CLASS, "org.apache.rocketmq.connect.runtime.service.TestConnector");
+ connectKeyValue1.put(ConnectorConfig.VALUE_CONVERTER, "source-record-converter1");
List<ConnectKeyValue> connectKeyValues = new ArrayList<ConnectKeyValue>(8) {
{
@@ -213,13 +213,13 @@ public class RestHandlerTest {
RetryWithToleranceOperator retryWithToleranceOperator = ReporterManagerUtil.createRetryWithToleranceOperator(connectKeyValue);
retryWithToleranceOperator.reporters(ReporterManagerUtil.sourceTaskReporters("testConnectorName1", connectKeyValue));
- WorkerSourceTask workerSourceTask1 = new WorkerSourceTask(new ConnectConfig(),new ConnectorTaskId("testConnectorName1", 1), sourceTask,null, connectKeyValue, positionManagementServiceImpl, converter, converter, producer, workerState, connectStatsManager, connectStatsService, transformChain,retryWithToleranceOperator, null);
+ WorkerSourceTask workerSourceTask1 = new WorkerSourceTask(new WorkerConfig(),new ConnectorTaskId("testConnectorName1", 1), sourceTask,null, connectKeyValue, positionManagementServiceImpl, converter, converter, producer, workerState, connectStatsManager, connectStatsService, transformChain,retryWithToleranceOperator, null);
// create retry operator
RetryWithToleranceOperator retryWithToleranceOperator02 = ReporterManagerUtil.createRetryWithToleranceOperator(connectKeyValue);
retryWithToleranceOperator02.reporters(ReporterManagerUtil.sourceTaskReporters("testConnectorName2", connectKeyValue));
- WorkerSourceTask workerSourceTask2 = new WorkerSourceTask(new ConnectConfig(), new ConnectorTaskId("testConnectorName2", 1), sourceTask,null, connectKeyValue1, positionManagementServiceImpl, converter, converter, producer, workerState, connectStatsManager, connectStatsService, transformChain,retryWithToleranceOperator02, null);
+ WorkerSourceTask workerSourceTask2 = new WorkerSourceTask(new WorkerConfig(), new ConnectorTaskId("testConnectorName2", 1), sourceTask,null, connectKeyValue1, positionManagementServiceImpl, converter, converter, producer, workerState, connectStatsManager, connectStatsService, transformChain,retryWithToleranceOperator02, null);
workerTasks = new HashSet<Runnable>() {
{
add(workerSourceTask1);
diff --git a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementServiceImplTest.java b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementServiceImplTest.java
index c3f700f..97c59c1 100644
--- a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementServiceImplTest.java
+++ b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementServiceImplTest.java
@@ -30,8 +30,8 @@ import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.connect.runtime.common.ConnAndTaskConfigs;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.apache.rocketmq.connect.runtime.store.KeyValueStore;
import org.apache.rocketmq.connect.runtime.controller.isolation.Plugin;
import org.apache.rocketmq.connect.runtime.utils.TestUtils;
@@ -59,7 +59,7 @@ public class ConfigManagementServiceImplTest {
private KeyValueStore<String, List<ConnectKeyValue>> taskKeyValueStore;
private Set<ConfigManagementService.ConnectorConfigUpdateListener> connectorConfigUpdateListener;
private DataSynchronizer<String, ConnAndTaskConfigs> dataSynchronizer;
- private ConnectConfig connectConfig;
+ private WorkerConfig connectConfig;
@Mock
private DefaultMQProducer producer;
@@ -81,7 +81,7 @@ public class ConfigManagementServiceImplTest {
String consumerGroup = UUID.randomUUID().toString();
String producerGroup = UUID.randomUUID().toString();
- connectConfig = new ConnectConfig();
+ connectConfig = new WorkerConfig();
connectConfig.setHttpPort(8081);
connectConfig.setStorePathRootDir(System.getProperty("user.home") + File.separator + "testConnectorStore");
connectConfig.setRmqConsumerGroup("testConsumerGroup");
@@ -95,8 +95,8 @@ public class ConfigManagementServiceImplTest {
connectConfig.setRmqMessageConsumeTimeout(3 * 1000);
connectKeyValue = new ConnectKeyValue();
- connectKeyValue.put(RuntimeConfigDefine.CONNECTOR_CLASS, "org.apache.rocketmq.connect.runtime.connectorwrapper.testimpl.TestConnector");
- connectKeyValue.put(RuntimeConfigDefine.SOURCE_RECORD_CONVERTER, "source-record-converter");
+ connectKeyValue.put(ConnectorConfig.CONNECTOR_CLASS, "org.apache.rocketmq.connect.runtime.connectorwrapper.testimpl.TestConnector");
+ connectKeyValue.put(ConnectorConfig.VALUE_CONVERTER, "source-record-converter");
doAnswer(new Answer() {
@Override
diff --git a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/service/PositionManagementServiceImplTest.java b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/service/PositionManagementServiceImplTest.java
index 990ec8c..1edd63f 100644
--- a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/service/PositionManagementServiceImplTest.java
+++ b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/service/PositionManagementServiceImplTest.java
@@ -35,7 +35,7 @@ import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.connect.runtime.common.ConnAndTaskConfigs;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.connect.runtime.store.ExtendRecordPartition;
import org.apache.rocketmq.connect.runtime.store.KeyValueStore;
import org.apache.rocketmq.connect.runtime.utils.TestUtils;
@@ -61,7 +61,7 @@ import static org.mockito.Mockito.doAnswer;
@RunWith(MockitoJUnitRunner.class)
public class PositionManagementServiceImplTest {
- private ConnectConfig connectConfig;
+ private WorkerConfig connectConfig;
@Mock
private DefaultMQProducer producer;
@@ -87,7 +87,7 @@ public class PositionManagementServiceImplTest {
private final String namespace = "namespace";
@Before
public void init() throws Exception {
- connectConfig = new ConnectConfig();
+ connectConfig = new WorkerConfig();
connectConfig.setHttpPort(8081);
connectConfig.setNamesrvAddr("localhost:9876");
connectConfig.setStorePathRootDir(System.getProperty("user.home") + File.separator + "testConnectorStore");
diff --git a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/utils/TransferUtilsTest.java b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/utils/TransferUtilsTest.java
index cd3ec7b..3c1f3af 100644
--- a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/utils/TransferUtilsTest.java
+++ b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/utils/TransferUtilsTest.java
@@ -24,7 +24,7 @@ import java.util.Map;
import java.util.Objects;
import org.apache.rocketmq.connect.runtime.common.ConnAndTaskConfigs;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
-import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
+import org.apache.rocketmq.connect.runtime.config.ConnectorConfig;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
@@ -74,8 +74,8 @@ public class TransferUtilsTest {
public void testToJsonStringToConnAndTaskConfigs() {
String connectName = "testConnector";
ConnectKeyValue connectKeyValue = new ConnectKeyValue();
- connectKeyValue.put(RuntimeConfigDefine.CONNECTOR_CLASS, "io.openmessaging.connect.runtime.service.TestConnector");
- connectKeyValue.put(RuntimeConfigDefine.SOURCE_RECORD_CONVERTER, "source-record-converter");
+ connectKeyValue.put(ConnectorConfig.CONNECTOR_CLASS, "io.openmessaging.connect.runtime.service.TestConnector");
+ connectKeyValue.put(ConnectorConfig.VALUE_CONVERTER, "source-record-converter");
List<ConnectKeyValue> connectKeyValues = new ArrayList<ConnectKeyValue>(8) {
{
add(connectKeyValue);
@@ -111,8 +111,8 @@ public class TransferUtilsTest {
ConnectKeyValue connectKeyValue1 = connectorConfigs1.get(connectName);
assertNotNull(connectKeyValue1);
- assertEquals("io.openmessaging.connect.runtime.service.TestConnector", connectKeyValue1.getString(RuntimeConfigDefine.CONNECTOR_CLASS));
- assertEquals("source-record-converter", connectKeyValue1.getString(RuntimeConfigDefine.SOURCE_RECORD_CONVERTER));
+ assertEquals("io.openmessaging.connect.runtime.service.TestConnector", connectKeyValue1.getString(ConnectorConfig.CONNECTOR_CLASS));
+ assertEquals("source-record-converter", connectKeyValue1.getString(ConnectorConfig.VALUE_CONVERTER));
Map<String, List<ConnectKeyValue>> taskConfigs1 = connAndTaskConfigs1.getTaskConfigs();
List<ConnectKeyValue> connectKeyValues1 = taskConfigs1.get(connectName);
@@ -121,8 +121,8 @@ public class TransferUtilsTest {
ConnectKeyValue connectKeyValue2 = connectKeyValues1.get(0);
assertNotNull(connectKeyValue2);
- assertEquals("io.openmessaging.connect.runtime.service.TestConnector", connectKeyValue2.getString(RuntimeConfigDefine.CONNECTOR_CLASS));
- assertEquals("source-record-converter", connectKeyValue2.getString(RuntimeConfigDefine.SOURCE_RECORD_CONVERTER));
+ assertEquals("io.openmessaging.connect.runtime.service.TestConnector", connectKeyValue2.getString(ConnectorConfig.CONNECTOR_CLASS));
+ assertEquals("source-record-converter", connectKeyValue2.getString(ConnectorConfig.VALUE_CONVERTER));
}
}
\ No newline at end of file
diff --git a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/utils/datasync/BrokerBasedLogTest.java b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/utils/datasync/BrokerBasedLogTest.java
index bb09fee..d71d0ef 100644
--- a/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/utils/datasync/BrokerBasedLogTest.java
+++ b/rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/utils/datasync/BrokerBasedLogTest.java
@@ -25,7 +25,7 @@ import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.common.message.Message;
-import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
+import org.apache.rocketmq.connect.runtime.config.WorkerConfig;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.junit.Before;
import org.junit.Test;
@@ -62,14 +62,14 @@ public class BrokerBasedLogTest {
@Mock
private Converter converter;
- private ConnectConfig connectConfig;
+ private WorkerConfig connectConfig;
@Before
public void init() throws IllegalAccessException, NoSuchFieldException {
topicName = "testTopicName";
consumerGroup = "testConsumerGroup1";
producerGroup = "testProducerGroup1";
- connectConfig = new ConnectConfig();
+ connectConfig = new WorkerConfig();
connectConfig.setRmqConsumerGroup(consumerGroup);
connectConfig.setRmqProducerGroup(producerGroup);
connectConfig.setNamesrvAddr("127.0.0.1:9876");
diff --git a/rocketmq-connect-sample/src/main/java/org/apache/rocketmq/connect/file/FilterTransform.java b/rocketmq-connect-sample/src/main/java/org/apache/rocketmq/connect/file/FilterTransform.java
index ffd0b29..7c34705 100644
--- a/rocketmq-connect-sample/src/main/java/org/apache/rocketmq/connect/file/FilterTransform.java
+++ b/rocketmq-connect-sample/src/main/java/org/apache/rocketmq/connect/file/FilterTransform.java
@@ -23,7 +23,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class FilterTransform implements Transform<ConnectRecord> {
-
private static final Logger log = LoggerFactory.getLogger(LoggerName.FILE_CONNECTOR);
private KeyValue keyValue;