private void createPulsarClient(CacheZoneCluster cluster) {
        LOGGER.info("start to init pulsar client for cluster={}", cluster);
        String clientKey = cluster.getBootstraps();
        if (clientKey == null) {
            LOGGER.error("bootstrap is null for cluster={}", cluster);
            return;
        }
        if (pulsarClients.containsKey(clientKey)) {
            LOGGER.info("Repeat to init pulsar client for cluster={}", cluster);
            return;
        }
        try {
            String token = cluster.getToken();
            Authentication auth = null;
            if (StringUtils.isNoneBlank(token)) {
                auth = AuthenticationFactory.token(token);
            }
            PulsarClient pulsarClient = PulsarClient.builder()
                    .serviceUrl(cluster.getBootstraps())
                    .authentication(auth)
                    .build();
            LOGGER.info("create pulsar client succ cluster:{}, token:{}",
                    cluster.getClusterId(),
                    cluster.getToken());
            PulsarClient oldClient = pulsarClients.putIfAbsent(cluster.getClusterId(), pulsarClient);
            if (oldClient != null && !oldClient.isClosed()) {
                LOGGER.warn("close new pulsar client for cluster={}", cluster);
                pulsarClient.close();
            }
        } catch (Exception e) {
            LOGGER.error("create pulsar client error for cluster={}", cluster, e);
            return;
        }
        LOGGER.info("success to init pulsar client for cluster={}", cluster);
    }
Inlong manager that maintain the single topic fetchers.
It is suitable to the cases that each topic has its own configurations.
And each consumer only consume the very one topic.
InlongMultiTopicManager was used since 1.9.0
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongTopicManager.java