@Override
        protected void doWork() {
            logger.debug("InLongSingleTopicManagerImpl doWork");
            if (stopAssign) {
                logger.warn("assign is stopped");
                return;
            }
            // get sortTask conf from manager
            if (queryConsumeConfig != null) {
                long start = System.currentTimeMillis();
                context.addRequestManager();
                ConsumeConfig consumeConfig = queryConsumeConfig
                        .queryCurrentConsumeConfig(context.getConfig().getSortTaskId());
                if (consumeConfig != null) {
                    List<InLongTopic> topicSubset = context.getConfig().getConsumerSubset(consumeConfig.getTopics());
                    handleUpdatedConsumeConfig(topicSubset);
                } else {
                    logger.warn("subscribedInfo is null");
                    context.addRequestManagerFail(System.currentTimeMillis() - start);
                }
            } else {
                logger.error("subscribedMetaDataInfo is null");
            }
        }
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