diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/admin/MqClientAdminImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/admin/MqClientAdminImpl.java
index 34f066c7ddd..d1b4186244f 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/admin/MqClientAdminImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/admin/MqClientAdminImpl.java
@@ -22,6 +22,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.client.MqClientAdmin;
@@ -84,7 +85,7 @@ public CompletableFuture<List<MessageExt>> queryMessage(String address, boolean
         CompletableFuture<List<MessageExt>> future = new CompletableFuture<>();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.QUERY_MESSAGE, requestHeader);
         request.addExtField(MixAll.UNIQUE_MSG_QUERY_FLAG, String.valueOf(uniqueKeyFlag));
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 List<MessageExt> wrappers = MessageDecoder.decodesBatch(ByteBuffer.wrap(response.getBody()), true, decompressBody, true);
                 future.complete(filterMessages(wrappers, requestHeader.getTopic(), requestHeader.getKey(), uniqueKeyFlag));
@@ -105,7 +106,7 @@ public CompletableFuture<TopicStatsTable> getTopicStatsInfo(String address,
         GetTopicStatsInfoRequestHeader requestHeader, long timeoutMillis) {
         CompletableFuture<TopicStatsTable> future = new CompletableFuture<>();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_TOPIC_STATS_INFO, requestHeader);
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 TopicStatsTable topicStatsTable = TopicStatsTable.decode(response.getBody(), TopicStatsTable.class);
                 future.complete(topicStatsTable);
@@ -122,7 +123,7 @@ public CompletableFuture<List<QueueTimeSpan>> queryConsumeTimeSpan(String addres
         QueryConsumeTimeSpanRequestHeader requestHeader, long timeoutMillis) {
         CompletableFuture<List<QueueTimeSpan>> future = new CompletableFuture<>();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.QUERY_CONSUME_TIME_SPAN, requestHeader);
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 QueryConsumeTimeSpanBody consumeTimeSpanBody = GroupList.decode(response.getBody(), QueryConsumeTimeSpanBody.class);
                 future.complete(consumeTimeSpanBody.getConsumeTimeSpanSet());
@@ -139,7 +140,7 @@ public CompletableFuture<Void> updateOrCreateTopic(String address, CreateTopicRe
         long timeoutMillis) {
         CompletableFuture<Void> future = new CompletableFuture<>();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_TOPIC, requestHeader);
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 future.complete(null);
             } else {
@@ -157,7 +158,7 @@ public CompletableFuture<Void> updateOrCreateSubscriptionGroup(String address, S
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_SUBSCRIPTIONGROUP, null);
         byte[] body = RemotingSerializable.encode(config);
         request.setBody(body);
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 future.complete(null);
             } else {
@@ -173,7 +174,7 @@ public CompletableFuture<Void> deleteTopicInBroker(String address, DeleteTopicRe
         long timeoutMillis) {
         CompletableFuture<Void> future = new CompletableFuture<>();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.DELETE_TOPIC_IN_BROKER, requestHeader);
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 future.complete(null);
             } else {
@@ -189,7 +190,7 @@ public CompletableFuture<Void> deleteTopicInNameserver(String address, DeleteTop
         long timeoutMillis) {
         CompletableFuture<Void> future = new CompletableFuture<>();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.DELETE_TOPIC_IN_NAMESRV, requestHeader);
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 future.complete(null);
             } else {
@@ -205,7 +206,7 @@ public CompletableFuture<Void> deleteKvConfig(String address, DeleteKVConfigRequ
         long timeoutMillis) {
         CompletableFuture<Void> future = new CompletableFuture<>();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.DELETE_KV_CONFIG, requestHeader);
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 future.complete(null);
             } else {
@@ -221,7 +222,7 @@ public CompletableFuture<Void> deleteSubscriptionGroup(String address, DeleteSub
         long timeoutMillis) {
         CompletableFuture<Void> future = new CompletableFuture<>();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.DELETE_SUBSCRIPTIONGROUP, requestHeader);
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 future.complete(null);
             } else {
@@ -237,7 +238,7 @@ public CompletableFuture<Map<MessageQueue, Long>> invokeBrokerToResetOffset(Stri
         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 -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS && null != response.getBody()) {
                 Map<MessageQueue, Long> offsetTable = ResetOffsetBody.decode(response.getBody(), ResetOffsetBody.class).getOffsetTable();
                 future.complete(offsetTable);
@@ -256,7 +257,7 @@ public CompletableFuture<MessageExt> viewMessage(String address, ViewMessageRequ
         long timeoutMillis) {
         CompletableFuture<MessageExt> future = new CompletableFuture<>();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.VIEW_MESSAGE_BY_ID, requestHeader);
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 ByteBuffer byteBuffer = ByteBuffer.wrap(response.getBody());
                 MessageExt messageExt = MessageDecoder.clientDecode(byteBuffer, true);
@@ -273,7 +274,7 @@ public CompletableFuture<MessageExt> viewMessage(String address, ViewMessageRequ
     public CompletableFuture<ClusterInfo> getBrokerClusterInfo(String address, long timeoutMillis) {
         CompletableFuture<ClusterInfo> future = new CompletableFuture<>();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_CLUSTER_INFO, null);
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 ClusterInfo clusterInfo = ClusterInfo.decode(response.getBody(), ClusterInfo.class);
                 future.complete(clusterInfo);
@@ -290,7 +291,7 @@ public CompletableFuture<ConsumerConnection> getConsumerConnectionList(String ad
         GetConsumerConnectionListRequestHeader requestHeader, long timeoutMillis) {
         CompletableFuture<ConsumerConnection> future = new CompletableFuture<>();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUMER_CONNECTION_LIST, requestHeader);
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 ConsumerConnection consumerConnection = ConsumerConnection.decode(response.getBody(), ConsumerConnection.class);
                 future.complete(consumerConnection);
@@ -307,7 +308,7 @@ public CompletableFuture<TopicList> queryTopicsByConsumer(String address,
         QueryTopicsByConsumerRequestHeader requestHeader, long timeoutMillis) {
         CompletableFuture<TopicList> future = new CompletableFuture<>();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.QUERY_TOPICS_BY_CONSUMER, requestHeader);
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 TopicList topicList = TopicList.decode(response.getBody(), TopicList.class);
                 future.complete(topicList);
@@ -324,7 +325,7 @@ public CompletableFuture<SubscriptionData> querySubscriptionByConsumer(String ad
         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 -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 QuerySubscriptionResponseBody subscriptionResponseBody =
                     QuerySubscriptionResponseBody.decode(response.getBody(), QuerySubscriptionResponseBody.class);
@@ -342,7 +343,7 @@ public CompletableFuture<ConsumeStats> getConsumeStats(String address, GetConsum
         long timeoutMillis) {
         CompletableFuture<ConsumeStats> future = new CompletableFuture<>();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUME_STATS, requestHeader);
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 ConsumeStats consumeStats = ConsumeStats.decode(response.getBody(), ConsumeStats.class);
                 future.complete(consumeStats);
@@ -359,7 +360,7 @@ public CompletableFuture<GroupList> queryTopicConsumeByWho(String address,
         QueryTopicConsumeByWhoRequestHeader requestHeader, long timeoutMillis) {
         CompletableFuture<GroupList> future = new CompletableFuture<>();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.QUERY_TOPIC_CONSUME_BY_WHO, requestHeader);
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 GroupList groupList = GroupList.decode(response.getBody(), GroupList.class);
                 future.complete(groupList);
@@ -376,7 +377,7 @@ public CompletableFuture<ConsumerRunningInfo> getConsumerRunningInfo(String addr
         GetConsumerRunningInfoRequestHeader requestHeader, long timeoutMillis) {
         CompletableFuture<ConsumerRunningInfo> future = new CompletableFuture<>();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUMER_RUNNING_INFO, requestHeader);
-        remotingClient.invoke(address, request, timeoutMillis).thenAccept(response -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 ConsumerRunningInfo info = ConsumerRunningInfo.decode(response.getBody(), ConsumerRunningInfo.class);
                 future.complete(info);
@@ -393,7 +394,7 @@ public CompletableFuture<ConsumeMessageDirectlyResult> consumeMessageDirectly(St
         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 -> {
+        handleResponse(remotingClient.invoke(address, request, timeoutMillis), future, response -> {
             if (response.getCode() == ResponseCode.SUCCESS) {
                 ConsumeMessageDirectlyResult info = ConsumeMessageDirectlyResult.decode(response.getBody(), ConsumeMessageDirectlyResult.class);
                 future.complete(info);
@@ -405,6 +406,21 @@ public CompletableFuture<ConsumeMessageDirectlyResult> consumeMessageDirectly(St
         return future;
     }
 
+    private <T> void handleResponse(CompletableFuture<RemotingCommand> responseFuture,
+        CompletableFuture<T> resultFuture, Consumer<RemotingCommand> responseHandler) {
+        responseFuture.whenComplete((response, throwable) -> {
+            if (throwable != null) {
+                resultFuture.completeExceptionally(throwable);
+                return;
+            }
+            try {
+                responseHandler.accept(response);
+            } catch (Throwable t) {
+                resultFuture.completeExceptionally(t);
+            }
+        });
+    }
+
     private List<MessageExt> filterMessages(List<MessageExt> messageFoundList, String topic, String key,
         boolean uniqueKeyFlag) {
         List<MessageExt> matchedMessages = new ArrayList<>();
diff --git a/client/src/test/java/org/apache/rocketmq/client/impl/admin/MqClientAdminImplTest.java b/client/src/test/java/org/apache/rocketmq/client/impl/admin/MqClientAdminImplTest.java
index 520f4da5f2a..00740dcec51 100644
--- a/client/src/test/java/org/apache/rocketmq/client/impl/admin/MqClientAdminImplTest.java
+++ b/client/src/test/java/org/apache/rocketmq/client/impl/admin/MqClientAdminImplTest.java
@@ -66,6 +66,7 @@
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -525,6 +526,76 @@ public void assertConsumeMessageDirectlyWithError() {
         assertTrue(mqException.getMessage().contains("CODE: 1  DESC: null"));
     }
 
+    @Test
+    public void allOperationsShouldPropagateRemotingFailure() throws Exception {
+        CompletableFuture<RemotingCommand> remotingFuture = new CompletableFuture<>();
+        when(remotingClient.invoke(any(String.class), any(RemotingCommand.class), any(Long.class)))
+            .thenReturn(remotingFuture);
+
+        List<CompletableFuture<?>> futures = new ArrayList<>();
+        futures.add(mqClientAdminImpl.queryMessage(defaultBrokerAddr, false, false,
+            mock(QueryMessageRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.getTopicStatsInfo(defaultBrokerAddr,
+            mock(GetTopicStatsInfoRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.queryConsumeTimeSpan(defaultBrokerAddr,
+            mock(QueryConsumeTimeSpanRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.updateOrCreateTopic(defaultBrokerAddr,
+            mock(CreateTopicRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.updateOrCreateSubscriptionGroup(defaultBrokerAddr,
+            new SubscriptionGroupConfig(), defaultTimeout));
+        futures.add(mqClientAdminImpl.deleteTopicInBroker(defaultBrokerAddr,
+            mock(DeleteTopicRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.deleteTopicInNameserver(defaultBrokerAddr,
+            mock(DeleteTopicFromNamesrvRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.deleteKvConfig(defaultBrokerAddr,
+            mock(DeleteKVConfigRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.deleteSubscriptionGroup(defaultBrokerAddr,
+            mock(DeleteSubscriptionGroupRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.invokeBrokerToResetOffset(defaultBrokerAddr,
+            mock(ResetOffsetRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.viewMessage(defaultBrokerAddr,
+            mock(ViewMessageRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.getBrokerClusterInfo(defaultBrokerAddr, defaultTimeout));
+        futures.add(mqClientAdminImpl.getConsumerConnectionList(defaultBrokerAddr,
+            mock(GetConsumerConnectionListRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.queryTopicsByConsumer(defaultBrokerAddr,
+            mock(QueryTopicsByConsumerRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.querySubscriptionByConsumer(defaultBrokerAddr,
+            mock(QuerySubscriptionByConsumerRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.getConsumeStats(defaultBrokerAddr,
+            mock(GetConsumeStatsRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.queryTopicConsumeByWho(defaultBrokerAddr,
+            mock(QueryTopicConsumeByWhoRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.getConsumerRunningInfo(defaultBrokerAddr,
+            mock(GetConsumerRunningInfoRequestHeader.class), defaultTimeout));
+        futures.add(mqClientAdminImpl.consumeMessageDirectly(defaultBrokerAddr,
+            mock(ConsumeMessageDirectlyResultRequestHeader.class), defaultTimeout));
+
+        RemotingException expected = new RemotingException("invoke failed");
+        remotingFuture.completeExceptionally(expected);
+
+        assertEquals(19, futures.size());
+        for (CompletableFuture<?> future : futures) {
+            assertTrue("The returned future must complete after its remoting request fails", future.isDone());
+            assertTrue(future.isCompletedExceptionally());
+            ExecutionException actual = assertThrows(ExecutionException.class, future::get);
+            assertEquals(expected, actual.getCause());
+        }
+    }
+
+    @Test
+    public void operationShouldPropagateResponseHandlerFailure() throws Exception {
+        setResponseSuccess(null);
+
+        CompletableFuture<MessageExt> actual = mqClientAdminImpl.viewMessage(defaultBrokerAddr,
+            mock(ViewMessageRequestHeader.class), defaultTimeout);
+
+        assertTrue("The returned future must complete when response decoding fails", actual.isDone());
+        assertTrue(actual.isCompletedExceptionally());
+        ExecutionException thrown = assertThrows(ExecutionException.class, actual::get);
+        assertTrue(thrown.getCause() instanceof NullPointerException);
+    }
+
     private byte[] getMessageResult() throws Exception {
         byte[] bytes = MessageDecoder.encode(createMessageExt(), false);
         ByteBuffer byteBuffer = ByteBuffer.allocate(bytes.length);