RockteMQ-AI commented on code in PR #145:
URL: https://github.com/apache/rocketmq-connect/pull/145#discussion_r3839568052
##########
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());
Review Comment:
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.
##########
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:
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.
##########
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);
Review Comment:
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.
##########
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()){
Review Comment:
Performance: The while-loop iterates all brokers in the target cluster, and
for each broker iterates ALL message queues from the consume stats, only
updating the offset when `brokerName.equals(mq.getBrokerName())`. This is
O(brokers x queues). Build a Map<brokerName, brokerAddress> first, then iterate
the message queues once and look up the matching broker address.
##########
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:
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.
##########
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());
+
connectRecord.addExtension("brokerName",msg.getBrokerName());
+ KeyValue kv = new DefaultKeyValue();
+ kv.put("queueId",msg.getQueueId());
+ connectRecord.addExtension(kv);
Review Comment:
Compatibility: Uses `io.openmessaging.internal.DefaultKeyValue`, an internal
package not part of the public OMS API. This may break in future dependency
versions. Additionally, verify that `connectRecord.addExtension(KeyValue)`
properly merges the int-typed queueId value and that
`getExtensions().getInt("queueId")` in WorkerSourceTask can round-trip it,
since topic and brokerName are added as String extensions but queueId is added
via a KeyValue with an int.
##########
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 {
Review Comment:
startTarMQAdminTool uses source ACL credentials (`isSrcAclEnable()`,
`getSrcAccessKey()`, `getSrcSecretKey()`) for the target cluster admin tool. If
the target cluster has ACL enabled with different credentials than the source,
authentication will fail. TaskConfig should have separate target ACL config
fields, or at minimum this limitation should be documented.
##########
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");
+ String brokerName =
sourceDataEntry.getExtension("brokerName");
+ MessageQueue mq = new
MessageQueue(targetTopic,brokerName,queueId);
Review Comment:
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.
##########
connectors/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/MetaSourceTask.java:
##########
@@ -120,28 +128,38 @@ public void resume() {
List<ConnectRecord> res = new ArrayList<>();
Review Comment:
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.
##########
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 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.
##########
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");
Review Comment:
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.
##########
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:
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.
##########
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:
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.
--
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]
