Auto create topic question

[makabakaboom]

BUG REPORT

1. Please describe the issue you observed:

• What did you do (The steps to reproduce)?

broker端开启了自动创建Topic，生产端默认配置，首先发送一条消息，超过30s后在发送消息，只会在一个master节点上面创建topic，不会再所有master节点上面创建topic

• What did you expect to see?

所有master节点都会自动创建topic

• What did you see instead?

只有一个master节点有topic

2. Please tell us about your environment:

broker 4.3.0
client 4.3.0
3 namesrv 2 master 2 slave

3. Other information (e.g. detailed explanation, logs, related issues, suggestions how to fix, etc):

     分析自动创建topic源码发现，client第一次发送会根据AUTO_CREATE_TOPIC_KEY从namesrv拿到topicRouteData作为新创建topic的topicRouteData，而MQClientInstance里面每30s会拉取topic的topicRouteData最新信息并更新。

     异常情况1：Producer发送一条信息后关闭掉Producer，这个时候只有在一个broker上面有topic信息，下一次启动Producer发送消息的时候就会直接从NameSrv拿到最新的topicRouteData，这样就只能在一个master节点上面创建topic

     异常情况2：Producer发送一条信息后，这个时候只有在一个broker上面有topic信息，隔35s在发送一条信息，在这段时间内，已经完成MQClientInstance对Topic Route的更新，topicRouteData只会拉取到一个master节点，这样也只能在一个master节点上面创建topic

[AnyOSR]

貌似会存在这个问题，达不到defaultTopicQueueNums数目的messageQueue

[makabakaboom]

目前就是存在这个问题，主要的原因是自动创建topic并不是真正直接的创建topic。而是拿到AUTO_CREATE_TOPIC_KEY_TOPIC的路由信息在消息发送到Broker端的时候才会真正创建topic，这样就会造成上面两种异常情况

[AnyOSR]

我知道创建新topic时会根据default的路由信息来创建，只是不确定有没有别的地方会去别的master上去创建这个topic信息，没有的话肯定会存在你说的这个问题，有的话，就不可能存在了

[AnyOSR]

明天看下代码~~

[zongtanghu]

@davidpdw Yes,you're right.If you opened the autoCreateTopicEnable flag,the cases what you said above will appear.But I think it's not the exception.Firstly,we recommend user's close the autoCreateTopicEnable flag if they will use it in production enviroment.

[makabakaboom]

@zongtanghu 恩恩，不过如果 建议 关闭自动创建topic是因为这个原因的话，那我觉得rocketmq对用户可能不太友好。而且正常情况下是能都创建的，既然这样，为什么就不保证在各个情况下都能创建成功呢，这样这个功能也会更加完善

[duhengforever]

@davidpdw 建议关闭自动创建topic的原因还有一个原因在于topic属于资源，topic的新增会带来很多的资源上的使用。另外，issue 一般用于bug report或者是feature quest，如果有这种咨询问题，建议订阅邮件列表（https://rocketmq.apache.org/about/contact/ ）里面搜索提问，会有很多小伙伴帮助你的。

[makabakaboom]

@duhengforever 好的，下次先使用邮件列表，但是这个问题我觉得是自动创建的bug，所以就在这里提交issue

[2259289435]

@zongtanghu 恩恩，不过如果 建议 关闭自动创建topic是因为这个原因的话，那我觉得rocketmq对用户可能不太友好。而且正常情况下是能都创建的，既然这样，为什么就不保证在各个情况下都能创建成功呢，这样这个功能也会更加完善

同样的疑问，不知后面你是通过何种方式实现自动创建topic到集群下所有的broker

[makabakaboom]

@2259289435 目前我们是在控制台直接修改，将Topic扩展到多个master节点上面。自己考虑的几种方式1、在控制台上创建Topic 2、producer和consumer以及admin API里面其实有创建Topic API，应用启动时将对应的Topic初始化就可以了 3、还有一种方式就是设计更合理的自动创建方式，这个就需要修改源码重新打包

[2259289435]

@2259289435 目前我们是在控制台直接修改，将Topic扩展到多个master节点上面。自己考虑的几种方式1、在控制台上创建Topic 2、producer和consumer以及admin API里面其实有创建Topic API，应用启动时将对应的Topic初始化就可以了 3、还有一种方式就是设计更合理的自动创建方式，这个就需要修改源码重新打包

了解，还是采取2自己来控制比较方便。顺序的就单broker，非顺序的可以集群创建，而不是单broker创建。
2的话MQAdminImpl里面的createTopic(key, newTopic, queueNum, topicSysFlag)倒是找到方法了，只是方法上都没有注释

[mali-nuist]

@2259289435 请问你们通过api是如何创建topic的呢，有例子么？也没文档。。

[2259289435]

clinet,broker version 4.6.0，if client，broker version not equals，get broker createtopickey first producer： SendMessageHook.sendMessageBefore（final SendMessageContext context） 1. getTopic String newTopic = NamespaceUtil.withoutNamespace(context.getMessage().getTopic()); 2.createTopicForCache createTopicCache.get(newTopic, new Callable<Boolean>() { @OverRide public Boolean call() throws Exception { Set<String> topicList = context.getProducer().getmQClientFactory().getMQClientAPIImpl().getTopicListFromNameServer(20000L).getTopicList(); if(!topicList.contains(newTopic)) { log.info("producerGroup : {}, newTopic : {} begin autoCreate", context.getProducer().getDefaultMQProducer().getProducerGroup(), newTopic); context.getProducer().getmQClientFactory().getMQAdminImpl().createTopic(MixAllPlus.AUTO_CREATE_TOPIC_KEY_TOPIC, newTopic, DEFAULT_AUTO_CREATE_TOPIC_QUEUE_SIZE); log.info("producerGroup : {}, newTopic : {} end autoCreate", context.getProducer().getDefaultMQProducer().getProducerGroup(), newTopic); } return true; } }); pushconsumer. 1. beforeStart(pushconsumer consumer) //custom method 2. create instance DefaultMQPushConsumerImpl impl = consumer.getDefaultMQPushConsumerImpl(); if (consumer.getMessageModel() == MessageModel.CLUSTERING) { consumer.changeInstanceNameToPID(); } RPCHook rpcHook = ReflectUtil.on(impl).field("rpcHook").get(); MQClientInstance instance = MQClientManager.getInstance().getOrCreateMQClientInstance(consumer, rpcHook); instance.start(); 3.getNewTopic Set<String> newTopics = Sets.newHashSet(); Set<String> topicList = instance.getMQClientAPIImpl().getTopicListFromNameServer(20000L).getTopicList(); for(String topic : impl.getSubscriptionInner().keySet()) { String newTopic = NamespaceUtil.withoutNamespace(topic); if(!topicList.contains(newTopic)) { newTopics.add(newTopic); } } 4.createTopic for(final String newTopic : newTopics) { createTopicCache.get(newTopic, new Callable<Boolean>() { @OverRide public Boolean call() throws Exception { log.info("consumerGroup : {}, newTopic : {} begin autoCreate", consumer.getConsumerGroup(), newTopic); instance.getMQAdminImpl().createTopic(instance.getDefaultMQProducer().getCreateTopicKey(), newTopic, DEFAULT_AUTO_CREATE_TOPIC_QUEUE_SIZE); log.info("consumerGroup : {}, newTopic : {} end autoCreate", consumer.getConsumerGroup(), newTopic); return true; } }); }