Fix the issue of sending message "No route information of this topic: xxx" when the producer does not configure the namespace

[aoshiguchen]

What is the purpose of the change

Solve the problem of sending message errors when producers do not configure namespaces。

In the current situation, when the namespace is not configured, the method of setting the namespace is called when instantiating the bean, causing the producer to initialize the namespace to be empty, but 'ClientConfig # namespaceInitialized' is set to true. This leads to the producer sending an error message: "No route information of this topic: xxx"

Brief changelog

When instantiating beans ("defaultMQProducer", "defaultLitePullConsumer"), the set namespace method is called only when the namespace is configured。

Verifying this change

Verified！

Follow this checklist to help us incorporate your contribution quickly and easily. Notice, it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR.

• Make sure there is a Github issue filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue.
• Format the pull request title like [ISSUE #123] Fix UnknownException when host config not exist. Each commit in the pull request should have a meaningful subject line and body.
• Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
• Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when cross module dependency exist.
• Run mvn -B clean apache-rat:check findbugs:findbugs checkstyle:checkstyle to make sure basic checks pass. Run mvn clean install -DskipITs to make sure unit-test pass. Run mvn clean test-compile failsafe:integration-test to make sure integration-test pass.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

[panzhi33]

Do you have detailed error information? The judgment of namespance is done in rocketmq-client

org.apache.rocketmq.common.protocol.NamespaceUtil#wrapNamespace

[aoshiguchen]

Do you have detailed error information? The judgment of namespance is done in rocketmq-client

org.apache.rocketmq.common.protocol.NamespaceUtil#wrapNamespace

1、My MQ instance has been released and I am currently unable to test and obtain detailed error information.
2、But I can provide the exact error location and a complete reproduction process on my end.
3、Error source code location：org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#sendDefaultImpl.
This line：
throw (new MQClientException("No route info of this topic: " + msg.getTopic() + FAQUrl.suggestTodo("http://rocketmq.apache.org/docs/faq/"), (Throwable)null)).setResponseCode(10005);
4、Problem reproduction process
4.1、Introducing dependencies: org.apache.rocketmq:rocketmq-spring-boot-starter:2.2.3
4.2、configuration groupId、accessKey、secretKey、nameServer .
4.3、Sending messages using RocketMQTemplate instance
4.4、Start the project and execute it

5、For ease of understanding, you can refer to the following program to manually build RocketMQTemplate instances to reproduce the problem

public class Test3 {

    public static void main(String[] args) throws Exception {
        String groupId = "xxxx";
        String accessKey = "xxx";
        String secretKey = "xxx";
        String onsAddr = "xxx";

        RocketMQProperties properties = new RocketMQProperties();
        RocketMQProperties.Producer p = new RocketMQProperties.Producer();
        p.setGroup(groupId);
        p.setAccessKey(accessKey);
        p.setSecretKey(secretKey);

        properties.setNameServer(onsAddr);
        properties.setProducer(p);
        DefaultMQProducer producer = defaultMQProducer(properties);
        producer.start();

        RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
        rocketMQTemplate.setProducer(producer);
        rocketMQTemplate.setMessageConverter(new MappingFastJsonMessageConverter());

        SendResult sendResult = rocketMQTemplate.syncSend("TP_TEST1", "hello");
        System.out.println("msgId:" + sendResult);
    }

    /**
     * This method is derived from：org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration#defaultMQProducer(org.apache.rocketmq.spring.autoconfigure.RocketMQProperties)
     * @param rocketMQProperties
     * @return
     */
    public static DefaultMQProducer defaultMQProducer(RocketMQProperties rocketMQProperties) {
        RocketMQProperties.Producer producerConfig = rocketMQProperties.getProducer();
        String nameServer = rocketMQProperties.getNameServer();
        String groupName = producerConfig.getGroup();
        Assert.hasText(nameServer, "[rocketmq.name-server] must not be null");
        Assert.hasText(groupName, "[rocketmq.producer.group] must not be null");
        String accessChannel = rocketMQProperties.getAccessChannel();
        String ak = rocketMQProperties.getProducer().getAccessKey();
        String sk = rocketMQProperties.getProducer().getSecretKey();
        boolean isEnableMsgTrace = rocketMQProperties.getProducer().isEnableMsgTrace();
        String customizedTraceTopic = rocketMQProperties.getProducer().getCustomizedTraceTopic();
        DefaultMQProducer producer = RocketMQUtil.createDefaultMQProducer(groupName, ak, sk, isEnableMsgTrace, customizedTraceTopic);
        producer.setNamesrvAddr(nameServer);
        if (StringUtils.hasLength(accessChannel)) {
            producer.setAccessChannel(AccessChannel.valueOf(accessChannel));
        }

        producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
        producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
        producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
        producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
        producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyThreshold());
        producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());
        producer.setUseTLS(producerConfig.isTlsEnable());
        producer.setNamespace(producerConfig.getNamespace());
        producer.setInstanceName(producerConfig.getInstanceName());
        return producer;
    }

}

[aoshiguchen]

Do you have detailed error information? The judgment of namespance is done in rocketmq-client

org.apache.rocketmq.common.protocol.NamespaceUtil#wrapNamespace

In the Default MQProducer, this line of code will be executed before sending a message: "msg. setTopic (this. with Namespace (msg. getTopic()); ". If "org. apache. locketmq. client. ClientConfig# namespaceInitialized" is false, it means that the namespace has not been initialized and the program will automatically initialize. But because in "org. apache. locketmq. spring. autoconfiguration. RocketMQAutoConfiguration # defaultMQProducer", even if a namespace is not configured, the value will still be affected by executing the code: "producer. setNamespace (productConfiguration. getNamespace())" And set to true. As a result, when sending messages, the defaultMQProducer mistakenly thought of the namespace.