RockteMQ-AI commented on code in PR #145:
URL: https://github.com/apache/rocketmq-connect/pull/145#discussion_r3909712744
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTask.java:
##########
@@ -307,7 +312,53 @@ private void sendRecord() throws InterruptedException,
RemotingException, MQClie
log.error("Send record, message size is greater than
{} bytes, sourceDataEntry: {}", RuntimeConfigDefine.MAX_MESSAGE_SIZE,
JSON.toJSONString(sourceDataEntry));
continue;
}
+ String targetTopic = sourceDataEntry.getExtension("topic");
+ if (targetTopic != null){
+ sourceMessage.setTopic(targetTopic);
+ }
sourceMessage.setBody(messageBody);
+ int queueId =
sourceDataEntry.getExtensions().getInt("queueId");
Review Comment:
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.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTask.java:
##########
@@ -265,6 +267,9 @@ public void cleanup() {
*/
private void sendRecord() throws InterruptedException, RemotingException,
MQClientException {
for (ConnectRecord sourceDataEntry : toSendRecord) {
+ if (recordConverter instanceof RocketMQMetaConverter){
Review Comment:
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.
##########
connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/MetaSourceTask.java:
##########
@@ -120,28 +128,38 @@ public void resume() {
List<ConnectRecord> res = new ArrayList<>();
for (String group : groups) {
ConsumeStats stats;
+ String brokerAddresMaster="";
+ String brokerName="";
try {
stats = this.srcMQAdminExt.examineConsumeStats(group);
+ ClusterInfo clusterInfo =
this.tarMQAdminExt.examineBrokerClusterInfo();
+ HashMap<String, Set<String>> clusterAddrTable =
clusterInfo.getClusterAddrTable();
+ HashMap<String, BrokerData> brokerAddrTable =
clusterInfo.getBrokerAddrTable();
+ Set<String> clusterNameSet =
clusterAddrTable.get(this.config.getTargetCluster());
+ Iterator<String> it = clusterNameSet.iterator();
+ while (it.hasNext()){
+ String clusterName = it.next();
+ BrokerData brokerData = brokerAddrTable.get(clusterName);
+ HashMap<Long, String> brokerAddrs =
brokerData.getBrokerAddrs();
+ brokerAddresMaster = brokerAddrs.get(new Long(0));
+ brokerName = brokerData.getBrokerName();
+ for (Map.Entry<MessageQueue, OffsetWrapper> offsetTable :
stats.getOffsetTable().entrySet()) {
+ MessageQueue mq = offsetTable.getKey();
+ long srcOffset =
offsetTable.getValue().getConsumerOffset();
+ long targetOffset = this.store.convertTargetOffset(mq,
group, srcOffset);
+ try{
+ if (brokerName.equals(mq.getBrokerName())){
Review Comment:
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).
##########
connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/MetaSourceTask.java:
##########
@@ -120,28 +128,38 @@ public void resume() {
List<ConnectRecord> res = new ArrayList<>();
for (String group : groups) {
ConsumeStats stats;
+ String brokerAddresMaster="";
+ String brokerName="";
try {
stats = this.srcMQAdminExt.examineConsumeStats(group);
+ ClusterInfo clusterInfo =
this.tarMQAdminExt.examineBrokerClusterInfo();
Review Comment:
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.
##########
connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/MetaSourceTask.java:
##########
@@ -120,28 +128,38 @@ public void resume() {
List<ConnectRecord> res = new ArrayList<>();
for (String group : groups) {
ConsumeStats stats;
+ String brokerAddresMaster="";
+ String brokerName="";
try {
stats = this.srcMQAdminExt.examineConsumeStats(group);
+ ClusterInfo clusterInfo =
this.tarMQAdminExt.examineBrokerClusterInfo();
+ HashMap<String, Set<String>> clusterAddrTable =
clusterInfo.getClusterAddrTable();
+ HashMap<String, BrokerData> brokerAddrTable =
clusterInfo.getBrokerAddrTable();
+ Set<String> clusterNameSet =
clusterAddrTable.get(this.config.getTargetCluster());
+ Iterator<String> it = clusterNameSet.iterator();
Review Comment:
`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.
##########
connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/RmqSourceTask.java:
##########
@@ -178,6 +179,11 @@ private List<ConnectRecord> pollCommonMessage() {
final Map<String, String> properties =
msg.getProperties();
final Set<String> keys = properties.keySet();
keys.forEach(key ->
connectRecord.addExtension(key, properties.get(key)));
+
connectRecord.addExtension("topic",taskTopicConfig.getTargetTopic());
Review Comment:
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.
##########
connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/MetaSourceTask.java:
##########
@@ -92,6 +99,7 @@ public void stop() {
started = false;
}
srcMQAdminExt.shutdown();
+ tarMQAdminExt.shutdown();
Review Comment:
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.
##########
connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/common/Utils.java:
##########
@@ -194,14 +196,30 @@ public static DefaultMQAdminExt startTargetMQAdminTool(
}
public static DefaultMQAdminExt startMQAdminTool(TaskConfig taskConfig)
throws MQClientException {
+ RPCHook rpcHook = null;
+ if (taskConfig.isSrcAclEnable()) {
+ rpcHook = new AclClientRPCHook(new
SessionCredentials(taskConfig.getSrcAccessKey(), taskConfig.getSrcSecretKey()));
+ }
+ DefaultMQAdminExt sourceMQAdminExt = new DefaultMQAdminExt(rpcHook);
+ sourceMQAdminExt.setNamesrvAddr(taskConfig.getSourceRocketmq());
+
sourceMQAdminExt.setAdminExtGroup(ConstDefine.REPLICATOR_TASK_ADMIN_GROUP);
+
sourceMQAdminExt.setInstanceName(Utils.createUniqInstanceName(taskConfig.getSourceRocketmq()));
+
+ sourceMQAdminExt.start();
+ log.info("Source: RocketMQ sourceMQAdminExt started.");
+
+ return sourceMQAdminExt;
+ }
+
+ public static DefaultMQAdminExt startTarMQAdminTool(TaskConfig taskConfig)
throws MQClientException {
RPCHook rpcHook = null;
if (taskConfig.isSrcAclEnable()) {
Review Comment:
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.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTask.java:
##########
@@ -316,45 +367,45 @@ private void sendRecord() throws InterruptedException,
RemotingException, MQClie
continue;
}
sourceMessage.setBody(messageBody);
- }
- try {
- producer.send(sourceMessage, new SendCallback() {
- @Override public void
onSuccess(org.apache.rocketmq.client.producer.SendResult result) {
- log.info("Successful send message to RocketMQ:{},
Topic {}", result.getMsgId(), result.getMessageQueue().getTopic());
- connectStatsManager.incSourceRecordWriteTotalNums();
-
connectStatsManager.incSourceRecordWriteNums(taskConfig.getString(RuntimeConfigDefine.TASK_ID));
- RecordPartition partition = position.getPartition();
- try {
- if (null != partition && null != position) {
- Map<String, String> offsetMap = (Map<String,
String>) offset.getOffset();
-
offsetMap.put(RuntimeConfigDefine.UPDATE_TIMESTAMP,
String.valueOf(sourceDataEntry.getTimestamp()));
- positionStorageWriter.putPosition(partition,
offset);
+ try {
Review Comment:
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).
##########
connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/MetaSourceTask.java:
##########
@@ -120,28 +128,38 @@ public void resume() {
List<ConnectRecord> res = new ArrayList<>();
for (String group : groups) {
ConsumeStats stats;
+ String brokerAddresMaster="";
+ String brokerName="";
try {
stats = this.srcMQAdminExt.examineConsumeStats(group);
+ ClusterInfo clusterInfo =
this.tarMQAdminExt.examineBrokerClusterInfo();
+ HashMap<String, Set<String>> clusterAddrTable =
clusterInfo.getClusterAddrTable();
+ HashMap<String, BrokerData> brokerAddrTable =
clusterInfo.getBrokerAddrTable();
+ Set<String> clusterNameSet =
clusterAddrTable.get(this.config.getTargetCluster());
+ Iterator<String> it = clusterNameSet.iterator();
+ while (it.hasNext()){
+ String clusterName = it.next();
+ BrokerData brokerData = brokerAddrTable.get(clusterName);
+ HashMap<Long, String> brokerAddrs =
brokerData.getBrokerAddrs();
+ brokerAddresMaster = brokerAddrs.get(new Long(0));
+ brokerName = brokerData.getBrokerName();
+ for (Map.Entry<MessageQueue, OffsetWrapper> offsetTable :
stats.getOffsetTable().entrySet()) {
+ MessageQueue mq = offsetTable.getKey();
+ long srcOffset =
offsetTable.getValue().getConsumerOffset();
+ long targetOffset = this.store.convertTargetOffset(mq,
group, srcOffset);
+ try{
+ if (brokerName.equals(mq.getBrokerName())){
+
this.tarMQAdminExt.updateConsumeOffset(brokerAddresMaster,group,mq,targetOffset);
+ }
+ }catch (Exception e){
+ log.error("admin update consumer offset err", e);
+ }
+ }
+ }
} catch (Exception e) {
log.error("admin get consumer info failed for consumer groups:
" + group, e);
continue;
}
-
- for (Map.Entry<MessageQueue, OffsetWrapper> offsetTable :
stats.getOffsetTable().entrySet()) {
- MessageQueue mq = offsetTable.getKey();
- long srcOffset = offsetTable.getValue().getConsumerOffset();
- long targetOffset = this.store.convertTargetOffset(mq, group,
srcOffset);
-
- List<Field> fields = new ArrayList<Field>();
- Schema schema = new Schema(SchemaEnum.OFFSET.name(),
FieldType.INT64, fields);
- schema.getFields().add(new Field(0, FieldName.OFFSET.getKey(),
SchemaBuilder.string().build()));
-
- JSONObject jsonObject = new JSONObject();
- jsonObject.put(FieldName.OFFSET.getKey(), targetOffset);
- ConnectRecord connectRecord = new
ConnectRecord(Utils.offsetKey(mq),
- Utils.offsetValue(srcOffset), System.currentTimeMillis(),
schema, jsonObject.toJSONString());
- res.add(connectRecord);
- }
}
return res;
Review Comment:
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.
##########
connectors/rocketmq-replicator/README.md:
##########
@@ -44,7 +44,7 @@
http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-replicator-name}/stop
注：此功能尚不成熟还需要后续版本优化
````
http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-replicator-name}
-?config={"connector-class":"org.apache.rocketmq.replicator.RmqMetaReplicator","source-rocketmq":"xxxx:9876","target-rocketmq":"xxxxxxx:9876","replicator-store-topic":"replicatorTopic","offset.sync.topic":"syncTopic","taskDivideStrategy":"0","white-list":"TopicTest,TopicTest2","task-parallelism":"2","source-record-converter":"org.apache.rocketmq.connect.runtime.converter.JsonConverter"}
+?config={"connector-class":"org.apache.rocketmq.replicator.RmqMetaReplicator","source-rocketmq":"xxxx:9876","target-rocketmq":"xxxxxxx:9876","target-cluster":"test1-rocketmq","source-cluster":"test1-rocketmq","replicator-store-topic":"replicatorTopic","offset.sync.topic":"syncTopic","taskDivideStrategy":"0","white-list":"TestGroup","task-parallelism":"2","source-record-converter":"org.apache.rocketmq.connect.runtime.converter.RocketMQMetaConverter"}
Review Comment:
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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
