@Override
    public CompletableFuture<List<MessageExt>> queryMessage(String address, boolean uniqueKeyFlag, boolean decompressBody,
        QueryMessageRequestHeader requestHeader, long timeoutMillis) {
        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 -> {
            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));
            } else if (response.getCode() == ResponseCode.QUERY_NOT_FOUND)  {
                List<MessageExt> wrappers = new ArrayList<>();
                future.complete(wrappers);
            } else {
                log.warn("queryMessage getResponseCommand failed, {} {}, header={}", response.getCode(), response.getRemark(), requestHeader);
                future.completeExceptionally(new MQClientException(response.getCode(), response.getRemark()));
            }
        });

        return future;
    }
@Test
    public void assertQueryMessageWithNotFound() throws Exception {
        when(response.getCode()).thenReturn(ResponseCode.QUERY_NOT_FOUND);
        QueryMessageRequestHeader requestHeader = mock(QueryMessageRequestHeader.class);
        CompletableFuture<List<MessageExt>> actual = mqClientAdminImpl.queryMessage(defaultBrokerAddr, false, false, requestHeader, defaultTimeout);
        List<MessageExt> messageExtList = actual.get();
        assertNotNull(messageExtList);
        assertEquals(0, messageExtList.size());
    }