[ISSUE #436]BrokerBasedLog supports initialization of compaction topic by sunxiaojian · Pull Request #437 · apache/rocketmq-connect

sunxiaojian · 2023-03-02T12:04:55Z

1.Upgrade the rocketmq client version to 5.1.0
2.BrokerBasedLog supports initialization of compaction topic
3. Extend rocketmq-connect-common module

… the BrokerBasedLog apache#436

…n/rocketmq-connect into fixed-create-compaction-topic

sunxiaojian · 2023-03-08T12:23:06Z

#438 #436

odbozhou · 2023-04-06T05:46:08Z

There are some incompatibility issues due to the dependent rocketmq versions 4.x and 5.x.
It is suggested that this pr plan will be merged after releasing a version based on the current 4.x version.
4.x and 5.x evolve in two versions

RockteMQ-AI · 2026-07-30T10:45:24Z

⚠️ 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 fixed-create-compaction-topic
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 94 file(s) with 3580 lines of diff. Changes look reasonable.

Automated review by github-manager-bot

Additional notes (not anchored to a changed line)

[INFO] README.md:1 — Large diff (3580 lines). Consider breaking into smaller, focused PRs for easier review. (line outside diff)

RockteMQ-AI

Summary

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

Diff size: 3580 lines
Author: sunxiaojian (CONTRIBUTOR)

Automated review by RockteMQ-AI

RockteMQ-AI

Summary

Review of PR #437: [ISSUE #436]BrokerBasedLog supports initialization of compaction topic

Findings: 9 issue(s) identified (0 critical).
CLA: unknown

Please address the inline comments above.

Automated review by github-manager-bot

RockteMQ-AI · 2026-08-23T20:23:39Z

+                .accessKey(configs.get(ACCESS_KEY))
+                .secretKey(configs.get(SECRET_KEY))
+                .groupId(groupId)
+                .build();


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.

RockteMQ-AI · 2026-08-23T20:23:39Z

-            if (defaultMQAdminExt != null) {
-                defaultMQAdminExt.shutdown();
-            }
        }


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.

RockteMQ-AI · 2026-08-23T20:23:39Z

            return topicStatsTable.getOffsetTable();
-        } catch (MQClientException | MQBrokerException | RemotingException | InterruptedException e) {
+        } catch (Exception e) {
            throw new RuntimeException(e);


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.

RockteMQ-AI · 2026-08-23T20:23:39Z

+     *
+     * @param adminClient
+     * @param topic
+     * @return


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.

RockteMQ-AI · 2026-08-23T20:23:39Z

+        <maven.compiler.target>8</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+    <dependencies>


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.

RockteMQ-AI · 2026-08-23T20:23:39Z

@@ -0,0 +1,533 @@
+/*


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.

RockteMQ-AI · 2026-08-23T20:23:39Z

        <commons-collections4.version>4.4</commons-collections4.version>
        <!-- RocketMQ Version-->
-        <rocketmq.version>4.7.1</rocketmq.version>
+        <rocketmq.version>5.1.0</rocketmq.version>


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).

RockteMQ-AI · 2026-08-23T20:23:39Z

            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-openmessaging</artifactId>
-            <version>5.1.0</version>
+            <version>4.9.4</version>


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.

RockteMQ-AI · 2026-08-23T20:23:39Z

+    @Builder(builderMethodName = "consumerBuilder")
+    public ConsumerConfiguration(String namesrvAddr, String groupId, boolean aclEnable, String accessKey,
+        String secretKey,
+        Integer batchSize, Long pollTimeoutMillis) {


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.

RockteMQ-AI

Summary

Re-review of PR #437 after new commits. This is a significant upgrade PR that:

Upgrades RocketMQ version from 4.x to 5.1.0 across all connectors
Adds compatibility layer for both 4.9.4 and 5.x versions
Updates imports from org.apache.rocketmq.common.* to org.apache.rocketmq.remoting.protocol.*
Adds compaction topic support in BrokerBasedLog

Review notes:

The version upgrade is necessary for staying current with RocketMQ releases
The compatibility layer (examineTopicStats wrapper) is a good approach for supporting both versions
Import changes are mechanical and consistent across modules

Concerns:

This is a large PR (94 files) — consider splitting into smaller, focused PRs for easier review
No test coverage visible for the new compatibility layer
The catch (Exception e) in offsets() method is too broad — consider catching specific exceptions

Suggestions

Add unit tests for the examineTopicStats() compatibility wrapper
Consider splitting the version upgrade into separate PRs per connector module
Narrow the exception handling in RocketMqAdminUtil.offsets()

Automated review by github-manager-bot

sunxiaojian added 19 commits August 31, 2022 18:22

fixed null pointer exception apache#294

73bc139

Merge branch 'master' of github.com:sunxiaojian/rocketmq-connect

8fa5457

Merge branch 'apache:master' into master

312d33c

Merge branch 'apache:master' into master

3278370

Merge branch 'apache:master' into master

6b3b798

Fix invalid offset submitted by sinktask apache#310

39b8eae

Merge branch 'apache:master' into master

0918f10

Merge branch 'apache:master' into master

2b94a7a

Merge branch 'apache:master' into master

63f815d

pull update

c137fab

Merge branch 'apache:master' into master

19ad60c

Merge branch 'apache:master' into master

bfd8e62

Merge branch 'apache:master' into master

faf0fba

Merge branch 'apache:master' into master

f6b65d2

Merge branch 'apache:master' into master

507770d

Merge branch 'apache:master' into master

1336681

Merge branch 'apache:master' into master

910b7da

Merge branch 'apache:master' into master

ab0ad85

Merge branch 'apache:master' into master

6ffc39e

sunxiaojian force-pushed the fixed-create-compaction-topic branch 6 times, most recently from 74be2e6 to 040613f Compare March 3, 2023 06:16

Adding the COMPACTION attribute is supported when creating a topic in…

47f0aac

… the BrokerBasedLog apache#436

sunxiaojian force-pushed the fixed-create-compaction-topic branch from 7ca39f5 to 47f0aac Compare March 7, 2023 13:02

sunxiaojian added 3 commits March 7, 2023 21:05

Merge branch 'master' into fixed-create-compaction-topic

80b4d74

fixed

cc040b2

Merge branch 'fixed-create-compaction-topic' of github.com:sunxiaojia…

f697554

…n/rocketmq-connect into fixed-create-compaction-topic

sunxiaojian force-pushed the fixed-create-compaction-topic branch from 905517d to 54fda9b Compare March 8, 2023 02:08

upgrade rocketmq client version to 5.1.0

1f5080b

sunxiaojian force-pushed the fixed-create-compaction-topic branch from 17d572b to 1f5080b Compare March 8, 2023 03:44

extend rocketmq-connect-common

01d16c8

sunxiaojian force-pushed the fixed-create-compaction-topic branch from 1b9fd88 to 01d16c8 Compare March 8, 2023 11:48

update README.md

da77b22

sunxiaojian force-pushed the fixed-create-compaction-topic branch from 08c4bbf to da77b22 Compare March 8, 2023 12:21

RockteMQ-AI reviewed Aug 5, 2026

View reviewed changes

RockteMQ-AI reviewed Aug 20, 2026

View reviewed changes

RockteMQ-AI reviewed Aug 23, 2026

View reviewed changes

RockteMQ-AI reviewed Aug 30, 2026

View reviewed changes

Uh oh!

Conversation

sunxiaojian commented Mar 2, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

sunxiaojian commented Mar 8, 2023

Uh oh!

odbozhou commented Apr 6, 2023

Uh oh!

RockteMQ-AI commented Jul 30, 2026

Uh oh!

RockteMQ-AI left a comment

Choose a reason for hiding this comment

Summary

Additional notes (not anchored to a changed line)

Uh oh!

RockteMQ-AI left a comment

Choose a reason for hiding this comment

Summary

Uh oh!

RockteMQ-AI left a comment

Choose a reason for hiding this comment

Summary

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

RockteMQ-AI left a comment

Choose a reason for hiding this comment

Summary

Suggestions

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

sunxiaojian commented Mar 2, 2023 •

edited

Loading