### Question:
DefaultBrokerRcvQltyStats implements BrokerRcvQltyStats { public void startBrokerStatistic() { if (this.statusId.compareAndSet(-1, 0)) { this.statisticThread.start(); } } DefaultBrokerRcvQltyStats(final RpcServiceFactory rpcServiceFactory,
                                     final TubeClientConfig producerConfig); void startBrokerStatistic(); boolean isStopped(); @Override List<Partition> getAllowedBrokerPartitions(
            Map<Integer, List<Partition>> brokerPartList); @Override void removeUnRegisteredBroker(List<Integer> registeredBrokerIdList); @Override void statisticDltBrokerStatus(); @Override void addSendStatistic(int brokerId); @Override void addReceiveStatistic(int brokerId, boolean isSuccess); @Override void stopBrokerStatistic(); @Override String toString();  }### Answer:
@Test public void testStartBrokerStatistic() throws Exception { RpcServiceFactory rpcServiceFactory = mock(RpcServiceFactory.class); when(rpcServiceFactory.getForbiddenAddrMap()).thenReturn(new ConcurrentHashMap<String, Long>()); when(rpcServiceFactory.getUnavilableBrokerMap()).thenReturn(new ConcurrentHashMap<Integer, Long>()); TubeClientConfig config = mock(TubeClientConfig.class); when(config.getSessionMaxAllowedDelayedMsgCount()).thenReturn(1000L); DefaultBrokerRcvQltyStats stats = new DefaultBrokerRcvQltyStats(rpcServiceFactory, config); stats.startBrokerStatistic(); assertFalse(stats.isStopped()); stats.addSendStatistic(0); stats.addReceiveStatistic(0, true); stats.statisticDltBrokerStatus(); Map<Integer, List<Partition>> brokerPartList = new HashMap<>(); List<Partition> partitions = new ArrayList<>(); partitions.add(new Partition(new BrokerInfo("0:192.168.0.1:18080"), "test_topic", 1)); brokerPartList.put(0, partitions); List<Partition> actualPartitions = stats.getAllowedBrokerPartitions(brokerPartList); assertEquals(1, actualPartitions.size()); assertEquals(1, actualPartitions.get(0).getPartitionId()); stats.removeUnRegisteredBroker(new ArrayList<Integer>()); stats.stopBrokerStatistic(); assertTrue(stats.isStopped()); }
@Test public void testStartBrokerStatistic() throws Exception { RpcServiceFactory rpcServiceFactory = mock(RpcServiceFactory.class); when(rpcServiceFactory.getForbiddenAddrMap()).thenReturn(new ConcurrentHashMap<String, Long>()); when(rpcServiceFactory.getUnavailableBrokerMap()).thenReturn(new ConcurrentHashMap<Integer, Long>()); TubeClientConfig config = mock(TubeClientConfig.class); when(config.getSessionMaxAllowedDelayedMsgCount()).thenReturn(1000L); DefaultBrokerRcvQltyStats stats = new DefaultBrokerRcvQltyStats(rpcServiceFactory, config); stats.startBrokerStatistic(); assertFalse(stats.isStopped()); stats.addSendStatistic(0); stats.addReceiveStatistic(0, true); stats.statisticDltBrokerStatus(); Map<Integer, List<Partition>> brokerPartList = new HashMap<>(); List<Partition> partitions = new ArrayList<>(); partitions.add(new Partition(new BrokerInfo("0:192.168.0.1:18080"), "test_topic", 1)); brokerPartList.put(0, partitions); List<Partition> actualPartitions = stats.getAllowedBrokerPartitions(brokerPartList); assertEquals(1, actualPartitions.size()); assertEquals(1, actualPartitions.get(0).getPartitionId()); stats.removeUnRegisteredBroker(new ArrayList<Integer>()); stats.stopBrokerStatistic(); assertTrue(stats.isStopped()); }