[ISSUE #10892] Tolerate concurrent topic cache eviction

client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java

@@ -21,6 +21,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Random;
 import java.util.Set;
@@ -178,10 +179,11 @@ public String resolve(String name) {
         }, serviceDetector);
     }
     private Optional<String> pickTopic() {
-        if (topicPublishInfoTable.isEmpty()) {
+        Iterator<String> iterator = topicPublishInfoTable.keySet().iterator();
+        if (!iterator.hasNext()) {
             return Optional.empty();
         }
-        return Optional.of(topicPublishInfoTable.keySet().iterator().next());
+        return Optional.of(iterator.next());
     }
     public void registerCheckForbiddenHook(CheckForbiddenHook checkForbiddenHook) {
         this.checkForbiddenHookList.add(checkForbiddenHook);

client/src/test/java/org/apache/rocketmq/client/producer/selector/DefaultMQProducerImplTest.java

@@ -54,6 +54,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
@@ -222,6 +223,22 @@ public void testInitTopicRoute() throws NoSuchMethodException, InvocationTargetE
         method.invoke(defaultMQProducerImpl);
     }
 
+    @Test
+    public void testPickTopicToleratesConcurrentCacheEviction() throws Exception {
+        ClearingConcurrentMap<String, TopicPublishInfo> topicPublishInfoTable = new ClearingConcurrentMap<>();
+        topicPublishInfoTable.put(defaultTopic, mock(TopicPublishInfo.class));
+        setField(defaultMQProducerImpl, "topicPublishInfoTable", topicPublishInfoTable);
+
+        Method method = DefaultMQProducerImpl.class.getDeclaredMethod("pickTopic");
+        method.setAccessible(true);
+
+        Optional<String> topic = (Optional<String>) method.invoke(defaultMQProducerImpl);
+
+        // ConcurrentHashMap iterators are weakly consistent, so a candidate observed before
+        // eviction is valid. The important contract is that selection never throws.
+        assertNotNull(topic);
+    }
+
     @Test
     public void assertFetchPublishMessageQueues() throws MQClientException {
         List<MessageQueue> actual = defaultMQProducerImpl.fetchPublishMessageQueues(defaultTopic);
@@ -382,4 +399,18 @@ private void setField(final Object target, final String fieldName, final Object
         field.setAccessible(true);
         field.set(target, newValue);
     }
+
+    private static class ClearingConcurrentMap<K, V> extends ConcurrentHashMap<K, V> {
+        private boolean clearOnFirstIsEmpty = true;
+
+        @Override
+        public boolean isEmpty() {
+            boolean empty = super.isEmpty();
+            if (clearOnFirstIsEmpty) {
+                clearOnFirstIsEmpty = false;
+                super.clear();
+            }
+            return empty;
+        }
+    }
 }

proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteService.java

@@ -22,6 +22,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -134,10 +135,11 @@ public String resolve(String name) {
 
     // pickup one topic in the topic cache
     private Optional<String> pickTopic() {
-        if (topicCache.asMap().isEmpty()) {
+        Iterator<String> iterator = topicCache.asMap().keySet().iterator();
+        if (!iterator.hasNext()) {
             return Optional.empty();
         }
-        return Optional.of(topicCache.asMap().keySet().iterator().next());
+        return Optional.of(iterator.next());
     }
 
     protected void init() {

proxy/src/test/java/org/apache/rocketmq/proxy/service/route/ClusterTopicRouteServiceTest.java

@@ -22,8 +22,12 @@
 import com.github.benmanes.caffeine.cache.LoadingCache;
 import com.google.common.net.HostAndPort;
 
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -50,6 +54,7 @@
 import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 public class ClusterTopicRouteServiceTest extends BaseServiceTest {
@@ -126,6 +131,22 @@ public void testGetTopicRouteForProxy() throws Throwable {
         assertEquals(addressList, proxyTopicRouteData.getBrokerDatas().get(0).getBrokerAddrs().get(MixAll.MASTER_ID));
     }
 
+    @Test
+    public void testPickTopicToleratesConcurrentCacheEviction() throws Exception {
+        LoadingCache<String, MessageQueueView> topicCache = mock(LoadingCache.class);
+        ClearingConcurrentMap<String, MessageQueueView> topicCacheMap = new ClearingConcurrentMap<>();
+        topicCacheMap.put(TOPIC, mock(MessageQueueView.class));
+        when(topicCache.asMap()).thenReturn(topicCacheMap);
+        setField(topicRouteService, "topicCache", topicCache);
+
+        Method method = TopicRouteService.class.getDeclaredMethod("pickTopic");
+        method.setAccessible(true);
+
+        Optional<String> topic = (Optional<String>) method.invoke(topicRouteService);
+
+        assertNotNull(topic);
+    }
+
     @Test
     public void testTopicRouteCaffeineCache() throws InterruptedException {
         String key = "abc";
@@ -164,4 +185,24 @@ public void testTopicRouteCaffeineCache() throws InterruptedException {
         TimeUnit.SECONDS.sleep(5);
         assertThat(value).isEqualTo(topicCache.get(key));
     }
+
+    private static void setField(Object target, String fieldName, Object value) throws Exception {
+        Field field = TopicRouteService.class.getDeclaredField(fieldName);
+        field.setAccessible(true);
+        field.set(target, value);
+    }
+
+    private static class ClearingConcurrentMap<K, V> extends ConcurrentHashMap<K, V> {
+        private boolean clearOnFirstIsEmpty = true;
+
+        @Override
+        public boolean isEmpty() {
+            boolean empty = super.isEmpty();
+            if (clearOnFirstIsEmpty) {
+                clearOnFirstIsEmpty = false;
+                super.clear();
+            }
+            return empty;
+        }
+    }
 }