FileSourceTask added a judgment whether the topic is null

2.FileSourceTask added a judgment whether the topic is null Co-authored-by: wangkai <wangkai@zhongan.com>
Jashinck · Mar 30, 2023 · b144d40 · b144d40
1 parent 1b8881f
commit b144d40
Show file tree

Hide file tree

Showing 5 changed files with 19 additions and 16 deletions.
diff --git a/...etmq-replicator/src/test/java/org/apache/rocketmq/replicator/RmqSourceReplicatorTest.java b/...etmq-replicator/src/test/java/org/apache/rocketmq/replicator/RmqSourceReplicatorTest.java
@@ -17,25 +17,10 @@
 package org.apache.rocketmq.replicator;
 
 import java.lang.reflect.Field;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.common.protocol.body.TopicList;
-import org.apache.rocketmq.common.protocol.route.QueueData;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
-import org.apache.rocketmq.remoting.exception.RemotingException;
 import org.apache.rocketmq.replicator.config.RmqConnectorConfig;
-import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.mockito.Mock;
 
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.when;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import org.mockito.Mockito;

diff --git a/pom.xml b/pom.xml
@@ -66,6 +66,7 @@
         <!-- compiler settings properties -->
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
+        <commons-lang3.version>3.12.0</commons-lang3.version>
     </properties>
     <dependencyManagement>
         <dependencies>
@@ -197,6 +198,11 @@
                 <artifactId>maven-artifact</artifactId>
                 <version>${maven-artifact.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.apache.commons</groupId>
+                <artifactId>commons-lang3</artifactId>
+                <version>${commons-lang3.version}</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 

diff --git a/.../src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTask.java b/.../src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTask.java
@@ -169,6 +169,10 @@ public WorkerSourceTask(WorkerConfig workerConfig,
     @Nullable
     private static String overwriteTopicFromRecord(ConnectRecord record) {
         KeyValue extensions = record.getExtensions();
+        if (extensions == null) {
+            log.error("record extensions null , lack of topic config");
+            return null;
+        }
         String o = extensions.getString(TOPIC, null);
         if (null == o) {
             log.error("Partition map element topic is null , lack of topic config");

diff --git a/rocketmq-connect-sample/pom.xml b/rocketmq-connect-sample/pom.xml
@@ -72,6 +72,10 @@
             <groupId>ch.qos.logback</groupId>
             <artifactId>logback-core</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
     </dependencies>
 
 </project>
diff --git a/rocketmq-connect-sample/src/main/java/org/apache/rocketmq/connect/file/FileSourceTask.java b/rocketmq-connect-sample/src/main/java/org/apache/rocketmq/connect/file/FileSourceTask.java
@@ -38,6 +38,8 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -146,7 +148,9 @@ public class FileSourceTask extends SourceTask {
                             fields.add(field);
                             schema.setFields(fields);
                             ConnectRecord connectRecord = new ConnectRecord(offsetKey(fileConfig.getFilename()), offsetValue(streamOffset), System.currentTimeMillis(), schema, line);
-                            connectRecord.addExtension("topic", fileConfig.getTopic());
+                            if (StringUtils.isNoneBlank(fileConfig.getTopic())) {
+                                connectRecord.addExtension("topic", fileConfig.getTopic());
+                            }
                             records.add(connectRecord);
                             if (records.size() >= batchSize) {
                                 return records;