@Override
    public boolean clean() {
        String sortTaskId = context.getConfig().getSortTaskId();
        try {
            LOGGER.info("start to clean topic manager, sortTaskId={}", sortTaskId);
            stopAssign = true;
            closeAllFetchers();
            // closeAllPulsarClients();
            closeAllTubeFactories();
            LOGGER.info("success to clean topic manager, sortTaskId={}", sortTaskId);
            return true;
        } catch (Exception e) {
            LOGGER.error("failed to clean topic manager, sortTaskId={}", sortTaskId, e);
        } finally {
            fetchers.clear();
            pulsarClients.clear();
            tubeFactories.clear();
            stopAssign = false;
        }
        return false;
    }
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