RockteMQ-AI commented on code in PR #437:
URL: https://github.com/apache/rocketmq-connect/pull/437#discussion_r3839508588
##########
connectors/rocketmq-connect-debezium/rocketmq-connect-debezium-core/src/main/java/org/apache/rocketmq/connect/debezium/RocketMqAdminUtil.java:
##########
@@ -231,9 +230,9 @@ public static Map<MessageQueue, TopicOffset>
offsets(RocketMqConfig config, Stri
DefaultMQAdminExt adminClient = null;
try {
adminClient = RocketMqAdminUtil.startMQAdminTool(config);
- TopicStatsTable topicStatsTable =
adminClient.examineTopicStats(topic);
+ TopicStatsTable topicStatsTable =
examineTopicStats(adminClient,topic);
return topicStatsTable.getOffsetTable();
- } catch (MQClientException | MQBrokerException | RemotingException |
InterruptedException e) {
+ } catch (Exception e) {
throw new RuntimeException(e);
Review Comment:
The offsets() method broadened its catch clause from specific exceptions
(MQClientException | MQBrokerException | RemotingException |
InterruptedException) to catch(Exception e). This can mask programming errors
like NullPointerException or IllegalArgumentException from startMQAdminTool or
getOffsetTable, making debugging harder. Consider keeping the specific
exception types or at least separating RuntimeException from expected checked
exceptions.
##########
metric-exporter/src/main/java/org/apache/rocketmq/connect/metrics/reporter/RocketMQScheduledReporter.java:
##########
@@ -128,27 +128,35 @@ public void config(Map<String, String> configs) {
}
this.topic = configs.get(METRICS_TOPIC);
String groupId = configs.get(GROUP_ID);
- DefaultMQAdminExt defaultMQAdminExt = null;
try {
- defaultMQAdminExt =
RocketMQClientUtil.startMQAdminTool(Boolean.valueOf(configs.get(ACL_ENABLED)),
configs.get(ACCESS_KEY), configs.get(SECRET_KEY), groupId,
configs.get(NAMESRV_ADDR));
- if (!RocketMQClientUtil.topicExist(defaultMQAdminExt, topic)) {
- RocketMQClientUtil.createTopic(defaultMQAdminExt, new
TopicConfig(topic));
+ RocketMqBaseConfiguration baseConfiguration =
RocketMqBaseConfiguration
+ .builder()
+ .namesrvAddr(configs.get(NAMESRV_ADDR))
+ .aclEnable(Boolean.valueOf(configs.get(ACL_ENABLED)))
+ .accessKey(configs.get(ACCESS_KEY))
+ .secretKey(configs.get(SECRET_KEY))
+ .groupId(groupId)
+ .build();
+
+ RocketMqUtils.maybeCreateTopic(baseConfiguration, new
TopicConfig(topic));
+ if
(!RocketMqUtils.fetchAllConsumerGroup(baseConfiguration).contains(groupId)) {
+ RocketMqUtils.createGroup(baseConfiguration, groupId);
}
- if
(!RocketMQClientUtil.fetchAllConsumerGroup(defaultMQAdminExt).contains(groupId))
{
- RocketMQClientUtil.createSubGroup(defaultMQAdminExt, groupId);
- }
- this.producer =
RocketMQClientUtil.initDefaultMQProducer(Boolean.valueOf(configs.get(ACL_ENABLED)),
configs.get(ACCESS_KEY), configs.get(SECRET_KEY), groupId,
configs.get(NAMESRV_ADDR));
+ ProducerConfiguration producerConfiguration = ProducerConfiguration
+ .producerBuilder()
+ .namesrvAddr(configs.get(NAMESRV_ADDR))
+ .aclEnable(Boolean.valueOf(configs.get(ACL_ENABLED)))
+ .accessKey(configs.get(ACCESS_KEY))
+ .secretKey(configs.get(SECRET_KEY))
+ .groupId(groupId)
+ .build();
Review Comment:
ProducerConfiguration is built without setting maxMessageSize or
sendMsgTimeout, leaving them null (Integer). The old code hardcoded
producer.setSendMsgTimeout(5000). If RocketMqUtils.initDefaultMQProducer calls
producer.setSendMsgTimeout(config.getSendMsgTimeout()) with a null Integer,
auto-unboxing will throw NullPointerException. The same risk applies to
maxMessageSize, batchSize, and pollTimeoutMillis in the configuration classes
when not explicitly set by callers.
##########
metric-exporter/src/main/java/org/apache/rocketmq/connect/metrics/reporter/RocketMQScheduledReporter.java:
##########
@@ -128,27 +128,35 @@ public void config(Map<String, String> configs) {
}
this.topic = configs.get(METRICS_TOPIC);
String groupId = configs.get(GROUP_ID);
- DefaultMQAdminExt defaultMQAdminExt = null;
try {
- defaultMQAdminExt =
RocketMQClientUtil.startMQAdminTool(Boolean.valueOf(configs.get(ACL_ENABLED)),
configs.get(ACCESS_KEY), configs.get(SECRET_KEY), groupId,
configs.get(NAMESRV_ADDR));
- if (!RocketMQClientUtil.topicExist(defaultMQAdminExt, topic)) {
- RocketMQClientUtil.createTopic(defaultMQAdminExt, new
TopicConfig(topic));
+ RocketMqBaseConfiguration baseConfiguration =
RocketMqBaseConfiguration
+ .builder()
+ .namesrvAddr(configs.get(NAMESRV_ADDR))
+ .aclEnable(Boolean.valueOf(configs.get(ACL_ENABLED)))
+ .accessKey(configs.get(ACCESS_KEY))
+ .secretKey(configs.get(SECRET_KEY))
+ .groupId(groupId)
+ .build();
+
+ RocketMqUtils.maybeCreateTopic(baseConfiguration, new
TopicConfig(topic));
+ if
(!RocketMqUtils.fetchAllConsumerGroup(baseConfiguration).contains(groupId)) {
+ RocketMqUtils.createGroup(baseConfiguration, groupId);
}
- if
(!RocketMQClientUtil.fetchAllConsumerGroup(defaultMQAdminExt).contains(groupId))
{
- RocketMQClientUtil.createSubGroup(defaultMQAdminExt, groupId);
- }
- this.producer =
RocketMQClientUtil.initDefaultMQProducer(Boolean.valueOf(configs.get(ACL_ENABLED)),
configs.get(ACCESS_KEY), configs.get(SECRET_KEY), groupId,
configs.get(NAMESRV_ADDR));
+ ProducerConfiguration producerConfiguration = ProducerConfiguration
+ .producerBuilder()
+ .namesrvAddr(configs.get(NAMESRV_ADDR))
+ .aclEnable(Boolean.valueOf(configs.get(ACL_ENABLED)))
+ .accessKey(configs.get(ACCESS_KEY))
+ .secretKey(configs.get(SECRET_KEY))
+ .groupId(groupId)
+ .build();
+ this.producer =
RocketMqUtils.initDefaultMQProducer(producerConfiguration);
this.producer.start();
} catch (Exception e) {
log.error("Init config failed ", e);
- } finally {
- if (defaultMQAdminExt != null) {
- defaultMQAdminExt.shutdown();
- }
}
Review Comment:
If config() catches an exception during init, this.producer remains null.
The send() method silently drops all metrics (null check returns without
logging or retrying), and close() also silently skips shutdown. This leads to
silent metric loss with no fail-fast or retry mechanism. Consider rethrowing or
setting a flag to indicate init failure.
##########
connectors/rocketmq-connect-debezium/rocketmq-connect-debezium-core/src/main/java/org/apache/rocketmq/connect/debezium/RocketMqAdminUtil.java:
##########
@@ -242,4 +241,59 @@ public static Map<MessageQueue, TopicOffset>
offsets(RocketMqConfig config, Stri
}
}
+ /**
+ * Compatible with 4.9.4 and earlier
+ *
+ * @param adminClient
+ * @param topic
+ * @return
+ */
+ private static TopicStatsTable examineTopicStats(DefaultMQAdminExt
adminClient, String topic) {
+ try {
+ return adminClient.examineTopicStats(topic);
+ } catch (MQBrokerException e) {
+ // Compatible with 4.9.4 and earlier
+ if (e.getResponseCode() ==
ResponseCode.REQUEST_CODE_NOT_SUPPORTED) {
+ try {
+ return overrideExamineTopicStats(adminClient, topic);
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ } else {
+ throw new RuntimeException(e);
+ }
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
+ /**
+ * Compatible with version 4.9.4
+ *
+ * @param adminClient
+ * @param topic
+ * @return
Review Comment:
overrideExamineTopicStats uses a hardcoded 5000ms timeout for
getTopicStatsInfo(addr, topic, 5000). For topics with many queues or under
network latency, this may be insufficient and cause intermittent failures on
the 4.9.4 compatibility path. Consider making the timeout configurable or
increasing it.
##########
rocketmq-connect-common/src/main/java/org/apache/rocketmq/connect/common/RocketMqUtils.java:
##########
@@ -0,0 +1,533 @@
+/*
Review Comment:
The new rocketmq-connect-common module contains 533+ lines of utility code
(RocketMqUtils.java) including topic creation, group management,
producer/consumer initialization, and admin operations, but no test files are
included in the diff. This is critical infrastructure shared across modules and
should have unit tests, especially for the compatibility fallback paths and
null configuration handling.
##########
connectors/rocketmq-connect-mongo/pom.xml:
##########
@@ -196,7 +196,7 @@
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-openmessaging</artifactId>
- <version>5.1.0</version>
+ <version>4.9.4</version>
Review Comment:
rocketmq-openmessaging is downgraded from 5.1.0 to 4.9.4 for the mongo
connector, which is the opposite direction of the overall upgrade. If this was
intentional (e.g., 5.1.0 incompatibility), it should be documented. If
accidental, it should be corrected to match the intended target version.
##########
rocketmq-connect-common/src/main/java/org/apache/rocketmq/connect/common/ConsumerConfiguration.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.common;
+
+import lombok.Builder;
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+@Setter
+public class ConsumerConfiguration extends RocketMqBaseConfiguration {
+ // consumer
+ private Integer batchSize;
+ private Long pollTimeoutMillis;
+
+
+ @Builder(builderMethodName = "consumerBuilder")
+ public ConsumerConfiguration(String namesrvAddr, String groupId, boolean
aclEnable, String accessKey,
+ String secretKey,
+ Integer batchSize, Long pollTimeoutMillis) {
Review Comment:
ConsumerConfiguration and ProducerConfiguration use Lombok @Builder with
inheritance. The parent class RocketMqBaseConfiguration also has @Builder,
generating a package-private all-args constructor. This works because all
classes share the same package, but it is fragile — if any subclass moves to a
different package in the future, super() calls will fail at compile time.
Consider adding an explicit protected constructor in the base class.
##########
pom.xml:
##########
@@ -57,7 +57,10 @@
<jackson.version>2.13.4.1</jackson.version>
<commons-collections4.version>4.4</commons-collections4.version>
<!-- RocketMQ Version-->
- <rocketmq.version>4.7.1</rocketmq.version>
+ <rocketmq.version>5.1.0</rocketmq.version>
Review Comment:
Inconsistent RocketMQ versions across the project: parent pom declares
rocketmq.version as 5.1.0, but most connectors pin rocketmq-openmessaging to
4.9.4 (activemq, cassandra, deltalake, hudi, jms, kafka, mongo, rabbitmq,
redis, replicator), while debezium uses 5.1.0. When connectors are co-deployed,
this version divergence can cause classpath conflicts and NoClassDefFoundError
from incompatible API changes between 4.x and 5.x packages (e.g.,
common.admin.TopicOffset moved to remoting.protocol.admin.TopicOffset).
##########
rocketmq-connect-common/pom.xml:
##########
@@ -0,0 +1,34 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <parent>
+ <artifactId>rocketmq-connect</artifactId>
+ <groupId>org.apache.rocketmq</groupId>
+ <version>0.0.1-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>rocketmq-connect-common</artifactId>
+
+ <properties>
+ <maven.compiler.source>8</maven.compiler.source>
+ <maven.compiler.target>8</maven.compiler.target>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+ <dependencies>
Review Comment:
The new rocketmq-connect-common module depends on rocketmq-client and
rocketmq-tools without specifying versions, but the parent pom's
dependencyManagement section removed the version entries for these artifacts.
If no BOM import provides these versions, the build will fail with a 'version
missing' error. Either restore the dependencyManagement entries in the parent
pom or add explicit versions here.
--
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]
