RockteMQ-AI commented on code in PR #151:
URL: https://github.com/apache/rocketmq-connect/pull/151#discussion_r3839557681
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/SinkConnectorConfig.java:
##########
@@ -19,25 +19,30 @@
package org.apache.rocketmq.connect.runtime.config;
import com.google.common.base.Splitter;
-import java.util.HashSet;
+import java.util.HashMap;
import java.util.List;
-import java.util.Set;
+import java.util.Map;
import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
public class SinkConnectorConfig extends ConnectConfig {
- public static Set<String> parseTopicList(ConnectKeyValue taskConfig) {
- String messageQueueStr =
taskConfig.getString(RuntimeConfigDefine.CONNECT_TOPICNAME);
- if (StringUtils.isBlank(messageQueueStr)) {
- return null;
+ public static Map<String, String> parseTopicList(ConnectKeyValue
taskConfig) {
Review Comment:
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.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/SinkConnectorConfig.java:
##########
@@ -19,25 +19,30 @@
package org.apache.rocketmq.connect.runtime.config;
import com.google.common.base.Splitter;
-import java.util.HashSet;
+import java.util.HashMap;
import java.util.List;
-import java.util.Set;
+import java.util.Map;
import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
public class SinkConnectorConfig extends ConnectConfig {
- public static Set<String> parseTopicList(ConnectKeyValue taskConfig) {
- String messageQueueStr =
taskConfig.getString(RuntimeConfigDefine.CONNECT_TOPICNAME);
- if (StringUtils.isBlank(messageQueueStr)) {
- return null;
+ public static Map<String, String> parseTopicList(ConnectKeyValue
taskConfig) {
+ 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) {
Review Comment:
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.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/SinkConnectorConfig.java:
##########
@@ -19,25 +19,30 @@
package org.apache.rocketmq.connect.runtime.config;
import com.google.common.base.Splitter;
-import java.util.HashSet;
+import java.util.HashMap;
import java.util.List;
-import java.util.Set;
+import java.util.Map;
import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
public class SinkConnectorConfig extends ConnectConfig {
Review Comment:
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.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSinkTask.java:
##########
@@ -372,7 +375,7 @@ private void pullMessageFromQueues() throws
InterruptedException {
final long beginPullMsgTimestamp = System.currentTimeMillis();
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);
Review Comment:
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 '*'.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/SinkConnectorConfig.java:
##########
@@ -19,25 +19,30 @@
package org.apache.rocketmq.connect.runtime.config;
import com.google.common.base.Splitter;
-import java.util.HashSet;
+import java.util.HashMap;
import java.util.List;
-import java.util.Set;
+import java.util.Map;
import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.connect.runtime.common.ConnectKeyValue;
public class SinkConnectorConfig extends ConnectConfig {
Review Comment:
The variable name 'topicNameAndTagss' (double 's') is a typo that propagates
through the method. Minor, but reduces readability.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSinkTask.java:
##########
@@ -273,7 +275,8 @@ private void setQueueOffset() {
}
private void registTopics() {
Review Comment:
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.
--
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]
