public void setRetainedMsg()

in mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/meta/RetainedMsgClient.java [57:96]
35 lines of code
3 McCabe index (conditional complexity)

    public void setRetainedMsg(String topic, Message msg, CompletableFuture<Boolean> future) throws RemotingException, InterruptedException {
        String groupId = whichGroup();
        HashMap<String, String> option = new HashMap<>();
        option.put("topic", topic);
        option.put("firstTopic", msg.getFirstTopic());
        option.put("isEmpty", String.valueOf(msg.isEmpty()));

        logger.debug("SetRetainedMsg option:" + option);

        final WriteRequest request = WriteRequest.newBuilder()
                .setGroup(groupId)
                .setData(ByteString.copyFrom(msg.getEncodeBytes()))
                .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.error("SetRetainedMsg failed. {}", rsp.getErrMsg());
                        future.complete(false);
                        return;
                    }
                    future.complete(true);
                } else {
                    logger.error("", err);
                    future.complete(false);
                }
            }

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

    }