public Map<BrokerInfo, List<PartitionSelectResult>> removeAndGetPartition(
            Map<BrokerInfo, List<Partition>> unRegisterInfoMap,
            List<String> partitionKeys, long inUseWaitPeriodMs,
            boolean isWaitTimeoutRollBack) {
        StringBuilder sBuilder = new StringBuilder(512);
        HashMap<BrokerInfo, List<PartitionSelectResult>> unNewRegisterInfoMap =
                new HashMap<>();
        pauseProcess();
        try {
            waitPartitions(partitionKeys, inUseWaitPeriodMs);
            boolean lastPackConsumed = false;
            for (Map.Entry<BrokerInfo, List<Partition>> entry : unRegisterInfoMap.entrySet()) {
                for (Partition partition : entry.getValue()) {
                    PartitionExt partitionExt =
                            rmvPartitionFromMap(partition.getPartitionKey());
                    if (partitionExt != null) {
                        lastPackConsumed = partitionExt.isLastPackConsumed();
                        if (!cancelTimeTask(partition.getPartitionKey())
                                && !indexPartition.remove(partition.getPartitionKey())) {
                            logger.info(sBuilder.append("[Process Interrupt] Partition : ")
                                    .append(partition.toString())
                                    .append(", data in processing, canceled").toString());
                            sBuilder.delete(0, sBuilder.length());
                            if (lastPackConsumed) {
                                if (isWaitTimeoutRollBack) {
                                    lastPackConsumed = false;
                                }
                            }
                        }
                        ConcurrentLinkedQueue<Partition> oldPartitionList =
                                topicPartitionConMap.get(partition.getTopic());
                        if (oldPartitionList != null) {
                            oldPartitionList.remove(partition);
                            if (oldPartitionList.isEmpty()) {
                                topicPartitionConMap.remove(partition.getTopic());
                            }
                        }
                        ConcurrentLinkedQueue<Partition> regMapPartitionList =
                                brokerPartitionConMap.get(entry.getKey());
                        if (regMapPartitionList != null) {
                            regMapPartitionList.remove(partition);
                            if (regMapPartitionList.isEmpty()) {
                                brokerPartitionConMap.remove(entry.getKey());
                            }
                        }
                        partitionOffsetMap.remove(partition.getPartitionKey());
                        partitionUsedMap.remove(partition.getPartitionKey());
                        PartitionSelectResult partitionRet =
                                new PartitionSelectResult(true, TErrCodeConstants.SUCCESS,
                                        "Ok!", partition, 0, lastPackConsumed);
                        List<PartitionSelectResult> targetPartitionList =
                                unNewRegisterInfoMap.computeIfAbsent(
                                        entry.getKey(), k -> new ArrayList<>());
                        targetPartitionList.add(partitionRet);
                    }
                }
            }
        } finally {
            resumeProcess();
        }
        return unNewRegisterInfoMap;
    }
Remove and get required partitions
@param unRegisterInfoMap  the unregistered partitions
@param partitionKeys      the partition keys
@param inUseWaitPeriodMs  the wait period in ms
@param isWaitTimeoutRollBack  whether wait timout

@return the removed partitions
inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/RmtDataCache.java