@Override
    public CompletableFuture<SubscriptionData> querySubscriptionByConsumer(String address,
        QuerySubscriptionByConsumerRequestHeader requestHeader, long timeoutMillis) {
        CompletableFuture<SubscriptionData> future = new CompletableFuture<>();
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.QUERY_SUBSCRIPTION_BY_CONSUMER, requestHeader);
        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
            if (response.getCode() == ResponseCode.SUCCESS) {
                QuerySubscriptionResponseBody subscriptionResponseBody =
                    QuerySubscriptionResponseBody.decode(response.getBody(), QuerySubscriptionResponseBody.class);
                future.complete(subscriptionResponseBody.getSubscriptionData());
            } else {
                log.warn("querySubscriptionByConsumer getResponseCommand failed, {} {}", response.getCode(), response.getRemark());
                future.completeExceptionally(new MQClientException(response.getCode(), response.getRemark()));
            }
        });
        return future;
    }
@Test
    public void assertQuerySubscriptionByConsumerWithSuccess() throws Exception {
        SubscriptionData responseBody = new SubscriptionData();
        setResponseSuccess(RemotingSerializable.encode(responseBody));
        QuerySubscriptionByConsumerRequestHeader requestHeader = mock(QuerySubscriptionByConsumerRequestHeader.class);
        CompletableFuture<SubscriptionData> actual = mqClientAdminImpl.querySubscriptionByConsumer(defaultBrokerAddr, requestHeader, defaultTimeout);
        assertNull(actual.get());
    }