This is an automated email from the ASF dual-hosted git repository.
aloyszhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 5b16eb45f3 [INLONG-10495][Manager] Fix the problem of
ScanStartupSubStartOffset is set to null in pulsar extranode (#10496)
5b16eb45f3 is described below
commit 5b16eb45f33e5ed689c589e1b3ceac31756c203f
Author: fuweng11 <[email protected]>
AuthorDate: Tue Jun 25 11:02:38 2024 +0800
[INLONG-10495][Manager] Fix the problem of ScanStartupSubStartOffset is set
to null in pulsar extranode (#10496)
---
.../inlong/manager/pojo/sort/node/provider/PulsarProvider.java | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/node/provider/PulsarProvider.java
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/node/provider/PulsarProvider.java
index 9493f78f41..1ece6c0184 100644
---
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/node/provider/PulsarProvider.java
+++
b/inlong-manager/manager-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.ExtractNode;
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 class PulsarProvider implements ExtractNodeProvider {
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(),
