public void GetRetainedMsg()

in mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/meta/RetainedMsgClient.java [149:196]
44 lines of code
5 McCabe index (conditional complexity)

    public void GetRetainedMsg(String topic, CompletableFuture<Message> future) throws RemotingException, InterruptedException {
        String groupId = whichGroup();
        HashMap<String, String> option = new HashMap<>();
        option.put("topic", topic);

        final ReadRequest request = ReadRequest.newBuilder()
                .setGroup(groupId)
                .setOperation("topic")
                .setType(READ_INDEX_TYPE)
                .putAllExtData(option)
                .setCategory(CATEGORY_RETAINED_MSG)
                .build();

        metaRpcClient.getCliClientService().getRpcClient().invokeAsync(metaRpcClient.getLeader(groupId).getEndpoint(), request, new InvokeCallback() {

            @Override
            public void complete(Object result, Throwable err) {
                if (err == null) {
                    Response rsp = (Response) result;
                    if (!rsp.getSuccess()) {
                        logger.info("GetRetainedMsg failed. {}", rsp.getErrMsg());
                        future.complete(null);
                        return;
                    }
                    if (rsp.getData().toStringUtf8().equals(NOT_FOUND)) {  //this topic doesn't exist retained msg
                        future.complete(null);
                        return;
                    }
                    Message message = null;
                    try {
                        message = Message.copyFromStoreMessage(StoreMessage.parseFrom(rsp.getData().toByteArray()));
                    } catch (InvalidProtocolBufferException e) {
                        future.complete(null);
                        throw new RuntimeException(e);
                    }
                    future.complete(message);
                } else {
                    logger.error("", err);
                    future.complete(null);
                }
            }

            @Override
            public Executor executor() {
                return null;
            }
        }, 5000);
    }