public PartitionSelectResult pullSelect() {
        PartitionSelectResult result = getCurrPartsStatus();
        if (!result.isSuccess()) {
            return result;
        }
        waitCont.incrementAndGet();
        try {
            rebProcessWait();
            if (this.isClosed.get()) {
                return new PartitionSelectResult(false,
                        TErrCodeConstants.BAD_REQUEST,
                        "Client instance has been shutdown!");
            }
            if (partitionMap.isEmpty()) {
                return new PartitionSelectResult(false,
                        TErrCodeConstants.NO_PARTITION_ASSIGNED,
                        "No partition info in local, please wait and try later");
            }
            String key = indexPartition.poll();
            if (key == null) {
                if (hasPartitionWait()) {
                    return new PartitionSelectResult(false,
                            TErrCodeConstants.ALL_PARTITION_WAITING,
                            "All partition in waiting, retry later!");
                } else if (!partitionUsedMap.isEmpty()) {
                    return new PartitionSelectResult(false,
                            TErrCodeConstants.ALL_PARTITION_INUSE,
                            "No idle partition to consume, please wait and try later");
                } else {
                    return new PartitionSelectResult(false,
                            TErrCodeConstants.ALL_PARTITION_FROZEN,
                            "All partition are frozen to consume, please unfreeze partition(s) or wait");
                }
            }
            PartitionExt partitionExt = partitionMap.get(key);
            if (partitionExt == null) {
                return new PartitionSelectResult(false,
                        TErrCodeConstants.BAD_REQUEST,
                        "No valid partition to consume, retry later 1");
            }
            long curTime = System.currentTimeMillis();
            Long newTime = partitionUsedMap.putIfAbsent(key, curTime);
            if (newTime != null) {
                return new PartitionSelectResult(false,
                        TErrCodeConstants.BAD_REQUEST,
                        "No valid partition to consume, retry later 2");
            }
            return new PartitionSelectResult(true, TErrCodeConstants.SUCCESS, "Ok!",
                    partitionExt, curTime, partitionExt.getAndResetLastPackConsumed());
        } catch (Throwable e1) {
            return new PartitionSelectResult(false,
                    TErrCodeConstants.BAD_REQUEST,
                    new StringBuilder(256)
                            .append("Wait partition to consume abnormal : ")
                            .append(e1.getMessage()).toString());
        } finally {
            waitCont.decrementAndGet();
        }
    }
Pull the selected partitions.

@return pull result
inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/RmtDataCache.java