【rocketmq-replicator】双集群单向同步，从集群A未推送到集群B，反而推送回A

[VY121011]

rocketmq-5.3.2集群A
IP: X.X.X.A （/rocketmq/bin/dledger/fast-try.sh）
rocketmq-5.3.2集群B
IP: X.X.X.B （/rocketmq/bin/dledger/fast-try.sh）

rocketmq-connect:
IP: X.X.X.A
单机模式启动
sh bin/connect-standalone.sh -c conf/connect-standalone.conf &

rocketmq-replicator:
ps:参数说明只有四个必填，但是实际启动region，cloud不填会报错500
curl -X POST -H "Content-Type: application/json" http://127.0.0.1:8082/connectors/full_msg_replicator -d '{
"connector.class": "org.apache.rocketmq.replicator.ReplicatorSourceConnector",
"src.endpoint": "X.X.X.A:9876",
"src.cluster": "RaftCluster",
"src.region": "regionA",
"src.cloud": "cloudA",
"src.topictags": "TopicTest,*",
"src.acl.enable": "false",
"max.task": "4",
"dest.endpoint": "X.X.X.B:9876",
"dest.cluster": "RaftCluster",
"dest.region": "regionB",
"dest.cloud": "cloudB",
"dest.topic": "TopicTest",
"dest.acl.enable": "false",
"errors.tolerance": "none"
}'

查看集群A消费组，CID_RMQ_SYS_REPLICATOR_full_msg_replicator开始正常消费，但是集群B的TopicTest的消息数并未增加，且集群A的TopicTest消息数开始增加。
查看connect运行时的日志，其中对应发送消息到的就是集群A，明明已经配置了dest.endpoint到了集群B的路由地址，为什么会出现这种问题？应该如何修改？
org.apache.rocketmq.client.exception.MQBrokerException: CODE: 2 DESC: [PC_SYNCHRONIZED]broker busy, start flow control for a while
BROKER: X.X.X.A:30911
For more information, please visit the url, http://rocketmq.apache.org/docs/faq/
at org.apache.rocketmq.client.impl.MQClientAPIImpl.processSendResponse(MQClientAPIImpl.java:675)
at org.apache.rocketmq.client.impl.MQClientAPIImpl.access$000(MQClientAPIImpl.java:175)
at org.apache.rocketmq.client.impl.MQClientAPIImpl$1.operationComplete(MQClientAPIImpl.java:564)
at org.apache.rocketmq.remoting.netty.ResponseFuture.executeInvokeCallback(ResponseFuture.java:54)
at org.apache.rocketmq.remoting.netty.NettyRemotingAbstract$2.run(NettyRemotingAbstract.java:321)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:829)

[VY121011]

查看历史问题已解决。在启动connect的时候对应的配置文件，连接的应该配置集群B的地址信息
#547 (comment)

[RockteMQ-AI]

As noted in the comments, this was already resolved by referencing #547. Key configuration points:

1. connect-standalone.conf must set rocketmq.namesrv.addr to the destination cluster
2. The replicator connector reads dest.endpoint for routing, but the underlying producer client is initialized from the standalone config

For future reference, the four required parameters listed in the docs are src.endpoint, src.cluster, dest.endpoint, dest.cluster, but src.region and src.cloud also need valid values to avoid HTTP 500 errors during connector creation.

---

Automated evaluation by RockteMQ-AI

[RockteMQ-AI]

Issue Evaluation

Category: bug | Status: Evaluated

This issue reports that unidirectional cluster sync (A→B) is incorrectly pushing data back to cluster A instead of only to cluster B.

Assessment: This appears to be a configuration or replicator routing issue. The replicator's dest.cluster configuration may not be correctly isolating the replication direction.

Recommended investigation:

• Verify dest.cluster and source/destination connector configurations
• Check replicator task assignment and partition routing logic
• Confirm no bidirectional sync tasks are inadvertently created

Please provide the replicator configuration files and task status output (GET /connectors/{name}/status) to help diagnose further.

---

Automated evaluation by RockteMQ-AI

[RockteMQ-AI]

Issue Evaluation

Category: bug | Status: Confirmed

The replicator is syncing in the reverse direction — data from cluster A is being pushed back to A instead of to B. This suggests the source/endpoint configuration is being misinterpreted or the replicator is defaulting to the wrong cluster.

Root Cause: Likely in the endpoint resolution logic of ReplicatorSourceConnector — the src/dst endpoints may be swapped during initialization.
Impact: Data flows in the wrong direction, potentially causing data duplication or loss.
Severity: High

Note: The issue also mentions that region and cloud parameters are documented as optional but cause 500 errors when omitted — this is a separate documentation/validation bug.

---

Automated evaluation by RockteMQ-AI