public void scanProducerHeartBeat() {
        Set<String> groupSet = groupClusterData.keySet();

        Map<String /* cluster */, List<HeartbeatData>> clusterHeartbeatData = new HashMap<>();
        for (String group : groupSet) {
            groupClusterData.computeIfPresent(group, (groupName, clusterDataSet) -> {
                if (clusterDataSet.isEmpty()) {
                    return null;
                }
                if (!this.producerManager.groupOnline(groupName)) {
                    return null;
                }

                ProducerData producerData = new ProducerData();
                producerData.setGroupName(groupName);

                for (ClusterData clusterData : clusterDataSet) {
                    List<HeartbeatData> heartbeatDataList = clusterHeartbeatData.get(clusterData.cluster);
                    if (heartbeatDataList == null) {
                        heartbeatDataList = new ArrayList<>();
                    }

                    HeartbeatData heartbeatData;
                    if (heartbeatDataList.isEmpty()) {
                        heartbeatData = new HeartbeatData();
                        heartbeatData.setClientID(TRANS_HEARTBEAT_CLIENT_ID);
                        heartbeatDataList.add(heartbeatData);
                    } else {
                        heartbeatData = heartbeatDataList.get(heartbeatDataList.size() - 1);
                        if (heartbeatData.getProducerDataSet().size() >= ConfigurationManager.getProxyConfig().getTransactionHeartbeatBatchNum()) {
                            heartbeatData = new HeartbeatData();
                            heartbeatData.setClientID(TRANS_HEARTBEAT_CLIENT_ID);
                            heartbeatDataList.add(heartbeatData);
                        }
                    }

                    heartbeatData.getProducerDataSet().add(producerData);
                    clusterHeartbeatData.put(clusterData.cluster, heartbeatDataList);
                }

                if (clusterDataSet.isEmpty()) {
                    return null;
                }
                return clusterDataSet;
            });
        }

        if (clusterHeartbeatData.isEmpty()) {
            return;
        }
        Map<String, String> brokerAddrNameMap = new ConcurrentHashMap<>();
        Set<Map.Entry<String, List<HeartbeatData>>> clusterEntry = clusterHeartbeatData.entrySet();
        for (Map.Entry<String, List<HeartbeatData>> entry : clusterEntry) {
            sendHeartBeatToCluster(entry.getKey(), entry.getValue(), brokerAddrNameMap);
        }
        this.brokerAddrNameMapRef.set(brokerAddrNameMap);
    }
@Test
    public void testScanProducerHeartBeat() throws Exception {
        when(this.producerManager.groupOnline(anyString())).thenReturn(true);

        Mockito.reset(this.topicRouteService);
        String brokerName2 = "broker-2-01";
        String clusterName2 = "broker-2";
        String brokerAddr2 = "127.0.0.2:10911";

        BrokerData brokerData = new BrokerData();
        QueueData queueData = new QueueData();
        queueData.setBrokerName(brokerName2);
        brokerData.setCluster(clusterName2);
        brokerData.setBrokerName(brokerName2);
        HashMap<Long, String> brokerAddrs = new HashMap<>();
        brokerAddrs.put(MixAll.MASTER_ID, brokerName2);
        brokerData.setBrokerAddrs(brokerAddrs);
        topicRouteData.getQueueDatas().add(queueData);
        topicRouteData.getBrokerDatas().add(brokerData);
        when(this.topicRouteService.getAllMessageQueueView(any(), eq(TOPIC))).thenReturn(new MessageQueueView(TOPIC, topicRouteData, null));

        TopicRouteData clusterTopicRouteData = new TopicRouteData();
        QueueData clusterQueueData = new QueueData();
        BrokerData clusterBrokerData = new BrokerData();

        clusterQueueData.setBrokerName(BROKER_NAME);
        clusterTopicRouteData.setQueueDatas(Lists.newArrayList(clusterQueueData));
        clusterBrokerData.setCluster(CLUSTER_NAME);
        clusterBrokerData.setBrokerName(BROKER_NAME);
        brokerAddrs = new HashMap<>();
        brokerAddrs.put(MixAll.MASTER_ID, BROKER_ADDR);
        clusterBrokerData.setBrokerAddrs(brokerAddrs);
        clusterTopicRouteData.setBrokerDatas(Lists.newArrayList(clusterBrokerData));
        when(this.topicRouteService.getAllMessageQueueView(any(), eq(CLUSTER_NAME))).thenReturn(new MessageQueueView(CLUSTER_NAME, clusterTopicRouteData, null));

        TopicRouteData clusterTopicRouteData2 = new TopicRouteData();
        QueueData clusterQueueData2 = new QueueData();
        BrokerData clusterBrokerData2 = new BrokerData();

        clusterQueueData2.setBrokerName(brokerName2);
        clusterTopicRouteData2.setQueueDatas(Lists.newArrayList(clusterQueueData2));
        clusterBrokerData2.setCluster(clusterName2);
        clusterBrokerData2.setBrokerName(brokerName2);
        brokerAddrs = new HashMap<>();
        brokerAddrs.put(MixAll.MASTER_ID, brokerAddr2);
        clusterBrokerData2.setBrokerAddrs(brokerAddrs);
        clusterTopicRouteData2.setBrokerDatas(Lists.newArrayList(clusterBrokerData2));
        when(this.topicRouteService.getAllMessageQueueView(any(), eq(clusterName2))).thenReturn(new MessageQueueView(clusterName2, clusterTopicRouteData2, null));

        ConfigurationManager.getProxyConfig().setTransactionHeartbeatBatchNum(2);
        this.clusterTransactionService.start();
        Set<String> groupSet = new HashSet<>();

        for (int i = 0; i < 3; i++) {
            groupSet.add(GROUP + i);
            this.clusterTransactionService.addTransactionSubscription(ctx, GROUP + i, TOPIC);
        }

        ArgumentCaptor<String> brokerAddrArgumentCaptor = ArgumentCaptor.forClass(String.class);
        ArgumentCaptor<HeartbeatData> heartbeatDataArgumentCaptor = ArgumentCaptor.forClass(HeartbeatData.class);
        when(mqClientAPIExt.sendHeartbeatOneway(
            brokerAddrArgumentCaptor.capture(),
            heartbeatDataArgumentCaptor.capture(),
            anyLong()
        )).thenReturn(CompletableFuture.completedFuture(null));

        this.clusterTransactionService.scanProducerHeartBeat();

        await().atMost(Duration.ofSeconds(1)).until(() -> brokerAddrArgumentCaptor.getAllValues().size() == 4);

        assertEquals(Lists.newArrayList(BROKER_ADDR, BROKER_ADDR, brokerAddr2, brokerAddr2),
            brokerAddrArgumentCaptor.getAllValues().stream().sorted().collect(Collectors.toList()));

        List<HeartbeatData> heartbeatDataList = heartbeatDataArgumentCaptor.getAllValues();

        for (final HeartbeatData heartbeatData : heartbeatDataList) {
            for (ProducerData producerData : heartbeatData.getProducerDataSet()) {
                groupSet.remove(producerData.getGroupName());
            }
        }

        assertTrue(groupSet.isEmpty());
        assertEquals(brokerName2, this.clusterTransactionService.getBrokerNameByAddr(brokerAddr2));
        assertEquals(BROKER_NAME, this.clusterTransactionService.getBrokerNameByAddr(BROKER_ADDR));
    }