@Override
    public CompletableFuture<Map<MessageQueue, Long>> invokeBrokerToResetOffset(String address,
        ResetOffsetRequestHeader requestHeader, long timeoutMillis) {
        CompletableFuture<Map<MessageQueue, Long>> future = new CompletableFuture<>();
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.INVOKE_BROKER_TO_RESET_OFFSET, requestHeader);
        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
            if (response.getCode() == ResponseCode.SUCCESS && null != response.getBody()) {
                Map<MessageQueue, Long> offsetTable = ResetOffsetBody.decode(response.getBody(), ResetOffsetBody.class).getOffsetTable();
                future.complete(offsetTable);
                log.info("Invoke broker to reset offset success. address:{}, header:{}, offsetTable:{}",
                    address, requestHeader, offsetTable);
            } else {
                log.warn("invokeBrokerToResetOffset getResponseCommand failed, {} {}, header={}", response.getCode(), response.getRemark(), requestHeader);
                future.completeExceptionally(new MQClientException(response.getCode(), response.getRemark()));
            }
        });
        return future;
    }
@Test
    public void assertInvokeBrokerToResetOffsetWithSuccess() throws Exception {
        ResetOffsetBody responseBody = new ResetOffsetBody();
        setResponseSuccess(RemotingSerializable.encode(responseBody));
        ResetOffsetRequestHeader requestHeader = mock(ResetOffsetRequestHeader.class);
        CompletableFuture<Map<MessageQueue, Long>> actual = mqClientAdminImpl.invokeBrokerToResetOffset(defaultBrokerAddr, requestHeader, defaultTimeout);
        assertEquals(0, actual.get().size());
    }