rocketmq-replicator 同步消息异常

[LittleBoy18]

1、readme 指导操作有误
2、同步消息到白名单设置的topic

[LittleBoy18]

1、优化了 消息同步
2、优化了 消费进度同步

[tsunghanjacktsai]

Hey @LittleBoy18, instead of RmqSourceTask and MetaSourceTask, how about trying the newer version of rocketmq-replicator, which applies the ReplicatorSourceConnector and ReplicatorSourceTask and provides more complete features?

[RockteMQ-AI]

⚠️ Merge conflict detected

This PR has conflicts with the base branch and cannot be merged. Please rebase or merge the base branch into your branch and resolve the conflicts:

git fetch origin
git checkout fix-syncmessage
git rebase origin/main
# resolve conflicts, then:
git push --force-with-lease

This is a one-time reminder. Feel free to @mention me for a re-review after conflicts are resolved.

---

Automated notification by github-manager-bot

[RockteMQ-AI]

Summary

This PR modifies 8 file(s) with 455 lines of diff. No test changes detected — consider adding test coverage.

---

Automated review by github-manager-bot

[RockteMQ-AI]

No test changes detected alongside source modifications. Consider adding tests to cover the changes.

[RockteMQ-AI]

This PR has been open since May 2022 (over 4 years) and currently has merge conflicts with the master branch.

Status check: Is this replicator sync fix still relevant? The changes to MetaSourceTask, RmqSourceTask, and WorkerSourceTask appear to address real issues, but the code needs to be rebased.

If this is still being worked on, please rebase and update. If abandoned, consider closing.

---

Automated review by github-manager-bot

[RockteMQ-AI]

Summary

PR received and logged for review. This PR requires detailed code review by a maintainer.

Diff size: 455 lines
Author: LittleBoy18 (CONTRIBUTOR)

---

Automated review by RockteMQ-AI

[RockteMQ-AI]

Summary

Review of PR #145: rocketmq-replicator 同步消息异常

Findings: 13 issue(s) identified (3 critical).
CLA: unknown

Please address the inline comments above.

---

Automated review by github-manager-bot

[RockteMQ-AI]

NPE risk: clusterAddrTable.get(this.config.getTargetCluster()) can return null if the target cluster name is misconfigured or not yet registered in the target cluster topology. The subsequent clusterNameSet.iterator() would throw NullPointerException with no useful error message. Add a null check and log a meaningful error identifying the missing cluster.

[RockteMQ-AI]

NPE risk: brokerAddrTable.get(clusterName) may return null if a broker listed in the cluster table is not yet in the broker address table (race during broker add/remove). The subsequent brokerData.getBrokerAddrs() would NPE. Additionally, brokerAddrs.get(new Long(0)) may return null if no master (brokerId=0) is available, causing updateConsumeOffset(null, ...) to silently fail inside the catch block. Add null checks for brokerData, brokerAddrs, and brokerAddresMaster.

[RockteMQ-AI]

NPE risk: sourceDataEntry.getExtensions().getInt("queueId") assumes getExtensions() is non-null and contains the "queueId" key. For any record not produced by RmqSourceTask (which is the only place queueId is set), this throws NPE or NoSuchElementException. Should null-check getExtensions() and provide a safe default or skip the MessageQueue-specific send path.

[RockteMQ-AI]

Performance: this.tarMQAdminExt.examineBrokerClusterInfo() is called inside the per-group loop. This is an expensive RPC that returns the same target cluster topology for every consumer group. Hoist this call outside the for-loop and reuse the result across all groups.

[RockteMQ-AI]

NPE risk in stop(): tarMQAdminExt.shutdown() is called unconditionally. If start() fails after creating srcMQAdminExt but before assigning tarMQAdminExt (e.g., startTarMQAdminTool throws MQClientException), stop() will throw NPE on a null tarMQAdminExt. Should null-check before calling shutdown.

[RockteMQ-AI]

Correctness: new MessageQueue(targetTopic, brokerName, queueId) is constructed unconditionally, but targetTopic and brokerName may be null — the null check at line 316 only guards sourceMessage.setTopic(), not the MessageQueue construction. Sending to a MessageQueue with null topic or broker name will fail at the broker. Validate these fields before entering the RocketMQConverter branch or ensure they are always set by the source task.

[RockteMQ-AI]

The early return (not continue) for RocketMQMetaConverter skips the toSendRecord = null cleanup at the end of sendRecord(). If the framework does not reset toSendRecord each cycle, records could accumulate across poll cycles for meta tasks. Use continue or ensure toSendRecord is cleared outside this method for meta tasks.

[RockteMQ-AI]

Maintainability: The SendCallback implementation (onSuccess/onException with stats tracking and position storage) is duplicated verbatim between the RocketMQConverter branch and the else branch (~40 lines each). Extract this into a helper method to avoid divergence bugs when one copy is updated but not the other.

[RockteMQ-AI]

Offset tracking concern: poll() now always returns an empty ConnectRecord list, and sendRecord() returns early for RocketMQMetaConverter, bypassing positionStorageWriter entirely. The framework's position/offset tracking is completely skipped for meta tasks. On task restart or reassignment, no progress is recorded. Verify this is intentional — the direct updateConsumeOffset approach may need at-least-once delivery guarantees that the framework no longer provides for this path.

[RockteMQ-AI]

No test coverage for the substantially changed meta offset sync logic (direct updateConsumeOffset, new startTarMQAdminTool, ClusterInfo traversal) or the RocketMQConverter send-to-specific-queue path in WorkerSourceTask. These are significant behavioral changes that warrant integration tests, especially given the NPE risks identified above.

[RockteMQ-AI]

Summary

12 finding(s) to address.

Findings

• [CRITICAL] rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTask.java:320 — Queue routing is applied unconditionally to every record in the null == recordConverter || recordConverter instanceof RocketMQConverter branch. ConnectRecord.getExtensions() returns null unless addExtension() was called (putExtendMsgProperty in this same method null-checks it), so sourceDataEntry.getExtensions().getInt("queueId") throws NPE for any record without extensions. Even when extensions exist but lack queueId/brokerName/topic, DefaultKeyValue.getInt() returns 0 and brokerName/targetTopic are null, so new MessageQueue(null, null, 0) is passed to producer.send() and the send fails. This is shared runtime code: every existing source connector that sets no converter or uses RocketMQConverter (not just the replicator) breaks. Guard this block (only route to a MessageQueue when topic+brokerName+queueId are all present) and fall back to producer.send(sourceMessage) otherwise.
• [WARNING] rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTask.java:270 — The RocketMQMetaConverter early-return sits inside the per-record loop and skips the toSendRecord = null cleanup at the end of sendRecord(). In run(), poll() is only invoked when toSendRecord is empty, so any meta-converter task that returns at least one record would busy-spin forever (100% CPU, no further polls, no position commits). Today MetaSourceTask.poll() always returns an empty list so this is dead code, but the check is converter-scoped, not record-scoped — move it before the loop (or clear the list) so the invariant can't be broken by a future task that emits records with this converter.
• [WARNING] connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/RmqSourceTask.java:182 — Adding routing metadata as plain extensions corrupts replicated messages: original message properties are copied to extensions first (line 181), then addExtension("topic"/"brokerName"/"queueId") overwrites any same-named user property, silently losing source data. Additionally, putExtendMsgProperty writes all extensions back as message properties (connect-ext-topic, connect-ext-brokerName, connect-ext-queueId), so every replicated message gains three properties the original never had. Use a dedicated prefix (e.g. connect-internal-*) or a separate mechanism for routing metadata instead of the generic "topic" key, which also risks hijacking the routing of any other connector whose records happen to carry a "topic" extension.
• [WARNING] connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/MetaSourceTask.java:139 — clusterAddrTable.get(this.config.getTargetCluster()) returns null when target-cluster is not configured (older connector configs; DefaultKeyValue.put even stores null as the string "null") and clusterNameSet.iterator() NPEs. The broad catch swallows it with the misleading message "admin get consumer info failed" and offset sync silently does nothing — an operational trap. Validate targetCluster/targetRocketmq in validate()/start() and fail fast, and null-check the lookup with an explicit error.
• [WARNING] connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/common/Utils.java:216 — startTarMQAdminTool connects to taskConfig.getTargetRocketmq() but authenticates with SOURCE ACL credentials (isSrcAclEnable/getSrcAccessKey/getSrcSecretKey). TaskConfig has no target ACL fields even though RmqConnectorConfig does (see startTargetMQAdminTool using isTargetAclEnable), so with ACL enabled on the target cluster under different credentials every admin call fails. Also this method is a near-duplicate of startMQAdminTool/startTargetMQAdminTool — add target ACL fields to TaskConfig and reuse one implementation.
• [WARNING] connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/MetaSourceTask.java:151 — Offset sync matches the TARGET broker name against the SOURCE MessageQueue broker name, and WorkerSourceTask similarly sends to the same queueId/brokerName — the whole change assumes the target cluster has identical broker names and at least as many queues per topic. When names or queue counts differ, offsets are silently skipped here and data sends throw MQClientException in the runtime (records dropped). Also brokerAddrs.get(new Long(0)) returns null when no master is registered, making updateConsumeOffset fail. Validate the target route exists and log a clear warning otherwise; use Long.valueOf(0)/MixAll.MASTER_ID instead of new Long(0).
• [WARNING] connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/MetaSourceTask.java:102 — stop() calls tarMQAdminExt.shutdown() without a null guard. If start() throws after srcMQAdminExt started (e.g. startTarMQAdminTool failure), tarMQAdminExt is null and stop() NPEs, masking the real error, while the already-started srcMQAdminExt is never shut down (resource leak). Guard both shutdowns and release srcMQAdminExt when the target admin fails to start.
• [WARNING] connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/MetaSourceTask.java:115 — No tests accompany the change: the new direct updateConsumeOffset path (target-cluster lookup, broker matching, error handling), the new task-split keys in Utils.groupPartitions, and the runtime queue-routing branch are all untested, and existing tests (RmqSourceReplicatorTest, DefaultTaskDivideStrategyTest) are untouched. Please add unit tests at least for targetCluster-missing/unknown behavior and for sendRecord() with and without queueId/brokerName extensions.
• [INFO] connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/MetaSourceTask.java:135 — examineBrokerClusterInfo() is an expensive admin RPC invoked inside the per-group loop on every poll, and poll() is called in a tight runtime loop with no sleep, so cluster info is re-fetched G times per iteration. Hoist the call outside the groups loop and cache it (or refresh periodically); the nested group × broker × queue iteration can also be inverted so each queue is matched against a broker-name map.
• [INFO] connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/MetaSourceTask.java:164 — poll() now always returns an empty list — the record/schema machinery (ConnectRecord, Schema, Field, SchemaBuilder, JSONObject, FieldName, SchemaEnum imports, and the res variable) is dead code and poll() is purely side-effecting. Note also that OffsetSyncStore.sync() is never invoked (its consumer is never started), so convertTargetOffset degenerates to identity and updateConsumeOffset copies source offsets verbatim — wrong if target offsets diverge (e.g. replication didn't start at offset 0). Clean up the dead code and document that offset.sync.topic is no longer used by this path.
• [INFO] rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTask.java:370 — The async send/callback/error-handling block (~90 lines) is now duplicated between the RocketMQConverter branch and the JSON branch. Extract a shared send-with-callback helper taking the optional MessageQueue so fixes to offset-commit-on-success or failure stats don't have to be applied twice (the copy already risks divergence).
• [INFO] connectors/rocketmq-replicator/README.md:47 — The meta replicator example sets source-cluster and target-cluster to the same value ("test1-rocketmq"), which is almost certainly a copy-paste mistake and confusing for users setting up cross-cluster replication; also the example still documents offset.sync.topic even though offset sync no longer flows through a topic (direct admin updates now).

---

Automated review by github-manager-bot

Additional notes (not anchored to a changed line)

• [WARNING] connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/MetaSourceTask.java:115 — No tests accompany the change: the new direct updateConsumeOffset path (target-cluster lookup, broker matching, error handling), the new task-split keys in Utils.groupPartitions, and the runtime queue-routing branch are all untested, and existing tests (RmqSourceReplicatorTest, DefaultTaskDivideStrategyTest) are untouched. Please add unit tests at least for targetCluster-missing/unknown behavior and for sendRecord() with and without queueId/brokerName extensions. (line outside diff)

[RockteMQ-AI]

Queue routing is applied unconditionally to every record in the null == recordConverter || recordConverter instanceof RocketMQConverter branch. ConnectRecord.getExtensions() returns null unless addExtension() was called (putExtendMsgProperty in this same method null-checks it), so sourceDataEntry.getExtensions().getInt("queueId") throws NPE for any record without extensions. Even when extensions exist but lack queueId/brokerName/topic, DefaultKeyValue.getInt() returns 0 and brokerName/targetTopic are null, so new MessageQueue(null, null, 0) is passed to producer.send() and the send fails. This is shared runtime code: every existing source connector that sets no converter or uses RocketMQConverter (not just the replicator) breaks. Guard this block (only route to a MessageQueue when topic+brokerName+queueId are all present) and fall back to producer.send(sourceMessage) otherwise.

[RockteMQ-AI]

The RocketMQMetaConverter early-return sits inside the per-record loop and skips the toSendRecord = null cleanup at the end of sendRecord(). In run(), poll() is only invoked when toSendRecord is empty, so any meta-converter task that returns at least one record would busy-spin forever (100% CPU, no further polls, no position commits). Today MetaSourceTask.poll() always returns an empty list so this is dead code, but the check is converter-scoped, not record-scoped — move it before the loop (or clear the list) so the invariant can't be broken by a future task that emits records with this converter.

[RockteMQ-AI]

Adding routing metadata as plain extensions corrupts replicated messages: original message properties are copied to extensions first (line 181), then addExtension("topic"/"brokerName"/"queueId") overwrites any same-named user property, silently losing source data. Additionally, putExtendMsgProperty writes all extensions back as message properties (connect-ext-topic, connect-ext-brokerName, connect-ext-queueId), so every replicated message gains three properties the original never had. Use a dedicated prefix (e.g. connect-internal-*) or a separate mechanism for routing metadata instead of the generic "topic" key, which also risks hijacking the routing of any other connector whose records happen to carry a "topic" extension.

[RockteMQ-AI]

clusterAddrTable.get(this.config.getTargetCluster()) returns null when target-cluster is not configured (older connector configs; DefaultKeyValue.put even stores null as the string "null") and clusterNameSet.iterator() NPEs. The broad catch swallows it with the misleading message "admin get consumer info failed" and offset sync silently does nothing — an operational trap. Validate targetCluster/targetRocketmq in validate()/start() and fail fast, and null-check the lookup with an explicit error.

[RockteMQ-AI]

startTarMQAdminTool connects to taskConfig.getTargetRocketmq() but authenticates with SOURCE ACL credentials (isSrcAclEnable/getSrcAccessKey/getSrcSecretKey). TaskConfig has no target ACL fields even though RmqConnectorConfig does (see startTargetMQAdminTool using isTargetAclEnable), so with ACL enabled on the target cluster under different credentials every admin call fails. Also this method is a near-duplicate of startMQAdminTool/startTargetMQAdminTool — add target ACL fields to TaskConfig and reuse one implementation.

[RockteMQ-AI]

stop() calls tarMQAdminExt.shutdown() without a null guard. If start() throws after srcMQAdminExt started (e.g. startTarMQAdminTool failure), tarMQAdminExt is null and stop() NPEs, masking the real error, while the already-started srcMQAdminExt is never shut down (resource leak). Guard both shutdowns and release srcMQAdminExt when the target admin fails to start.

[RockteMQ-AI]

examineBrokerClusterInfo() is an expensive admin RPC invoked inside the per-group loop on every poll, and poll() is called in a tight runtime loop with no sleep, so cluster info is re-fetched G times per iteration. Hoist the call outside the groups loop and cache it (or refresh periodically); the nested group × broker × queue iteration can also be inverted so each queue is matched against a broker-name map.

[RockteMQ-AI]

poll() now always returns an empty list — the record/schema machinery (ConnectRecord, Schema, Field, SchemaBuilder, JSONObject, FieldName, SchemaEnum imports, and the res variable) is dead code and poll() is purely side-effecting. Note also that OffsetSyncStore.sync() is never invoked (its consumer is never started), so convertTargetOffset degenerates to identity and updateConsumeOffset copies source offsets verbatim — wrong if target offsets diverge (e.g. replication didn't start at offset 0). Clean up the dead code and document that offset.sync.topic is no longer used by this path.

[RockteMQ-AI]

The async send/callback/error-handling block (~90 lines) is now duplicated between the RocketMQConverter branch and the JSON branch. Extract a shared send-with-callback helper taking the optional MessageQueue so fixes to offset-commit-on-success or failure stats don't have to be applied twice (the copy already risks divergence).

[RockteMQ-AI]

The meta replicator example sets source-cluster and target-cluster to the same value ("test1-rocketmq"), which is almost certainly a copy-paste mistake and confusing for users setting up cross-cluster replication; also the example still documents offset.sync.topic even though offset sync no longer flows through a topic (direct admin updates now).