[INLONG-10495][Manager] Fix the problem of ScanStartupSubStartOffset …

…is set to null in pulsar extranode (#10496)
apache · Jun 25, 2024 · 5b16eb4 · 5b16eb4
1 parent 5d56e41
commit 5b16eb4
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/...-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/node/provider/PulsarProvider.java b/...-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/node/provider/PulsarProvider.java
@@ -30,6 +30,7 @@
 import org.apache.inlong.sort.protocol.node.extract.PulsarExtractNode;
 import org.apache.inlong.sort.protocol.node.format.Format;
 
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Service;
 
 import java.util.ArrayList;
@@ -67,7 +68,9 @@ public ExtractNode createExtractNode(StreamNode streamNodeInfo) {
         final String primaryKey = pulsarSource.getPrimaryKey();
         final String serviceUrl = pulsarSource.getServiceUrl();
         final String adminUrl = pulsarSource.getAdminUrl();
-        final String scanStartupSubStartOffset = null;
+        final String scanStartupSubStartOffset =
+                StringUtils.isNotBlank(pulsarSource.getSubscription()) ? PulsarScanStartupMode.EARLIEST.getValue()
+                        : null;
 
         return new PulsarExtractNode(pulsarSource.getSourceName(),
                 pulsarSource.getSourceName(),