@Override
    public ExtractNode createExtractNode(StreamNode streamNodeInfo) {
        PulsarSource pulsarSource = (PulsarSource) streamNodeInfo;
        List<FieldInfo> fieldInfos = parseStreamFieldInfos(pulsarSource.getFieldList(), pulsarSource.getSourceName());
        Map<String, String> properties = parseProperties(pulsarSource.getProperties());

        String fullTopicName =
                pulsarSource.getPulsarTenant() + "/" + pulsarSource.getNamespace() + "/" + pulsarSource.getTopic();
        Format format = parsingFormat(pulsarSource.getSerializationType(),
                pulsarSource.getWrapType(),
                pulsarSource.getDataSeparator(),
                pulsarSource.getKvSeparator(),
                pulsarSource.getDataEscapeChar(),
                pulsarSource.getIgnoreParseError());

        String startupMode =
                StringUtils.isNotBlank(pulsarSource.getScanStartupMode()) ? PulsarScanStartupMode.forName(
                        pulsarSource.getScanStartupMode()).getValue() : null;
        final String primaryKey = pulsarSource.getPrimaryKey();
        final String serviceUrl = pulsarSource.getServiceUrl();
        final String adminUrl = pulsarSource.getAdminUrl();
        final String scanStartupSubStartOffset =
                StringUtils.isNotBlank(pulsarSource.getSubscription()) ? PulsarScanStartupMode.EARLIEST.getValue()
                        : null;

        return new PulsarExtractNode(pulsarSource.getSourceName(),
                pulsarSource.getSourceName(),
                fieldInfos,
                null,
                properties,
                fullTopicName,
                adminUrl,
                serviceUrl,
                format,
                startupMode,
                primaryKey,
                pulsarSource.getSubscription(),
                scanStartupSubStartOffset,
                pulsarSource.getClientAuthPluginClassName(),
                pulsarSource.getClientAuthParams());
    }
The Provider for creating Pulsar extract nodes.
inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/node/provider/PulsarProvider.java