@Override
    public CompletableFuture<ConsumeMessageDirectlyResult> consumeMessageDirectly(String address,
        ConsumeMessageDirectlyResultRequestHeader requestHeader, long timeoutMillis) {
        CompletableFuture<ConsumeMessageDirectlyResult> future = new CompletableFuture<>();
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONSUME_MESSAGE_DIRECTLY, requestHeader);
        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
            if (response.getCode() == ResponseCode.SUCCESS) {
                ConsumeMessageDirectlyResult info = ConsumeMessageDirectlyResult.decode(response.getBody(), ConsumeMessageDirectlyResult.class);
                future.complete(info);
            } else {
                log.warn("consumeMessageDirectly getResponseCommand failed, {} {}", response.getCode(), response.getRemark());
                future.completeExceptionally(new MQClientException(response.getCode(), response.getRemark()));
            }
        });
        return future;
    }
@Test
    public void assertConsumeMessageDirectlyWithSuccess() throws Exception {
        ConsumeMessageDirectlyResult responseBody = new ConsumeMessageDirectlyResult();
        setResponseSuccess(RemotingSerializable.encode(responseBody));
        ConsumeMessageDirectlyResultRequestHeader requestHeader = mock(ConsumeMessageDirectlyResultRequestHeader.class);
        CompletableFuture<ConsumeMessageDirectlyResult> actual = mqClientAdminImpl.consumeMessageDirectly(defaultBrokerAddr, requestHeader, defaultTimeout);
        ConsumeMessageDirectlyResult result = actual.get();
        assertNotNull(result);
        assertTrue(result.isAutoCommit());
    }