public CompletableFuture<List<SendResult>> sendMessage(ProxyContext ctx, QueueSelector queueSelector,
        String producerGroup, int sysFlag, List<Message> messageList, long timeoutMillis) {
        CompletableFuture<List<SendResult>> future = new CompletableFuture<>();
        long beginTimestampFirst = System.currentTimeMillis();
        AddressableMessageQueue messageQueue = null;
        try {
            Message message = messageList.get(0);
            String topic = message.getTopic();
            if (ConfigurationManager.getProxyConfig().isEnableTopicMessageTypeCheck()) {
                if (topicMessageTypeValidator != null) {
                    // Do not check retry or dlq topic
                    if (!NamespaceUtil.isRetryTopic(topic) && !NamespaceUtil.isDLQTopic(topic)) {
                        TopicMessageType topicMessageType = serviceManager.getMetadataService().getTopicMessageType(ctx, topic);
                        TopicMessageType messageType = TopicMessageType.parseFromMessageProperty(message.getProperties());
                        topicMessageTypeValidator.validate(topicMessageType, messageType);
                    }
                }
            }
            messageQueue = queueSelector.select(ctx,
                this.serviceManager.getTopicRouteService().getCurrentMessageQueueView(ctx, topic));
            if (messageQueue == null) {
                throw new ProxyException(ProxyExceptionCode.FORBIDDEN, "no writable queue");
            }

            for (Message msg : messageList) {
                MessageClientIDSetter.setUniqID(msg);
            }
            SendMessageRequestHeader requestHeader = buildSendMessageRequestHeader(messageList, producerGroup, sysFlag, messageQueue.getQueueId());

            AddressableMessageQueue finalMessageQueue = messageQueue;
            future = this.serviceManager.getMessageService().sendMessage(
                ctx,
                messageQueue,
                messageList,
                requestHeader,
                timeoutMillis)
                .thenApplyAsync(sendResultList -> {
                    for (SendResult sendResult : sendResultList) {
                        int tranType = MessageSysFlag.getTransactionValue(requestHeader.getSysFlag());
                        if (SendStatus.SEND_OK.equals(sendResult.getSendStatus()) &&
                            tranType == MessageSysFlag.TRANSACTION_PREPARED_TYPE &&
                            StringUtils.isNotBlank(sendResult.getTransactionId())) {
                            fillTransactionData(ctx, producerGroup, finalMessageQueue, sendResult, messageList);
                        }
                    }
                    return sendResultList;
                }, this.executor)
                    .whenComplete((result, exception) -> {
                        long endTimestamp = System.currentTimeMillis();
                        if (exception != null) {
                            this.serviceManager.getTopicRouteService().updateFaultItem(finalMessageQueue.getBrokerName(), endTimestamp - beginTimestampFirst, true, false);
                        } else {
                            this.serviceManager.getTopicRouteService().updateFaultItem(finalMessageQueue.getBrokerName(),endTimestamp - beginTimestampFirst, false, true);
                        }
                    });
        } catch (Throwable t) {
            future.completeExceptionally(t);
        }
        return FutureUtils.addExecutor(future, this.executor);
    }
@Test
    public void testSendRetryMessage() throws Throwable {
        String txId = MessageClientIDSetter.createUniqID();
        String msgId = MessageClientIDSetter.createUniqID();
        long commitLogOffset = 1000L;
        long queueOffset = 100L;

        SendResult sendResult = new SendResult();
        sendResult.setSendStatus(SendStatus.SEND_OK);
        sendResult.setTransactionId(txId);
        sendResult.setMsgId(msgId);
        sendResult.setOffsetMsgId(createOffsetMsgId(commitLogOffset));
        sendResult.setQueueOffset(queueOffset);
        ArgumentCaptor<SendMessageRequestHeader> requestHeaderArgumentCaptor = ArgumentCaptor.forClass(SendMessageRequestHeader.class);
        when(this.messageService.sendMessage(any(), any(), any(), requestHeaderArgumentCaptor.capture(), anyLong()))
            .thenReturn(CompletableFuture.completedFuture(Lists.newArrayList(sendResult)));

        List<Message> messageExtList = new ArrayList<>();
        Message messageExt = createMessageExt(MixAll.getRetryTopic(CONSUMER_GROUP), "tag", 0, 0);
        MessageAccessor.putProperty(messageExt, MessageConst.PROPERTY_RECONSUME_TIME, "1");
        MessageAccessor.putProperty(messageExt, MessageConst.PROPERTY_MAX_RECONSUME_TIMES, "16");
        messageExtList.add(messageExt);
        AddressableMessageQueue messageQueue = mock(AddressableMessageQueue.class);
        when(messageQueue.getBrokerName()).thenReturn("mockBroker");

        ArgumentCaptor<String> brokerNameCaptor = ArgumentCaptor.forClass(String.class);
        ArgumentCaptor<Long> tranStateTableOffsetCaptor = ArgumentCaptor.forClass(Long.class);
        ArgumentCaptor<Long> commitLogOffsetCaptor = ArgumentCaptor.forClass(Long.class);
        when(transactionService.addTransactionDataByBrokerName(
            any(),
            brokerNameCaptor.capture(),
            anyString(),
            anyString(),
            tranStateTableOffsetCaptor.capture(),
            commitLogOffsetCaptor.capture(),
            anyString(), any())).thenReturn(mock(TransactionData.class));

        List<SendResult> sendResultList = this.producerProcessor.sendMessage(
            createContext(),
            (ctx, messageQueueView) -> messageQueue,
            PRODUCER_GROUP,
            MessageSysFlag.TRANSACTION_PREPARED_TYPE,
            messageExtList,
            3000
        ).get();

        assertNotNull(sendResultList);
        assertEquals("mockBroker", brokerNameCaptor.getValue());
        assertEquals(queueOffset, tranStateTableOffsetCaptor.getValue().longValue());
        assertEquals(commitLogOffset, commitLogOffsetCaptor.getValue().longValue());

        SendMessageRequestHeader requestHeader = requestHeaderArgumentCaptor.getValue();
        assertEquals(PRODUCER_GROUP, requestHeader.getProducerGroup());
        assertEquals(MixAll.getRetryTopic(CONSUMER_GROUP), requestHeader.getTopic());
        assertEquals(1, requestHeader.getReconsumeTimes().intValue());
        assertEquals(16, requestHeader.getMaxReconsumeTimes().intValue());
    }