public void subscribe(String topic, String subExpression) throws MQClientException {
        try {
            SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(topic, subExpression);
            this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
            if (this.mQClientFactory != null) {
                this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
            }
        } catch (Exception e) {
            throw new MQClientException("subscription exception", e);
        }
    }
@Test
    public void testSubscribe() throws MQClientException {
        defaultMQPushConsumerImpl.subscribe(defaultTopic, "fullClassname", "filterClassSource");
        RebalanceImpl actual = defaultMQPushConsumerImpl.getRebalanceImpl();
        assertEquals(1, actual.getSubscriptionInner().size());
    }