@Override
    public CompletableFuture<Void> updateOrCreateTopic(String address, CreateTopicRequestHeader requestHeader,
        long timeoutMillis) {
        CompletableFuture<Void> future = new CompletableFuture<>();
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_TOPIC, requestHeader);
        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
            if (response.getCode() == ResponseCode.SUCCESS) {
                future.complete(null);
            } else {
                log.warn("updateOrCreateTopic getResponseCommand failed, {} {}, header={}", response.getCode(), response.getRemark(), requestHeader);
                future.completeExceptionally(new MQClientException(response.getCode(), response.getRemark()));
            }
        });
        return future;
    }
@Test
    public void assertUpdateOrCreateTopicWithError() {
        setResponseError();
        CreateTopicRequestHeader requestHeader = mock(CreateTopicRequestHeader.class);
        CompletableFuture<Void> actual = mqClientAdminImpl.updateOrCreateTopic(defaultBrokerAddr, requestHeader, defaultTimeout);
        Throwable thrown = assertThrows(ExecutionException.class, actual::get);
        assertTrue(thrown.getCause() instanceof MQClientException);
        MQClientException mqException = (MQClientException) thrown.getCause();
        assertEquals(ResponseCode.SYSTEM_ERROR, mqException.getResponseCode());
        assertTrue(mqException.getMessage().contains("CODE: 1  DESC: null"));
    }