[INLONG-4598][Manager] Fix the Pulsar topic not match error (#4599)

Author: woofyzhao

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/source/pulsar/PulsarSource.java

@@ -43,7 +43,7 @@
 public class PulsarSource extends StreamSource {
 
     @ApiModelProperty("Pulsar tenant")
-    private String tenant = "default";
+    private String tenant = "public";
 
     @ApiModelProperty("Pulsar namespace")
     private String namespace;

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/CreateSortConfigListenerV2.java

@@ -144,9 +144,12 @@ private Map<String, List<StreamSource>> createPulsarSources(
         PulsarClusterInfo pulsarCluster = (PulsarClusterInfo) clusterInfo;
         String adminUrl = pulsarCluster.getAdminUrl();
         String serviceUrl = pulsarCluster.getUrl();
+        String tenant = StringUtils.isEmpty(pulsarCluster.getTenant()) ? InlongGroupSettings.DEFAULT_PULSAR_TENANT
+                : pulsarCluster.getTenant();
         streamInfoList.forEach(streamInfo -> {
             PulsarSource pulsarSource = new PulsarSource();
             String streamId = streamInfo.getInlongStreamId();
+            pulsarSource.setTenant(tenant);
             pulsarSource.setSourceName(streamId);
             pulsarSource.setNamespace(groupInfo.getMqResource());
             pulsarSource.setTopic(streamInfo.getMqResource());

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/ExtractNodeUtils.java

@@ -231,7 +231,8 @@ public static PulsarExtractNode createExtractNode(PulsarSource pulsarSource) {
         List<FieldInfo> fieldInfos = streamFields.stream()
                 .map(streamFieldInfo -> FieldInfoUtils.parseStreamFieldInfo(streamFieldInfo, name))
                 .collect(Collectors.toList());
-        String topic = pulsarSource.getTopic();
+        String fullTopicName =
+                pulsarSource.getTenant() + "/" + pulsarSource.getNamespace() + "/" + pulsarSource.getTopic();
 
         Format format;
         DataTypeEnum dataType = DataTypeEnum.forName(pulsarSource.getSerializationType());
@@ -269,7 +270,7 @@ public static PulsarExtractNode createExtractNode(PulsarSource pulsarSource) {
                 fieldInfos,
                 null,
                 Maps.newHashMap(),
-                topic,
+                fullTopicName,
                 adminUrl,
                 serviceUrl,
                 format,