Resolving the Thread-Safety Problem in registerReferenceKeyAndBeanName Using an Overloaded computeIfAbsent

dubbo-common/src/main/java/org/apache/dubbo/common/utils/ConcurrentHashMapUtils.java

@@ -18,6 +18,7 @@
 
 import java.util.Objects;
 import java.util.concurrent.ConcurrentMap;
+import java.util.function.Consumer;
 import java.util.function.Function;
 
 /**
@@ -27,11 +28,17 @@ public class ConcurrentHashMapUtils {
 
     /**
      * A temporary workaround for Java 8 ConcurrentHashMap#computeIfAbsent specific performance issue: JDK-8161372.</br>
-     * @see <a href="https://bugs.openjdk.java.net/browse/JDK-8161372">https://bugs.openjdk.java.net/browse/JDK-8161372</a>
      *
+     * @see <a href="https://bugs.openjdk.java.net/browse/JDK-8161372">https://bugs.openjdk.java.net/browse/JDK-8161372</a>
      */
     public static <K, V> V computeIfAbsent(ConcurrentMap<K, V> map, K key, Function<? super K, ? extends V> func) {
+        return computeIfAbsent(map, key, func, null);
+    }
+
+    public static <K, V> V computeIfAbsent(
+            ConcurrentMap<K, V> map, K key, Function<? super K, ? extends V> func, Consumer<V> threadSafeOperation) {
         Objects.requireNonNull(func);
+        V value;
         if (JRE.JAVA_8.isCurrentVersion()) {
             V v = map.get(key);
             if (null == v) {
@@ -47,13 +54,23 @@ public static <K, V> V computeIfAbsent(ConcurrentMap<K, V> map, K key, Function<
                 if (null != res) {
                     // if pre value present, means other thread put value already, and putIfAbsent not effect
                     // return exist value
-                    return res;
+                    value = res;
+                } else {
+                    value = v;
                 }
                 // if pre value is null, means putIfAbsent effected, return current value
+            } else {
+                value = v;
             }
-            return v;
         } else {
-            return map.computeIfAbsent(key, func);
+            value = map.computeIfAbsent(key, func);
+        }
+        if (value != null && threadSafeOperation != null) {
+            // make sure value operations are thread - safe.
+            synchronized (value) {
+                threadSafeOperation.accept(value);
+            }
         }
+        return value;
     }
 }

dubbo-common/src/main/java/org/apache/dubbo/config/annotation/ProvidedBy.java

@@ -25,7 +25,9 @@
 
 /**
  * Class-level annotation used for declaring Dubbo interface.
- * Example:
+ * Example: <br/>
+ * <pre>
+ * {@code
  * @ProvidedBy("dubbo-samples-xds-provider")
  * public interface GreetingService {
  *     String sayHello(String name);
@@ -36,6 +38,7 @@
  *     @DubboReference(version = "1.0.0")
  *     private GreetingService greetingService;
  * }
+ * </pre>
  */
 @Documented
 @Retention(RetentionPolicy.RUNTIME)

dubbo-common/src/test/java/org/apache/dubbo/common/utils/ConcurrentHashMapUtilsTest.java

@@ -16,16 +16,58 @@
  */
 package org.apache.dubbo.common.utils;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.RepeatedTest;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.condition.EnabledForJreRange;
+import org.junit.jupiter.api.condition.JRE;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
 class ConcurrentHashMapUtilsTest {
 
+    private static final String TEST_KEY = "testKey";
+
+    /**
+     * number of executions
+     * 10 K
+     */
+    private static final int REPEATED_TEST_NUM = 10_000;
+
+    private static ConcurrentMap<String, List<Long>> SHARED_MAP_JDK8;
+
+    private static ConcurrentMap<String, List<Long>> SHARED_MAP_JDK17;
+
+    @BeforeAll
+    public static void initSharedMap() {
+        SHARED_MAP_JDK8 = new ConcurrentHashMap<>();
+        SHARED_MAP_JDK17 = new ConcurrentHashMap<>();
+    }
+
+    @AfterAll
+    public static void verifyTestResults() {
+        if (!SHARED_MAP_JDK8.isEmpty()) {
+            assertEquals(REPEATED_TEST_NUM, SHARED_MAP_JDK8.get(TEST_KEY).size());
+        }
+
+        if (!SHARED_MAP_JDK17.isEmpty()) {
+            assertEquals(REPEATED_TEST_NUM, SHARED_MAP_JDK17.get(TEST_KEY).size());
+        }
+
+        SHARED_MAP_JDK8 = null;
+        SHARED_MAP_JDK17 = null;
+    }
+
     @Test
     public void testComputeIfAbsent() {
         ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();
@@ -60,8 +102,27 @@ public void issue11986ForJava17Test() {
         final ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
 
         // JDK9+ has been resolved JDK-8161372 bug, when cause dead then throw IllegalStateException
-        assertThrows(IllegalStateException.class, () -> {
-            ConcurrentHashMapUtils.computeIfAbsent(map, "AaAa", key -> map.computeIfAbsent("BBBB", key2 -> 42));
-        });
+        assertThrows(
+                IllegalStateException.class,
+                () -> ConcurrentHashMapUtils.computeIfAbsent(
+                        map, "AaAa", key -> map.computeIfAbsent("BBBB", key2 -> 42)));
+    }
+
+    @EnabledForJreRange(max = org.junit.jupiter.api.condition.JRE.JAVA_8)
+    @RepeatedTest(value = REPEATED_TEST_NUM)
+    @Execution(ExecutionMode.CONCURRENT)
+    public void threadSafetyOperatorForJava8Test() {
+        List<Long> value = ConcurrentHashMapUtils.computeIfAbsent(
+                SHARED_MAP_JDK8, TEST_KEY, key -> new ArrayList<>(), list -> list.add(System.currentTimeMillis()));
+        assertNotNull(value);
+    }
+
+    @EnabledForJreRange(max = JRE.JAVA_17)
+    @Execution(ExecutionMode.CONCURRENT)
+    @RepeatedTest(value = REPEATED_TEST_NUM)
+    public void threadSafetyOperatorForJava17Test() {
+        List<Long> value = ConcurrentHashMapUtils.computeIfAbsent(
+                SHARED_MAP_JDK17, TEST_KEY, key -> new ArrayList<>(), list -> list.add(System.currentTimeMillis()));
+        assertNotNull(value);
     }
 }

dubbo-common/src/test/resources/junit-platform.properties

@@ -0,0 +1,6 @@
+junit.jupiter.execution.parallel.enabled = true
+junit.jupiter.execution.parallel.mode.default = concurrent
+
+# the maximum pool size can be configured using a ParallelExecutionConfigurationStrategy
+junit.jupiter.execution.parallel.config.strategy=fixed
+junit.jupiter.execution.parallel.config.fixed.parallelism=10

dubbo-config/dubbo-config-spring/src/main/java/org/apache/dubbo/config/spring/reference/ReferenceBeanManager.java

@@ -43,20 +43,21 @@
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.CONFIG_DUBBO_BEAN_INITIALIZER;
 
 public class ReferenceBeanManager implements ApplicationContextAware {
+
     public static final String BEAN_NAME = "dubboReferenceBeanManager";
     private final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(getClass());
 
     // reference key -> reference bean names
-    private ConcurrentMap<String, List<String>> referenceKeyMap = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, List<String>> referenceKeyMap = new ConcurrentHashMap<>();
 
     // reference alias -> reference bean name
-    private ConcurrentMap<String, String> referenceAliasMap = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, String> referenceAliasMap = new ConcurrentHashMap<>();
 
     // reference bean name -> ReferenceBean
-    private ConcurrentMap<String, ReferenceBean> referenceBeanMap = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, ReferenceBean> referenceBeanMap = new ConcurrentHashMap<>();
 
     // reference key -> ReferenceConfig instance
-    private ConcurrentMap<String, ReferenceConfig> referenceConfigMap = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, ReferenceConfig> referenceConfigMap = new ConcurrentHashMap<>();
 
     private ApplicationContext applicationContext;
     private volatile boolean initialized = false;
@@ -74,8 +75,8 @@ public void addReference(ReferenceBean referenceBean) throws Exception {
                     "",
                     "Early initialize reference bean before DubboConfigBeanInitializer,"
                             + " the BeanPostProcessor has not been loaded at this time, which may cause abnormalities in some components (such as seata): "
-                            + referenceBeanName
-                            + " = " + ReferenceBeanSupport.generateReferenceKey(referenceBean, applicationContext));
+                            + referenceBeanName + " = "
+                            + ReferenceBeanSupport.generateReferenceKey(referenceBean, applicationContext));
         }
         String referenceKey = getReferenceKeyByBeanName(referenceBeanName);
         if (StringUtils.isEmpty(referenceKey)) {
@@ -112,13 +113,13 @@ private String getReferenceKeyByBeanName(String referenceBeanName) {
     }
 
     public void registerReferenceKeyAndBeanName(String referenceKey, String referenceBeanNameOrAlias) {
-        List<String> list =
-                ConcurrentHashMapUtils.computeIfAbsent(referenceKeyMap, referenceKey, (key) -> new ArrayList<>());
-        if (!list.contains(referenceBeanNameOrAlias)) {
-            list.add(referenceBeanNameOrAlias);
-            // register bean name as alias
-            referenceAliasMap.put(referenceBeanNameOrAlias, list.get(0));
-        }
+        ConcurrentHashMapUtils.computeIfAbsent(referenceKeyMap, referenceKey, (key) -> new ArrayList<>(), list -> {
+            if (!list.contains(referenceBeanNameOrAlias)) {
+                list.add(referenceBeanNameOrAlias);
+                // register bean name as alias
+                referenceAliasMap.put(referenceBeanNameOrAlias, list.get(0));
+            }
+        });
     }
 
     public ReferenceBean getById(String referenceBeanNameOrAlias) {

pom.xml

@@ -170,6 +170,10 @@
     <protobuf-protoc_version>3.22.3</protobuf-protoc_version>
     <grpc_version>1.54.0</grpc_version>
     <spotless-maven-plugin.version>2.44.3</spotless-maven-plugin.version>
+    <!-- use a lower version of this dependency to override the version
+     defined by spotless-maven-plugin.version to support JDK
+     versions ranging from 1.8 to 11. -->
+    <spotless-maven-plugin.lower.version>2.22.0</spotless-maven-plugin.lower.version>
     <dubbo-shared-resources.version>1.0.0</dubbo-shared-resources.version>
     <palantirJavaFormat.version>2.38.0</palantirJavaFormat.version>
     <revision>3.3.4-SNAPSHOT</revision>
@@ -813,12 +817,13 @@
     </profile>
 
     <profile>
-      <id>jdk9-jdk11-spotless</id>
+      <id>jdk8-jdk11-spotless</id>
       <activation>
         <jdk>[1.8, 11)</jdk>
       </activation>
       <properties>
         <palantirJavaFormat.version>1.1.0</palantirJavaFormat.version>
+        <spotless-maven-plugin.version>${spotless-maven-plugin.lower.version}</spotless-maven-plugin.version>
       </properties>
     </profile>