@Override
    public CompletableFuture<ResponseFuture> invokeImpl(final Channel channel, final RemotingCommand request,
        final long timeoutMillis) {
        Stopwatch stopwatch = Stopwatch.createStarted();
        String channelRemoteAddr = RemotingHelper.parseChannelRemoteAddr(channel);
        doBeforeRpcHooks(channelRemoteAddr, request);

        return super.invokeImpl(channel, request, timeoutMillis).thenCompose(responseFuture -> {
            RemotingCommand response = responseFuture.getResponseCommand();
            if (response.getCode() == ResponseCode.GO_AWAY) {
                if (nettyClientConfig.isEnableReconnectForGoAway()) {
                    ChannelWrapper channelWrapper = channelWrapperTables.computeIfPresent(channel, (channel0, channelWrapper0) -> {
                        try {
                            if (channelWrapper0.reconnect()) {
                                LOGGER.info("Receive go away from channel {}, recreate the channel", channel0);
                                channelWrapperTables.put(channelWrapper0.getChannel(), channelWrapper0);
                            }
                        } catch (Throwable t) {
                            LOGGER.error("Channel {} reconnect error", channelWrapper0, t);
                        }
                        return channelWrapper0;
                    });
                    if (channelWrapper != null) {
                        if (nettyClientConfig.isEnableTransparentRetry()) {
                            RemotingCommand retryRequest = RemotingCommand.createRequestCommand(request.getCode(), request.readCustomHeader());
                            retryRequest.setBody(request.getBody());
                            if (channelWrapper.isOK()) {
                                long duration = stopwatch.elapsed(TimeUnit.MILLISECONDS);
                                stopwatch.stop();
                                Channel retryChannel = channelWrapper.getChannel();
                                if (retryChannel != null && channel != retryChannel) {
                                    return super.invokeImpl(retryChannel, retryRequest, timeoutMillis - duration);
                                }
                            } else {
                                CompletableFuture<ResponseFuture> future = new CompletableFuture<>();
                                ChannelFuture channelFuture = channelWrapper.getChannelFuture();
                                channelFuture.addListener(f -> {
                                    long duration = stopwatch.elapsed(TimeUnit.MILLISECONDS);
                                    stopwatch.stop();
                                    if (f.isSuccess()) {
                                        Channel retryChannel0 = channelFuture.channel();
                                        if (retryChannel0 != null && channel != retryChannel0) {
                                            super.invokeImpl(retryChannel0, retryRequest, timeoutMillis - duration).whenComplete((v, t) -> {
                                                if (t != null) {
                                                    future.completeExceptionally(t);
                                                } else {
                                                    future.complete(v);
                                                }
                                            });
                                        }
                                    } else {
                                        future.completeExceptionally(new RemotingConnectException(channelWrapper.channelAddress));
                                    }
                                });
                                return future;
                            }
                        }
                    }
                }
            }
            return CompletableFuture.completedFuture(responseFuture);
        }).whenComplete((v, t) -> {
            if (t == null) {
                doAfterRpcHooks(channelRemoteAddr, request, v.getResponseCommand());
            }
        });
    }
@Test
    public void testInvokeImplFail() {
        remotingClient.registerRPCHook(rpcHookMock);
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, null);

        Channel channel = new LocalChannel();
        CompletableFuture<ResponseFuture> future = new CompletableFuture<>();
        future.completeExceptionally(new RemotingException(null));

        doReturn(future).when(remotingClient).invoke0(any(Channel.class), any(RemotingCommand.class), anyLong());

        assertThatThrownBy(() -> remotingClient.invokeImpl(channel, request, 1000).get()).getCause().isInstanceOf(RemotingException.class);

        verify(rpcHookMock).doBeforeRequest(anyString(), eq(request));
        verify(rpcHookMock, never()).doAfterResponse(anyString(), eq(request), any());
    }