### Question:
DeFiConsumerManager extends ConsumerManager { @Override public void doChannelCloseEvent(final String remoteAddr, final Channel channel) { Set<String> subscribeTopics = null; Iterator<Map.Entry<String, ConsumerGroupInfo>> it = this.consumerTable.entrySet().iterator(); while (it.hasNext()) { Map.Entry<String, ConsumerGroupInfo> next = it.next(); ConsumerGroupInfo info = next.getValue(); if (info.getChannelInfoTable().get(channel) != null) { ClientChannelInfo clientChannelInfo = info.getChannelInfoTable().get(channel); DeFiConsumerGroupInfo deFiConsumerGroupInfo = (DeFiConsumerGroupInfo) info; subscribeTopics = deFiConsumerGroupInfo.findSubscribedTopicByClientId(clientChannelInfo.getClientId()); } } super.doChannelCloseEvent(remoteAddr, channel); if (subscribeTopics != null) { for (String topic : subscribeTopics) { adjustQueueNumStrategy.decreaseQueueNum(topic); } } } DeFiConsumerManager(final ConsumerIdsChangeListener consumerIdsChangeListener,
        final AdjustQueueNumStrategy strategy); @Override boolean registerConsumer(final String group, final ClientChannelInfo clientChannelInfo,
        ConsumeType consumeType, MessageModel messageModel, ConsumeFromWhere consumeFromWhere,
        final Set<SubscriptionData> subList, boolean isNotifyConsumerIdsChangedEnable); @Override void unregisterConsumer(final String group, final ClientChannelInfo clientChannelInfo,
        boolean isNotifyConsumerIdsChangedEnable); @Override void doChannelCloseEvent(final String remoteAddr, final Channel channel); @Override void scanNotActiveChannel(); void notifyWhenTopicConfigChange(String topic); ConcurrentHashMap<String, ConsumerGroupInfo> getConsumerTable();  }### Answer:
@Test public void testDoChannelCloseEvent() { assertThat(deFiConsumerManager.getConsumerTable().size()).isEqualTo(1); deFiConsumerManager.doChannelCloseEvent("127.0.0.1", mockChannel); assertThat(deFiConsumerManager.getConsumerTable().size()).isEqualTo(0); }