add replicator checkpoint doc (#486)
modify metrics.conf default config
diff --git a/connectors/rocketmq-replicator/README.md b/connectors/rocketmq-replicator/README.md
index 7d1bec6..21d6494 100644
--- a/connectors/rocketmq-replicator/README.md
+++ b/connectors/rocketmq-replicator/README.md
@@ -74,6 +74,68 @@
}'
````
+同步位点
+
+````
+curl -X POST -H "Content-Type: application/json" http://${runtime-port}:${runtime-ip}/connectors/${replicator-name} -d '{
+ "connector.class": "org.apache.rocketmq.replicator.ReplicatorCheckpointConnector",
+ "src.cluster": "${srcDefaultCluster}",
+ "src.endpoint": "${namesrvEndpoint}",
+ "dest.acl.enable": "false",
+ "src.secret.key": "${sk}",
+ "dest.access.key": "${ak}",
+ "src.topictags": "test1",
+ "sync.gids":"test_gid1",
+ "src.acl.enable": "false",
+ "errors.tolerance": "all",
+ "dest.secret.key": "${sk}",
+ "dest.endpoint": "${namesrvEndpoint}",
+ "src.access.key": "${ak}",
+ "dest.cluster": "${targetDefaultCluster}",
+ "source.cluster": "${sourceDefaultCluster}",
+ "dest.region": "${regionA}",
+ "src.region": "${regionB}",
+ "dest.cloud": "${cloud1}",
+ "source.cloud": "${cloud2}",
+ "key.converter":"org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
+ "value.converter":"org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter"
+}'
+````
+例如
+````
+curl -X POST -H "Content-Type: application/json" http://127.0.0.1:8082/connectors/test_checkpoint_replicator -d '{
+ "connector.class": "org.apache.rocketmq.replicator.ReplicatorCheckpointConnector",
+ "src.endpoint": "127.0.0.1:9876",
+ "src.cluster": "DefaultCluster",
+ "src.region": "regionA",
+ "src.cloud": "cloud1",
+ "dest.acl.enable": "false",
+ "src.topictags": "TopicTest",
+ "src.acl.enable": "false",
+ "dest.endpoint": "127.0.0.2:9876",
+ "dest.region": "regionB",
+ "dest.cluster": "DefaultCluster",
+ "errors.tolerance": "all",
+ "sync.gids":"test_gid1",
+ "dest.cloud": "dest-cloud",
+ "key.converter":"org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
+ "value.converter":"org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter"
+}'
+````
+同步位点任务创建成功以后，可以再目的 RocketMQ 集群 replicator_checkpoint Topic中消费到位点信息
+
+位点信息格式
+
+````
+Struct{consumerGroup=test_gid1,topic=TopicTest,upstreamLastTimestamp=1681962375040,downstreamLastTimestamp=1681439588399,metadata=1682068439879}
+````
+
+parameter | type| description |
+---|-------------|----------------------------------------|
+consumerGroup | String | consumerGroup |
+topic | String | topic |
+upstreamLastTimestamp | long | 源 RocketMQ topic consumerGourp 的消费位点 |
+downstreamLastTimestamp | long | 目的 RocketMQ topic consumerGourp 的消费位点 |
## rocketmq-replicator停止
````
@@ -82,24 +144,28 @@
## rocketmq-replicator参数说明
-parameter | type | must | description | sample value
----|---|------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------------|
-src.endpoint | String | Yes | namesrv address of source rocketmq cluster | 127.0.0.1:9876 |
-src.topictags | String | Yes | source cluster topic and tag，${topic},{tag} | test1,* |
-dest.topic | String | Yes | target cluster topic | test2 |
-dest.endpoint | String | Yes | namesrv address of target rocketmq cluster | 127.0.0.1:9876 |
-max.task | String | No | maximum number of tasks | 2 |
-dest.acl.enable | String | No | acl switch,enumeration value : true/false | false |
-dest.access.key | String | No | please refer to the RocketMQ ACL module，when dest.acl.enable is false, this parameter does not take effect | accesskey |
-dest.secret.key | String | No | please refer to the RocketMQ ACL module，when dest.acl.enable is false, this parameter does not take effect | secretkey |
-src.acl.enable | String | No | acl switch,enumeration value : true/false | true |
-src.access.key | String | No | please refer to the RocketMQ ACL module，when dest.acl.enable is false, this parameter does not take effect | accesskey |
-src.secret.key | String | No | please refer to the RocketMQ ACL module，when dest.acl.enable is false, this parameter does not take effect | secretkey |
+parameter | type | must | description | sample value
+---|---|------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------------|
+src.endpoint | String | Yes | namesrv address of source rocketmq cluster | 127.0.0.1:9876 |
+src.topictags | String | Yes | source cluster topic and tag，${topic},{tag} | test1,* |
+dest.topic | String | Yes | target cluster topic | test2 |
+dest.endpoint | String | Yes | namesrv address of target rocketmq cluster | 127.0.0.1:9876 |
+max.task | String | No | maximum number of tasks | 2 |
+dest.acl.enable | String | No | acl switch,enumeration value : true/false | false |
+dest.access.key | String | No | please refer to the RocketMQ ACL module，when dest.acl.enable is false, this parameter does not take effect | accesskey |
+dest.secret.key | String | No | please refer to the RocketMQ ACL module，when dest.acl.enable is false, this parameter does not take effect | secretkey |
+src.acl.enable | String | No | acl switch,enumeration value : true/false | true |
+src.access.key | String | No | please refer to the RocketMQ ACL module，when dest.acl.enable is false, this parameter does not take effect | accesskey |
+src.secret.key | String | No | please refer to the RocketMQ ACL module，when dest.acl.enable is false, this parameter does not take effect | secretkey |
errors.tolerance | String | No | error tolerance ，enumeration value : all . all means to tolerate all errors, the synchronization message failure will be skipped and error log will be printed. If there is no error tolerance configured, all errors will not be tolerated by default, a synchronization failure occurs, and the task will stop after multiple retries | all |
-src.cluster | String | No | source cluster | DefaultCluster |
-dest.cluster | String | No | target cluster | DefaultCluster |
-src.region | String | No | source region | regionA |
-dest.region | String | No | source region | regionB |
-src.cloud | String | No | source cloud | cloud1 |
-dest.cloud | String | No | source cloud | cloud2 |
+src.cluster | String | No | source cluster | DefaultCluster |
+dest.cluster | String | No | target cluster | DefaultCluster |
+src.region | String | No | source region | regionA |
+dest.region | String | No | source region | regionB |
+src.cloud | String | No | source cloud | cloud1 |
+dest.cloud | String | No | source cloud | cloud2 |
+sync.gids | String | No | consumeGroup | consumerGroup1 |
+key.converter | String | No | key.converter | org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter |
+value.converter | String | No | value converter | org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter |
+
diff --git a/connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/ReplicatorCheckpointTask.java b/connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/ReplicatorCheckpointTask.java
index 6c9946d..5a44ec6 100644
--- a/connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/ReplicatorCheckpointTask.java
+++ b/connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/ReplicatorCheckpointTask.java
@@ -258,7 +258,7 @@
connectorConfig.setSrcCluster(config.getString(connectorConfig.SRC_CLUSTER));
connectorConfig.setSrcInstanceId(config.getString(connectorConfig.SRC_INSTANCEID));
connectorConfig.setSrcEndpoint(config.getString(connectorConfig.SRC_ENDPOINT));
- connectorConfig.setSrcTopicTags(config.getString(connectorConfig.getSrcTopicTags()));
+ connectorConfig.setSrcTopicTags(config.getString(connectorConfig.SRC_TOPICTAGS));
connectorConfig.setDestCloud(config.getString(connectorConfig.DEST_CLOUD));
connectorConfig.setDestRegion(config.getString(connectorConfig.DEST_REGION));
connectorConfig.setDestCluster(config.getString(connectorConfig.DEST_CLUSTER));
diff --git a/connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/ReplicatorSourceTask.java b/connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/ReplicatorSourceTask.java
index af9352a..f20cba4 100644
--- a/connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/ReplicatorSourceTask.java
+++ b/connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/ReplicatorSourceTask.java
@@ -386,7 +386,7 @@
}
public synchronized boolean putPulledQueueOffset(MessageQueue mq, long currentOffset, int needAck, MessageExt msg) {
- log.info("putPulledQueueOffset " + mq + ", currentOffset : " + currentOffset + ", ackCount : " + needAck);
+ log.trace("putPulledQueueOffset " + mq + ", currentOffset : " + currentOffset + ", ackCount : " + needAck);
TreeMap<Long, UnAckMessage> offsets = queue2Offsets.get(mq);
if (offsets == null) {
TreeMap<Long, UnAckMessage> newOffsets = new TreeMap<>();
diff --git a/distribution/conf/metrics.conf b/distribution/conf/metrics.conf
index 45158f7..98c1d6d 100644
--- a/distribution/conf/metrics.conf
+++ b/distribution/conf/metrics.conf
@@ -15,10 +15,11 @@
# metrics reporter class
-metrics.reporter=org.apache.rocketmq.connect.runtime.metrics.RocketMQScheduledReporter
+metrics.reporter=org.apache.rocketmq.connect.metrics.reporter.RocketMQScheduledReporter
# metrics topic
metrics.topic=metrics-topic
+group.id=metrics-gid
# Rocketmq namesrvAddr
name.srv.addr=localhost:9876
