[ISSUE #150] sink support topic tag by odbozhou · Pull Request #151 · apache/rocketmq-connect

odbozhou · 2022-05-27T09:44:33Z

What is the purpose of the change

close #150

Brief changelog

XX

Verifying this change

Follow this checklist to help us incorporate your contribution quickly and easily. Notice, it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR.

Make sure there is a Github issue filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue.
Format the pull request title like [ISSUE #123] Fix UnknownException when host config not exist. Each commit in the pull request should have a meaningful subject line and body.
Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when cross module dependency exist. If the new feature or significant change is committed, please remember to add integration-test in test module.
Run mvn -B clean apache-rat:check findbugs:findbugs checkstyle:checkstyle to make sure basic checks pass. Run mvn clean install -DskipITs to make sure unit-test pass. Run mvn clean test-compile failsafe:integration-test to make sure integration-test pass.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

lizhiboo · 2022-06-17T06:17:29Z

+        List<String> topicTagList = Splitter.on(COMMA).omitEmptyStrings().trimResults().splitToList(topicNameAndTagss);
+        Map<String, String> topicNameAndTagssMap = new HashMap<>(8);
+        for (String topicTagPair : topicTagList) {
+            List<String> topicAndTag = Splitter.on(SEMICOLON).omitEmptyStrings().trimResults().splitToList(topicTagPair);


IMO, exchange COMMA and SEMICOLON to maintain upgrade compatible.

When there is no tag, if you configure multiple topics, it should be topic1, topic2, topic3. If there is a tag, it should be topic1;tag1, topic2;tag2, topic3;tag3, if so, is there no compatibility problem?

ShannonDing · 2022-08-15T12:17:03Z

files conflicted. @odbozhou

ShannonDing

files conflicted.

RockteMQ-AI · 2026-07-30T10:46:08Z

⚠️ 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 sink_support_tag
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 2 file(s) with 76 lines of diff. No test changes detected — consider adding test coverage.

Automated review by github-manager-bot

RockteMQ-AI · 2026-08-05T15:12:38Z

@@ -19,25 +19,30 @@
 package org.apache.rocketmq.connect.runtime.config;


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

RockteMQ-AI

Summary

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

Diff size: 76 lines
Author: odbozhou (CONTRIBUTOR)

Automated review by RockteMQ-AI

RockteMQ-AI

Summary

Review of PR #151: [ISSUE #150] sink support topic tag

Findings: 6 issue(s) identified (2 critical).
CLA: unknown

Please address the inline comments above.

Automated review by github-manager-bot

RockteMQ-AI · 2026-08-23T20:46:52Z

-        String messageQueueStr = taskConfig.getString(RuntimeConfigDefine.CONNECT_TOPICNAME);
-        if (StringUtils.isBlank(messageQueueStr)) {
-            return null;
+    public static Map<String, String> parseTopicList(ConnectKeyValue taskConfig) {


No null/blank check on topicNameAndTagss before passing to Splitter. If the config key is absent or blank, Splitter.on(COMMA).splitToList(null) will throw a NullPointerException. The old code guarded with StringUtils.isBlank and returned null; that guard was removed without replacement.

RockteMQ-AI · 2026-08-23T20:46:52Z

+        String topicNameAndTagss = taskConfig.getString(RuntimeConfigDefine.CONNECT_TOPICNAME);
+        List<String> topicTagList = Splitter.on(COMMA).omitEmptyStrings().trimResults().splitToList(topicNameAndTagss);
+        Map<String, String> topicNameAndTagssMap = new HashMap<>(8);
+        for (String topicTagPair : topicTagList) {


When a topic-tag pair contains more than two SEMICOLON-separated segments (e.g. 'topicA;tag1;tag2'), the extra segments are silently discarded and only the first tag is used. There is no validation or error reported, which could hide misconfigured connector configs.

RockteMQ-AI · 2026-08-23T20:46:52Z

 import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;

 public class SinkConnectorConfig extends ConnectConfig {



The separator semantics have been swapped from the previous version: topics were previously split by SEMICOLON, now they are split by COMMA, and SEMICOLON now separates topic from tag. This is a backward-incompatible configuration format change. Any existing deployment using SEMICOLON-delimited topic lists will silently treat the entire string as one topic name, breaking existing connectors without any warning or migration path.

RockteMQ-AI · 2026-08-23T20:46:52Z

            try {
                shouldStopPullMsg();
-                pullResult = consumer.pullBlockIfNotFound(entry.getKey(), "*", entry.getValue(), MAX_MESSAGE_NUM);
+                pullResult = consumer.pullBlockIfNotFound(entry.getKey(), topicTagMap.get(entry.getKey().getTopic()), entry.getValue(), MAX_MESSAGE_NUM);


topicTagMap.get(entry.getKey().getTopic()) can return null if the MessageQueue's topic is not present in the map (e.g. dynamically assigned queues or rebalance edge cases). Passing null as the subscription expression to pullBlockIfNotFound will likely cause an NPE or an unfiltered pull depending on the client implementation, and is not equivalent to '*'.

RockteMQ-AI · 2026-08-23T20:46:52Z

@@ -273,7 +275,8 @@ private void setQueueOffset() {
    }

    private void registTopics() {


topicTagMap is assigned in registTopics() but read in pullMessageFromQueues(). If pullMessageFromQueues() is ever called before registTopics() (e.g. due to a future refactor or concurrent access), it will throw a NullPointerException. The field should be initialized to an empty map or the access should be guarded.

RockteMQ-AI · 2026-08-23T20:46:52Z

 import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;

 public class SinkConnectorConfig extends ConnectConfig {



The variable name 'topicNameAndTagss' (double 's') is a typo that propagates through the method. Minor, but reduces readability.

sink support tag

946b9f9

lizhiboo reviewed Jun 17, 2022

View reviewed changes

ShannonDing added the enhancement New feature or request label Aug 15, 2022

ShannonDing requested changes Aug 15, 2022

View reviewed changes

RockteMQ-AI reviewed Aug 5, 2026

View reviewed changes

RockteMQ-AI reviewed Aug 20, 2026

View reviewed changes

RockteMQ-AI suggested changes Aug 23, 2026

View reviewed changes

		@@ -19,25 +19,30 @@
		package org.apache.rocketmq.connect.runtime.config;

		import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;

		public class SinkConnectorConfig extends ConnectConfig {

		@@ -273,7 +275,8 @@ private void setQueueOffset() {
		}

		private void registTopics() {

Uh oh!

Conversation

odbozhou commented May 27, 2022 • edited by ShannonDing Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief changelog

Verifying this change

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ShannonDing commented Aug 15, 2022

Uh oh!

ShannonDing left a comment

Choose a reason for hiding this comment

Uh oh!

RockteMQ-AI commented Jul 30, 2026

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!

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

odbozhou commented May 27, 2022 •

edited by ShannonDing

Loading