public void filterCachedPartitionInfo(Map<BrokerInfo, List<Partition>> registerInfoMap,
            List<Partition> unRegPartitionList) {
        List<BrokerInfo> brokerInfoList = new ArrayList<>();
        for (Map.Entry<BrokerInfo, List<Partition>> entry : registerInfoMap.entrySet()) {
            if (entry.getKey() == null || entry.getValue() == null) {
                continue;
            }
            ConcurrentLinkedQueue<Partition> partitionList =
                    brokerPartitionConMap.get(entry.getKey());
            if (partitionList == null || partitionList.isEmpty()) {
                unRegPartitionList.addAll(entry.getValue());
            } else {
                boolean isNewBroker = true;
                for (Partition regPartition : entry.getValue()) {
                    if (!partitionList.contains(regPartition)) {
                        unRegPartitionList.add(regPartition);
                        isNewBroker = false;
                    }
                }
                if (isNewBroker) {
                    brokerInfoList.add(entry.getKey());
                }
            }
        }
        for (BrokerInfo brokerInfo : brokerInfoList) {
            registerInfoMap.remove(brokerInfo);
        }
    }
Filter cached partition information

@param registerInfoMap     the partitions to register
@param unRegPartitionList  the unregistered partition list
inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/RmtDataCache.java