public void resumeTimeoutConsumePartitions(boolean isPullConsume, long allowedPeriodTimes) {
        if (isPullConsume) {
            // For pull consume, do timeout check on partitions pulled without confirm
            if (!partitionUsedMap.isEmpty()) {
                List<String> partKeys = new ArrayList<>(partitionUsedMap.keySet());
                for (String keyId : partKeys) {
                    Long oldTime = partitionUsedMap.get(keyId);
                    if (oldTime != null && System.currentTimeMillis() - oldTime > allowedPeriodTimes) {
                        oldTime = partitionUsedMap.remove(keyId);
                        if (oldTime != null) {
                            PartitionExt partitionExt = partitionMap.get(keyId);
                            if (partitionExt != null) {
                                partitionExt.setLastPackConsumed(false);
                                releaseIdlePartition(keyId);
                            }
                        }
                    }
                }
            }
        }
        // add timeout expired check
        if (!timeouts.isEmpty()) {
            Timeout timeout1 = null;
            List<String> partKeys = new ArrayList<>(timeouts.keySet());
            for (String keyId : partKeys) {
                timeout1 = timeouts.get(keyId);
                if (timeout1 != null && timeout1.isExpired()) {
                    timeout1 = timeouts.remove(keyId);
                    if (timeout1 != null) {
                        releaseIdlePartition(keyId);
                    }
                }
            }
        }
    }
Resume consume timeout partitions

@param isPullConsume       whether is pull consume
@param allowedPeriodTimes  allowed hold duration
inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/RmtDataCache.java