private boolean checkAndCreateNewPulsarClient(InLongTopic topic) {
        if (!pulsarClients.containsKey(topic.getInLongCluster().getClusterId())) {
            if (topic.getInLongCluster().getBootstraps() != null) {
                try {
                    String token = topic.getInLongCluster().getToken();
                    Authentication auth = null;
                    if (StringUtils.isNoneBlank(token)) {
                        auth = AuthenticationFactory.token(token);
                    }
                    PulsarClient pulsarClient = PulsarClient.builder()
                            .serviceUrl(topic.getInLongCluster().getBootstraps())
                            .authentication(auth)
                            .build();
                    pulsarClients.put(topic.getInLongCluster().getClusterId(), pulsarClient);
                    LOGGER.info("create pulsar client succ cluster:{}, topic:{}, token:{}",
                            topic.getInLongCluster().getClusterId(),
                            topic.getTopic(),
                            topic.getInLongCluster().getToken());
                } catch (Exception e) {
                    LOGGER.error("create pulsar client error {}", topic);
                    LOGGER.error(e.getMessage(), e);
                    return false;
                }
            } else {
                LOGGER.error("bootstrap is null {}", topic.getInLongCluster());
                return false;
            }
        }
        LOGGER.info("create pulsar client true {}", topic);
        return true;
    }
online new inlong topic

@param newSubscribedInLongTopics List
@param reallyNewTopic List
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongSingleTopicManager.java