public void checkClientInBroker() throws MQClientException {

        for (Entry<String, MQConsumerInner> entry : this.consumerTable.entrySet()) {
            Set<SubscriptionData> subscriptionInner = entry.getValue().subscriptions();
            if (subscriptionInner == null || subscriptionInner.isEmpty()) {
                return;
            }

            for (SubscriptionData subscriptionData : subscriptionInner) {
                if (ExpressionType.isTagType(subscriptionData.getExpressionType())) {
                    continue;
                }
                // may need to check one broker every cluster...
                // assume that the configs of every broker in cluster are the same.
                String addr = findBrokerAddrByTopic(subscriptionData.getTopic());

                if (addr != null) {
                    try {
                        this.getMQClientAPIImpl().checkClientInBroker(
                            addr, entry.getKey(), this.clientId, subscriptionData, clientConfig.getMqClientApiTimeout()
                        );
                    } catch (Exception e) {
                        if (e instanceof MQClientException) {
                            throw (MQClientException) e;
                        } else {
                            throw new MQClientException("Check client in broker error, maybe because you use "
                                + subscriptionData.getExpressionType() + " to filter message, but server has not been upgraded to support!"
                                + "This error would not affect the launch of consumer, but may has impact on message receiving if you " +
                                "have use the new features which are not supported by server, please check the log!", e);
                        }
                    }
                }
            }
        }
    }
@Test
    public void testCheckClientInBroker() throws MQClientException, RemotingSendRequestException, RemotingConnectException, RemotingTimeoutException, InterruptedException {
        doThrow(new MQClientException("checkClientInBroker exception", null)).when(mQClientAPIImpl).checkClientInBroker(
                any(),
                any(),
                any(),
                any(SubscriptionData.class),
                anyLong());
        topicRouteTable.put(topic, createTopicRouteData());
        MQConsumerInner mqConsumerInner = createMQConsumerInner();
        mqConsumerInner.subscriptions().clear();
        SubscriptionData subscriptionData = new SubscriptionData();
        subscriptionData.setTopic(topic);
        subscriptionData.setExpressionType("type");
        mqConsumerInner.subscriptions().add(subscriptionData);
        consumerTable.put(group, mqConsumerInner);
        Throwable thrown = assertThrows(MQClientException.class, mqClientInstance::checkClientInBroker);
        assertTrue(thrown.getMessage().contains("checkClientInBroker exception"));
    }