diff --git a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java index 8537929be75..511d29e12ad 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java @@ -293,11 +293,7 @@ public TopicConfig createTopicInSendMessageMethod(final String topic, final Stri } if (createNew) { - if (brokerController.getBrokerConfig().isEnableSingleTopicRegister()) { - this.brokerController.registerSingleTopicAll(topicConfig); - } else { - this.brokerController.registerIncrementBrokerData(topicConfig, dataVersion); - } + registerBrokerData(topicConfig); } return topicConfig; @@ -337,11 +333,7 @@ public TopicConfig createTopicIfAbsent(TopicConfig topicConfig, boolean register log.error("createTopicIfAbsent ", e); } if (createNew && register) { - if (brokerController.getBrokerConfig().isEnableSingleTopicRegister()) { - this.brokerController.registerSingleTopicAll(topicConfig); - } else { - this.brokerController.registerIncrementBrokerData(topicConfig, dataVersion); - } + registerBrokerData(topicConfig); } return getTopicConfig(topicConfig.getTopicName()); } @@ -401,11 +393,7 @@ public TopicConfig createTopicInSendMessageBackMethod( } if (createNew) { - if (brokerController.getBrokerConfig().isEnableSingleTopicRegister()) { - this.brokerController.registerSingleTopicAll(topicConfig); - } else { - this.brokerController.registerIncrementBrokerData(topicConfig, dataVersion); - } + registerBrokerData(topicConfig); } return topicConfig; @@ -446,11 +434,7 @@ public TopicConfig createTopicOfTranCheckMaxTime(final int clientDefaultTopicQue } if (createNew) { - if (brokerController.getBrokerConfig().isEnableSingleTopicRegister()) { - this.brokerController.registerSingleTopicAll(topicConfig); - } else { - this.brokerController.registerIncrementBrokerData(topicConfig, dataVersion); - } + registerBrokerData(topicConfig); } return topicConfig; @@ -476,11 +460,7 @@ public void updateTopicUnitFlag(final String topic, final boolean unit) { dataVersion.nextVersion(stateMachineVersion); this.persist(); - if (brokerController.getBrokerConfig().isEnableSingleTopicRegister()) { - this.brokerController.registerSingleTopicAll(topicConfig); - } else { - this.brokerController.registerIncrementBrokerData(topicConfig, dataVersion); - } + registerBrokerData(topicConfig); } } @@ -503,11 +483,7 @@ public void updateTopicUnitSubFlag(final String topic, final boolean hasUnitSub) dataVersion.nextVersion(stateMachineVersion); this.persist(); - if (brokerController.getBrokerConfig().isEnableSingleTopicRegister()) { - this.brokerController.registerSingleTopicAll(topicConfig); - } else { - this.brokerController.registerIncrementBrokerData(topicConfig, dataVersion); - } + registerBrokerData(topicConfig); } } @@ -699,6 +675,14 @@ private Map current(String topic) { } } + private void registerBrokerData(TopicConfig topicConfig) { + if (brokerController.getBrokerConfig().isEnableSingleTopicRegister()) { + this.brokerController.registerSingleTopicAll(topicConfig); + } else { + this.brokerController.registerIncrementBrokerData(topicConfig, dataVersion); + } + } + public boolean containsTopic(String topic) { return topicConfigTable.containsKey(topic); }