Fetch topic route when registering message queue changed listener

apache · May 15, 2023 · 8f8716c · 8f8716c
1 parent 8d6c708
commit 8f8716c
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 4 deletions.
diff --git a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/consumer/PullConsumer.java b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/consumer/PullConsumer.java
@@ -37,7 +37,7 @@ public interface PullConsumer extends Closeable {
      * @param topic    the topic that needs to be monitored.
      * @param listener the callback to detect the message queue changes.
      */
-    void registerMessageQueueChangeListenerByTopic(String topic, TopicMessageQueueChangeListener listener);
+    void registerMessageQueueChangeListenerByTopic(String topic, TopicMessageQueueChangeListener listener) throws ClientException;
 
     /**
      * Fetch message queues of the topic.

diff --git a/.../client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PullConsumerImpl.java b/.../client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PullConsumerImpl.java
@@ -164,8 +164,9 @@ public String getConsumerGroup() {
     }
 
     @Override
-    public synchronized void registerMessageQueueChangeListenerByTopic(String topic,
-        TopicMessageQueueChangeListener listener) {
+    public void registerMessageQueueChangeListenerByTopic(String topic, TopicMessageQueueChangeListener listener)
+        throws ClientException {
+        // TODO: add lock?
         checkNotNull(topic, "topic should not be null");
         checkNotNull(listener, "listener should not be null");
         if (!this.isRunning()) {
@@ -178,6 +179,7 @@ public synchronized void registerMessageQueueChangeListenerByTopic(String topic,
                 topic);
         }
         topicMessageQueueChangeListenerMap.put(topic, listener);
+        fetchMessageQueues(topic);
     }
 
     public int getMaxCacheMessageCountEachQueue() {
@@ -537,7 +539,8 @@ private List<MessageQueueImpl> transformTopicRouteData(TopicRouteData topicRoute
             .collect(Collectors.toList());
     }
 
-    public synchronized void onTopicRouteDataUpdate0(String topic, TopicRouteData topicRouteData) {
+    public void onTopicRouteDataUpdate0(String topic, TopicRouteData topicRouteData) {
+        // TODO: add lock?
         final List<MessageQueueImpl> newMqs = transformTopicRouteData(topicRouteData);
         Set<MessageQueue> newMqSet = new HashSet<>(newMqs);
         final List<MessageQueueImpl> oldMqs = topicMessageQueuesCache.get(topic);