@Override
    public void sendMessage(final Message message, final MessageSentCallback cb) throws TubeClientException,
            InterruptedException {
        MessageSentResult result = checkMessageAndStatus(message);
        if (!result.isSuccess()) {
            cb.onMessageSent(result);
            return;
        }
        final Partition partition =
                this.selectPartition(message, BrokerWriteService.AsyncService.class);
        final int brokerId = partition.getBrokerId();
        long startTime = System.currentTimeMillis();
        try {
            this.brokerRcvQltyStats.addSendStatistic(brokerId);
            getAsyncBrokerService(partition.getBroker()).sendMessageP2B(
                    createSendMessageRequest(partition, message),
                    AddressUtils.getLocalAddress(), producerConfig.isTlsEnable(),
                    new Callback() {

                        @Override
                        public void handleResult(Object result) {
                            if (!(result instanceof ClientBroker.SendMessageResponseB2P)) {
                                return;
                            }
                            final ClientBroker.SendMessageResponseB2P responseB2P =
                                    (ClientBroker.SendMessageResponseB2P) result;
                            final MessageSentResult rt =
                                    SimpleMessageProducer.this.buildMsgSentResult(
                                            System.currentTimeMillis() - startTime,
                                            message, partition, responseB2P);
                            partition.resetRetries();
                            brokerRcvQltyStats.addReceiveStatistic(brokerId,
                                    responseB2P.getSuccess());
                            if (!responseB2P.getSuccess()
                                    && responseB2P.getErrCode() == TErrCodeConstants.SERVICE_UNAVAILABLE) {
                                rpcServiceFactory.addUnavailableBroker(brokerId);
                            }
                            cb.onMessageSent(rt);
                        }

                        @Override
                        public void handleError(Throwable error) {
                            producerManager.getClientMetrics().bookFailRpcCall(
                                    TErrCodeConstants.UNSPECIFIED_ABNORMAL);
                            partition.increRetries(1);
                            brokerRcvQltyStats.addReceiveStatistic(brokerId, false);
                            cb.onException(error);
                        }
                    });
            rpcServiceFactory.resetRmtAddrErrCount(partition.getBroker().getBrokerAddr());
        } catch (final Throwable e) {
            if (e instanceof LocalConnException) {
                rpcServiceFactory.addRmtAddrErrCount(partition.getBroker().getBrokerAddr());
            }
            // if failed,increment the counter
            partition.increRetries(1);
            this.brokerRcvQltyStats.addReceiveStatistic(brokerId, false);
            cb.onException(e);
        }
    }
Check if the given topic accept publish message.

@param topic topic name
@return if accept message
@throws TubeClientException
inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/producer/SimpleMessageProducer.java