refactor: optimize Configuration Cache (apache#6420)

changes/en-us/2.x.md

@@ -113,6 +113,7 @@ Add changes here for all PR submitted to the 2.x branch.
 - [[#6387](https://github.com/apache/incubator-seata/pull/6387)] optimize tcc use compatible
 - [[#6402](https://github.com/apache/incubator-seata/pull/6402)] optimize rm-datasource use compatible
 - [[#6419](https://github.com/apache/incubator-seata/pull/6419)] optimize integration-tx-api compatible
+- [[#6405](https://github.com/apache/incubator-seata/pull/6405)] fix kotlin compile failure
 - [[#6412](https://github.com/apache/incubator-seata/pull/6412)] optimize core compatible module
 - [[#6429](https://github.com/apache/incubator-seata/pull/6429)] remove repetitive words
 
@@ -140,6 +141,7 @@ Add changes here for all PR submitted to the 2.x branch.
 
 - [[#6280](https://github.com/apache/incubator-seata/pull/6280)] refactor Saga designer using diagram-js
 - [[#6269](https://github.com/apache/incubator-seata/pull/6269)] standardize Seata Exception
+- [[#6420](https://github.com/apache/incubator-seata/pull/6420)] refactor Configuration Cache
 
 Thanks to these contributors for their code commits. Please report an unintended omission.
 

changes/zh-cn/2.x.md

@@ -115,10 +115,10 @@
 - [[#6387](https://github.com/apache/incubator-seata/pull/6387)] 优化tcc使用兼容
 - [[#6402](https://github.com/apache/incubator-seata/pull/6402)] 优化rm-datasource向下兼容
 - [[#6419](https://github.com/apache/incubator-seata/pull/6419)] 优化integration-tx-api向下兼容
+- [[#6405](https://github.com/apache/incubator-seata/pull/6405)] 修复 kotlin 编译失败
 - [[#6412](https://github.com/apache/incubator-seata/pull/6412)] 优化 core 兼容模块
 - [[#6429](https://github.com/apache/incubator-seata/pull/6429)] 移除重复注释
 
-
 ### security:
 - [[#6069](https://github.com/apache/incubator-seata/pull/6069)] 升级Guava依赖版本，修复安全漏洞
 - [[#6144](https://github.com/apache/incubator-seata/pull/6144)] 升级Nacos依赖版本至1.4.6
@@ -141,6 +141,7 @@
 
 - [[#6280](https://github.com/apache/incubator-seata/pull/6280)] 使用diagram-js重构Saga设计器
 - [[#6269](https://github.com/apache/incubator-seata/pull/6269)] 统一Seata异常规范
+- [[#6420](https://github.com/apache/incubator-seata/pull/6420)] 优化配置缓存
 
 非常感谢以下 contributors 的代码贡献。若有无意遗漏，请报告。
 

config/seata-config-consul/src/main/java/org/apache/seata/config/consul/ConsulConfiguration.java

@@ -20,9 +20,9 @@
 import java.net.InetSocketAddress;
 import java.util.Enumeration;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Properties;
 import java.util.Set;
-import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
@@ -346,15 +346,15 @@ public void onChangeEvent(ConfigurationChangeEvent event) {
                                 for (ConfigurationChangeListener changeListener : entry.getValue()) {
                                     event.setDataId(key).setNewValue(valueNew);
                                     ConfigurationChangeListener listener = ((ConsulListener) changeListener).getTargetListener();
-                                    listener.onChangeEvent(event);
+                                    listener.onProcessEvent(event);
                                 }
                             }
                         }
                         seataConfig = seataConfigNew;
                     } else {
                         // The old config change listener,it would be deleted in next edition
                         event.setDataId(dataId).setNewValue(value);
-                        listener.onChangeEvent(event);
+                        listener.onProcessEvent(event);
                     }
                 }
             }

config/seata-config-core/src/main/java/org/apache/seata/config/CachedConfigurationChangeListener.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.seata.config;
+
+public interface CachedConfigurationChangeListener extends ConfigurationChangeListener {
+
+    ConfigurationCache CONFIGURATION_CACHE = ConfigurationCache.getInstance();
+
+    @Override
+    default void afterEvent(ConfigurationChangeEvent event) {
+        ConfigurationChangeListener listener = (ConfigurationChangeListener)CONFIGURATION_CACHE;
+        listener.onProcessEvent(event);
+    }
+}

config/seata-config-core/src/main/java/org/apache/seata/config/ConfigurationCache.java

@@ -21,8 +21,9 @@
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
-import org.apache.seata.common.util.CollectionUtils;
+
 import org.apache.seata.common.util.DurationUtil;
 import org.apache.seata.common.util.StringUtils;
 
@@ -37,45 +38,7 @@ public class ConfigurationCache implements ConfigurationChangeListener {
 
     private static final Map<String, ObjectWrapper> CONFIG_CACHE = new ConcurrentHashMap<>();
 
-    private Map<String, HashSet<ConfigurationChangeListener>> configListenersMap = new HashMap<>();
-
-    public static void addConfigListener(String dataId, ConfigurationChangeListener... listeners) {
-        if (StringUtils.isBlank(dataId)) {
-            return;
-        }
-        synchronized (ConfigurationCache.class) {
-            HashSet<ConfigurationChangeListener> listenerHashSet =
-                getInstance().configListenersMap.computeIfAbsent(dataId, key -> new HashSet<>());
-            if (!listenerHashSet.contains(getInstance())) {
-                ConfigurationFactory.getInstance().addConfigListener(dataId, getInstance());
-                listenerHashSet.add(getInstance());
-            }
-            if (null != listeners && listeners.length > 0) {
-                for (ConfigurationChangeListener listener : listeners) {
-                    if (!listenerHashSet.contains(listener)) {
-                        listenerHashSet.add(listener);
-                        ConfigurationFactory.getInstance().addConfigListener(dataId, listener);
-                    }
-                }
-            }
-        }
-    }
-
-    public static void removeConfigListener(String dataId, ConfigurationChangeListener... listeners) {
-        if (StringUtils.isBlank(dataId)) {
-            return;
-        }
-        synchronized (ConfigurationCache.class) {
-            final HashSet<ConfigurationChangeListener> listenerSet = getInstance().configListenersMap.get(dataId);
-            if (CollectionUtils.isNotEmpty(listenerSet)) {
-                for (ConfigurationChangeListener listener : listeners) {
-                    if (listenerSet.remove(listener)) {
-                        ConfigurationFactory.getInstance().removeConfigListener(dataId, listener);
-                    }
-                }
-            }
-        }
-    }
+    private static final Set<String> DATA_ID_CACHED = new HashSet<>();
 
     public static ConfigurationCache getInstance() {
         return ConfigurationCacheInstance.INSTANCE;
@@ -91,11 +54,12 @@ public void onChangeEvent(ConfigurationChangeEvent event) {
             } else {
                 Object newValue = new ObjectWrapper(event.getNewValue(), null).convertData(oldWrapper.getType());
                 if (!Objects.equals(oldWrapper.getData(), newValue)) {
-                    CONFIG_CACHE.put(event.getDataId(), new ObjectWrapper(newValue, oldWrapper.getType(),oldWrapper.getLastDefaultValue()));
+                    CONFIG_CACHE.put(event.getDataId(),
+                        new ObjectWrapper(newValue, oldWrapper.getType(), oldWrapper.getLastDefaultValue()));
                 }
             }
         } else {
-            CONFIG_CACHE.remove(event.getDataId());
+            CONFIG_CACHE.remove(event.getDataId(), oldWrapper);
         }
     }
 
@@ -115,6 +79,9 @@ public Configuration proxy(Configuration originalConfiguration) throws Exception
                     }
                     if (null == wrapper
                             || (null != defaultValue && !Objects.equals(defaultValue, wrapper.lastDefaultValue))) {
+                        if (DATA_ID_CACHED.add(rawDataId)) {
+                            originalConfiguration.addConfigListener(rawDataId, this);
+                        }
                         Object result = method.invoke(originalConfiguration, args);
                         // The wrapper.data only exists in the cache when it is not null.
                         if (result != null) {

config/seata-config-core/src/main/java/org/apache/seata/config/ConfigurationChangeListener.java

@@ -58,9 +58,9 @@ public interface ConfigurationChangeListener {
      */
     default void onProcessEvent(ConfigurationChangeEvent event) {
         getExecutorService().submit(() -> {
-            beforeEvent();
+            beforeEvent(event);
             onChangeEvent(event);
-            afterEvent();
+            afterEvent(event);
         });
     }
 
@@ -83,14 +83,14 @@ default ExecutorService getExecutorService() {
     /**
      * Before event.
      */
-    default void beforeEvent() {
+    default void beforeEvent(ConfigurationChangeEvent event) {
 
     }
 
     /**
      * After event.
      */
-    default void afterEvent() {
+    default void afterEvent(ConfigurationChangeEvent event) {
 
     }
 }

config/seata-config-core/src/main/java/org/apache/seata/config/FileConfiguration.java

@@ -394,7 +394,7 @@ public void onChangeEvent(ConfigurationChangeEvent event) {
                                 event.setDataId(dataId).setNewValue(currentConfig).setOldValue(oldConfig);
 
                                 for (ConfigurationChangeListener listener : dataIdMap.get(dataId)) {
-                                    listener.onChangeEvent(event);
+                                    listener.onProcessEvent(event);
                                 }
                             }
                         }

config/seata-config-core/src/test/java/org/apache/seata/config/ConfigurationCacheTests.java

@@ -16,17 +16,12 @@
  */
 package org.apache.seata.config;
 
-import org.apache.seata.common.util.CollectionUtils;
+import java.time.Duration;
+
 import org.apache.seata.common.util.DurationUtil;
-import org.apache.seata.common.util.ReflectionUtil;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
-import java.lang.reflect.Field;
-import java.time.Duration;
-import java.util.HashSet;
-import java.util.Map;
-
 
 public class ConfigurationCacheTests {
 
@@ -70,37 +65,6 @@ public void testChangeValue() throws Exception {
         Assertions.assertNull(test);
     }
 
-    // FIXME: 2023/2/19 wait bugfix
-    // @Test
-    public void testConfigListener() throws Exception {
-        Configuration configuration = new FileConfiguration("registry");
-        configuration = ConfigurationCache.getInstance().proxy(configuration);
-
-        // get config listeners map
-        Field configListenersMapField = ReflectionUtil.getField(ConfigurationCache.class, "configListenersMap");
-        Map<String, HashSet<ConfigurationChangeListener>> configListenersMap = (Map<String,
-            HashSet<ConfigurationChangeListener>>)configListenersMapField.get(ConfigurationCache.getInstance());
-
-        boolean value = configuration.getBoolean("service.disableGlobalTransaction");
-        TestListener listener = new TestListener();
-        ConfigurationCache.addConfigListener("service.disableGlobalTransaction", listener);
-        // check listener if exist
-        HashSet<ConfigurationChangeListener> listeners = configListenersMap.get("service.disableGlobalTransaction");
-        Assertions.assertTrue(CollectionUtils.isNotEmpty(listeners));
-        // change value,trigger listener
-        System.setProperty("service.disableGlobalTransaction", String.valueOf(!value));
-        // remove null
-        ConfigurationCache.removeConfigListener(null);
-        // check listener if exist
-        listeners = configListenersMap.get("service.disableGlobalTransaction");
-        Assertions.assertTrue(CollectionUtils.isNotEmpty(listeners));
-        // remove listener
-        ConfigurationCache.removeConfigListener("service.disableGlobalTransaction", listener);
-        // check listener if exist
-        listeners = configListenersMap.get("service.disableGlobalTransaction");
-        // is empty
-        Assertions.assertTrue(CollectionUtils.isEmpty(listeners));
-    }
 
     public static class TestListener implements ConfigurationChangeListener {
 

config/seata-config-core/src/test/java/org/apache/seata/config/FileConfigurationTest.java

@@ -18,6 +18,7 @@
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
@@ -40,19 +41,27 @@ void tearDown() {
     void addConfigListener() throws InterruptedException {
         Configuration fileConfig = ConfigurationFactory.getInstance();
         CountDownLatch countDownLatch = new CountDownLatch(1);
-        boolean value = fileConfig.getBoolean("service.disableGlobalTransaction");
-        ConfigurationCache.addConfigListener("service.disableGlobalTransaction", (event) -> {
-            Assertions.assertEquals(Boolean.parseBoolean(event.getNewValue()), !Boolean.parseBoolean(event.getOldValue()));
-            countDownLatch.countDown();
+        String dataId = "service.disableGlobalTransaction";
+        boolean value = fileConfig.getBoolean(dataId);
+        fileConfig.addConfigListener(dataId, new CachedConfigurationChangeListener() {
+            @Override
+            public void onChangeEvent(ConfigurationChangeEvent event) {
+                Assertions.assertEquals(Boolean.parseBoolean(event.getNewValue()),
+                    !Boolean.parseBoolean(event.getOldValue()));
+                countDownLatch.countDown();
+            }
         });
-        System.setProperty("service.disableGlobalTransaction", String.valueOf(!value));
-        countDownLatch.await(5, TimeUnit.SECONDS);
+        System.setProperty(dataId, String.valueOf(!value));
+        countDownLatch.await(2, TimeUnit.SECONDS);
         System.setProperty("file.listener.enabled", "false");
-        System.setProperty("service.disableGlobalTransaction", String.valueOf(value));
-        Thread.sleep(2000);
-        boolean currentValue = fileConfig.getBoolean("service.disableGlobalTransaction");
+        //wait for loop safety, loop time is LISTENER_CONFIG_INTERVAL=1s
+        Thread.sleep(1500);
+        System.setProperty(dataId, String.valueOf(value));
+        //sleep for a period of time to simulate waiting for a cache refresh.Actually, it doesn't trigger.
+        Thread.sleep(1000);
+        boolean currentValue = fileConfig.getBoolean(dataId);
         Assertions.assertNotEquals(value, currentValue);
-        System.setProperty("service.disableGlobalTransaction", String.valueOf(!value));
+        System.setProperty(dataId, String.valueOf(!value));
     }
 
     @Test

config/seata-config-core/src/test/java/org/apache/seata/config/ProConfigurationFactoryTest.java

@@ -33,7 +33,7 @@ void getInstance() {
         Assertions.assertNull(ConfigurationFactory.CURRENT_FILE_INSTANCE.getConfig("config.file.testNull"));
         Assertions.assertNull(ConfigurationFactory.CURRENT_FILE_INSTANCE.getConfig("config.file.testExist"));
         Configuration instance = ConfigurationFactory.getInstance();
-        Assertions.assertEquals(instance.getConfig("service.disableGlobalTransaction"), "true");
+        Assertions.assertEquals(instance.getConfig("client.undo.compress.enable"), "true");
         Assertions.assertEquals(instance.getConfig("service.default.grouplist"), "127.0.0.1:8092");
 
     }

config/seata-config-core/src/test/java/org/apache/seata/config/RegistryConfigurationFactoryTest.java

@@ -30,7 +30,6 @@ void getInstance() {
         ConfigurationFactory.reload();
         Assertions.assertEquals(ConfigurationFactory.CURRENT_FILE_INSTANCE.getConfig("config.file.name"),"file-test.conf");
         Configuration instance = ConfigurationFactory.getInstance();
-        Assertions.assertEquals(instance.getConfig("service.disableGlobalTransaction"),"true");
         Assertions.assertEquals(instance.getConfig("service.default.grouplist"), "127.0.0.1:8091");
 
     }

config/seata-config-core/src/test/java/org/apache/seata/config/YamlConfigurationFactoryTest.java

@@ -33,7 +33,7 @@ public void getInstance() {
         Assertions.assertNull(ConfigurationFactory.CURRENT_FILE_INSTANCE.getConfig("config.file.testNull"));
         Assertions.assertNull(ConfigurationFactory.CURRENT_FILE_INSTANCE.getConfig("config.file.testExist"));
         Configuration instance = ConfigurationFactory.getInstance();
-        Assertions.assertEquals(instance.getConfig("service.disableGlobalTransaction"), "true");
+        Assertions.assertEquals(instance.getConfig("transport.heartbeat"), "true");
         Assertions.assertEquals(instance.getConfig("service.default.grouplist"), "127.0.0.1:8093");
     }
 

config/seata-config-core/src/test/resources/file-test-pro.conf

@@ -22,4 +22,7 @@ service {
   default.grouplist = "127.0.0.1:8092"
   #disable seata
   disableGlobalTransaction = true
+}
+client {
+   undo.compress.enable=true
 }

config/seata-config-core/src/test/resources/file-test-yaml.conf

@@ -22,4 +22,8 @@ service {
   default.grouplist = "127.0.0.1:8093"
   #disable seata
   disableGlobalTransaction = true
+}
+
+transport {
+  heartbeat = true
 }

config/seata-config-custom/src/test/java/org/apache/seata/config/CustomConfigurationForTest.java

@@ -60,7 +60,6 @@ public boolean removeConfig(String dataId, long timeoutMills) {
 
     @Override
     public void addConfigListener(String dataId, ConfigurationChangeListener listener) {
-        throw new UnsupportedOperationException();
     }
 
     @Override

config/seata-config-etcd3/src/main/java/org/apache/seata/config/etcd3/EtcdConfiguration.java

@@ -394,7 +394,7 @@ public void onNext(WatchResponse watchResponse) {
                         List<KeyValue> keyValues = getResponse.getKvs();
                         if (CollectionUtils.isNotEmpty(keyValues)) {
                             event.setDataId(dataId).setNewValue(keyValues.get(0).getValue().toString(UTF_8));
-                            listener.onChangeEvent(event);
+                            listener.onProcessEvent(event);
                         }
                     } catch (Exception e) {
                         LOGGER.error("error occurred while getting value{}", e.getMessage(), e);

core/src/main/java/org/apache/seata/core/rpc/netty/NettyClientConfig.java

@@ -51,7 +51,7 @@ public class NettyClientConfig extends NettyBaseConfig {
     private static final int MAX_CHECK_ALIVE_RETRY = 300;
     private static final int CHECK_ALIVE_INTERVAL = 10;
     private static final String SOCKET_ADDRESS_START_CHAR = "/";
-    private static final long MAX_ACQUIRE_CONN_MILLS = 60 * 1000L;
+    private static final long MAX_ACQUIRE_CONN_MILLS = 10 * 1000L;
     private static final String RPC_DISPATCH_THREAD_PREFIX = "rpcDispatch";
     private static final int DEFAULT_MAX_POOL_ACTIVE = 1;
     private static final int DEFAULT_MIN_POOL_IDLE = 0;