Merge pull request #4 from seata/develop

bugfix: fix getConfig cache value is 'null'(seata#2904)

README.md

@@ -20,7 +20,7 @@ Naturally, data consistency will be guaranteed by the local transaction.
 
 ![Monolithic App](https://cdn.nlark.com/lark/0/2018/png/18862/1545296770244-4cedf37e-9dc6-4fc0-a97f-f4240b9d8640.png) 
 
-Things have changed in microservices architecture. The 3 modules mentioned above are designed to be 3 services on top of 3 different data sources ([Pattern: Database per service](http://microservices.io/patterns/data/database-per-service.html)). Data consistency within every single service is naturally guaranteed by the local transaction. 
+Things have changed in a microservices architecture. The 3 modules mentioned above are designed to be 3 services on top of 3 different data sources ([Pattern: Database per service](http://microservices.io/patterns/data/database-per-service.html)). Data consistency within every single service is naturally guaranteed by the local transaction. 
 
 **But how about the whole business logic scope?**
 
@@ -50,7 +50,7 @@ A typical lifecycle of Seata managed distributed transaction:
 
 1. TM asks TC to begin a new global transaction. TC generates an XID representing the global transaction.
 2. XID is propagated through microservices' invoke chain.
-3. RM register local transaction as a branch of the corresponding global transaction of XID to TC. 
+3. RM registers local transaction as a branch of the corresponding global transaction of XID to TC. 
 4. TM asks TC for committing or rollbacking the corresponding global transaction of XID.
 5. TC drives all branch transactions under the corresponding global transaction of XID to finish branch committing or rollbacking.
 
@@ -68,14 +68,14 @@ For more details about principle and design, please go to [Seata wiki page](http
 
 ##### Alibaba
 
-- **TXC**: Taobao Transaction Constructor. Alibaba middleware team start this project since 2014 to meet distributed transaction problem caused by application architecture change from monolithic to microservices.
+- **TXC**: Taobao Transaction Constructor. Alibaba middleware team started this project since 2014 to meet the distributed transaction problems caused by application architecture change from monolithic to microservices.
 - **GTS**: Global Transaction Service. TXC as an Aliyun middleware product with new name GTS was published since 2016.
-- **Fescar**: we start the open source project Fescar based on TXC/GTS since 2019 to work closely with the community in the future.
+- **Fescar**: we started the open source project Fescar based on TXC/GTS since 2019 to work closely with the community in the future.
 
 
 ##### Seata Community
 
-- **Seata** :Simple Extensible Autonomous Transaction Architecture. Ant Financial joins Fescar, which make it to be a more neutral and open community for distributed transaction，and Fescar be renamed to Seata.
+- **Seata** :Simple Extensible Autonomous Transaction Architecture. Ant Financial joins Fescar, which make it to be a more neutral and open community for distributed transaction, and Fescar be renamed to Seata.
 
 
 

bom/pom.xml

@@ -83,7 +83,7 @@
         <curator-test.version>2.9.1</curator-test.version>
         <spring-context-support.version>1.0.2</spring-context-support.version>
         <jacoco-maven-plugin.version>0.8.3</jacoco-maven-plugin.version>
-        <apollo-client.version>1.1.0</apollo-client.version>
+        <apollo-client.version>1.6.0</apollo-client.version>
         <redis-clients.version>3.2.0</redis-clients.version>
         <mock-jedis.version>0.1.16</mock-jedis.version>
         <eureka-clients.version>1.9.5</eureka-clients.version>

core/src/main/java/io/seata/core/constants/DefaultValues.java → common/src/main/java/io/seata/common/DefaultValues.java

@@ -13,7 +13,7 @@
  *  See the License for the specific language governing permissions and
  *  limitations under the License.
  */
-package io.seata.core.constants;
+package io.seata.common;
 
 import java.util.concurrent.ThreadLocalRandom;
 
@@ -80,6 +80,8 @@ public class DefaultValues {
     public static final String DEFAULT_TC_CLUSTER = "default";
     public static final String DEFAULT_GROUPLIST = "127.0.0.1:8091";
 
+    public static final String DEFAULT_DATA_SOURCE_PROXY_MODE = "AT";
+
     public static final boolean DEFAULT_DISABLE_GLOBAL_TRANSACTION = false;
 
     public static final int SERVER_DEFAULT_PORT = 8091;
@@ -89,4 +91,7 @@ public class DefaultValues {
     public static final String DEFAULT_SAGA_JSON_PARSER = "fastjson";
 
     public static final boolean DEFAULT_SERVER_ENABLE_CHECK_AUTH = true;
+
+    public static final String DEFAULT_LOAD_BALANCE = "RandomLoadBalance";
+    public static final int VIRTUAL_NODES_DEFAULT = 10;
 }

common/src/main/java/io/seata/common/util/IdWorker.java

@@ -94,25 +94,24 @@ public IdWorker(long workerId) {
      *
      * @return SnowflakeId
      */
-    public long nextId() {
+    public synchronized long nextId() {
         long timestamp = timeGen();
 
         if (timestamp < lastTimestamp) {
             throw new RuntimeException(String.format(
                 "clock moved backwards.  Refusing to generate id for %d milliseconds", lastTimestamp - timestamp));
         }
 
-        synchronized (this) {
-            if (lastTimestamp == timestamp) {
-                sequence = (sequence + 1) & sequenceMask;
-                if (sequence == 0) {
-                    timestamp = tilNextMillis(lastTimestamp);
-                }
-            } else {
-                sequence = 0L;
+        if (lastTimestamp == timestamp) {
+            sequence = (sequence + 1) & sequenceMask;
+            if (sequence == 0) {
+                timestamp = tilNextMillis(lastTimestamp);
             }
-            lastTimestamp = timestamp;
+        } else {
+            sequence = 0L;
         }
+        lastTimestamp = timestamp;
+
         return ((timestamp - twepoch) << timestampLeftShift) | (workerId << workerIdShift) | sequence;
     }
 

config/seata-config-apollo/src/main/java/io/seata/config/apollo/ApolloConfiguration.java

@@ -31,6 +31,7 @@
 import io.netty.util.internal.ConcurrentSet;
 import io.seata.common.exception.NotSupportYetException;
 import io.seata.common.thread.NamedThreadFactory;
+import io.seata.common.util.StringUtils;
 import io.seata.config.AbstractConfiguration;
 import io.seata.config.ConfigFuture;
 import io.seata.config.Configuration;
@@ -52,8 +53,10 @@ public class ApolloConfiguration extends AbstractConfiguration {
     private static final String REGISTRY_TYPE = "apollo";
     private static final String APP_ID = "appId";
     private static final String APOLLO_META = "apolloMeta";
+    private static final String APOLLO_SECRET = "apolloAccesskeySecret";
     private static final String PROP_APP_ID = "app.id";
     private static final String PROP_APOLLO_META = "apollo.meta";
+    private static final String PROP_APOLLO_SECRET = "apollo.accesskey.secret";
     private static final String NAMESPACE = "namespace";
     private static final String DEFAULT_NAMESPACE = "application";
     private static final Configuration FILE_CONFIG = ConfigurationFactory.CURRENT_FILE_INSTANCE;
@@ -75,7 +78,7 @@ private ApolloConfiguration() {
                         MAX_CONFIG_OPERATE_THREAD, Integer.MAX_VALUE, TimeUnit.MILLISECONDS,
                         new LinkedBlockingQueue<>(),
                         new NamedThreadFactory("apolloConfigOperate", MAX_CONFIG_OPERATE_THREAD));
-                    config.addChangeListener((changeEvent) -> {
+                    config.addChangeListener(changeEvent -> {
                         for (String key : changeEvent.changedKeys()) {
                             if (!LISTENER_SERVICE_MAP.containsKey(key)) {
                                 continue;
@@ -167,6 +170,12 @@ private void readyApolloConfig() {
         if (!properties.containsKey(PROP_APOLLO_META)) {
             System.setProperty(PROP_APOLLO_META, FILE_CONFIG.getConfig(getApolloMetaFileKey()));
         }
+        if (!properties.containsKey(PROP_APOLLO_SECRET)) {
+            String secretKey = FILE_CONFIG.getConfig(getApolloSecretFileKey());
+            if (!StringUtils.isBlank(secretKey)) {
+                System.setProperty(PROP_APOLLO_SECRET, secretKey);
+            }
+        }
     }
 
     @Override
@@ -178,6 +187,10 @@ private static String getApolloMetaFileKey() {
         return String.join(FILE_CONFIG_SPLIT_CHAR, FILE_ROOT_CONFIG, REGISTRY_TYPE, APOLLO_META);
     }
 
+    private static String getApolloSecretFileKey() {
+        return String.join(FILE_CONFIG_SPLIT_CHAR, FILE_ROOT_CONFIG, REGISTRY_TYPE, APOLLO_SECRET);
+    }
+
     private static String getApolloAppIdFileKey() {
         return String.join(FILE_CONFIG_SPLIT_CHAR, FILE_ROOT_CONFIG, REGISTRY_TYPE, APP_ID);
     }

config/seata-config-core/src/main/java/io/seata/config/ConfigurationCache.java

@@ -88,7 +88,7 @@ public Configuration proxy(Configuration originalConfiguration) {
                             CONFIG_CACHE.put(rawDataId, result);
                         }
                     }
-                    if (method.getReturnType().equals(String.class)) {
+                    if (null != result && method.getReturnType().equals(String.class)) {
                         return String.valueOf(result);
                     }
                     return result;

config/seata-config-core/src/main/resources/registry.conf

@@ -1,6 +1,8 @@
 registry {
   # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
   type = "file"
+  loadBalance = "RandomLoadBalance"
+  loadBalanceVirtualNodes = 10
 
   nacos {
     application = "seata-server"

config/seata-config-core/src/test/resources/registry.conf

@@ -1,6 +1,8 @@
 registry {
   # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
   type = "file"
+  loadBalance = "RandomLoadBalance"
+  loadBalanceVirtualNodes = 10
 
   nacos {
     serverAddr = "localhost"

core/src/main/java/io/seata/core/constants/ConfigurationKeys.java

@@ -508,4 +508,9 @@ public class ConfigurationKeys {
      * The constant TX_SERVICE_GROUP.
      */
     public static final String TX_SERVICE_GROUP = "txServiceGroup";
+
+    /**
+     * The constant DATA_SOURCE_PROXY_MODE.
+     */
+    public static final String DATA_SOURCE_PROXY_MODE = "dataSourceProxyMode";
 }

core/src/main/java/io/seata/core/context/RootContext.java

@@ -19,6 +19,9 @@
 
 import io.seata.common.exception.ShouldNeverHappenException;
 import io.seata.common.util.StringUtils;
+import io.seata.config.ConfigurationFactory;
+import io.seata.core.constants.ConfigurationKeys;
+import io.seata.common.DefaultValues;
 import io.seata.core.model.BranchType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -51,6 +54,9 @@ private RootContext() {
 
     private static ContextCore CONTEXT_HOLDER = ContextCoreLoader.load();
 
+    private static final String DATA_SOURCE_PROXY_MODE = ConfigurationFactory.getInstance()
+            .getConfig(ConfigurationKeys.DATA_SOURCE_PROXY_MODE, DefaultValues.DEFAULT_DATA_SOURCE_PROXY_MODE);
+
     /**
      * Gets xid.
      *
@@ -129,8 +135,8 @@ public static String getBranchType() {
             if (StringUtils.isNotBlank(branchType)) {
                 return branchType;
             }
-            //default branchType is AT
-            return BranchType.AT.name();
+            //default branchType is the dataSourceProxyMode
+            return BranchType.XA.name().equalsIgnoreCase(DATA_SOURCE_PROXY_MODE) ? BranchType.XA.name() : BranchType.AT.name();
         }
         return null;
     }

core/src/main/java/io/seata/core/context/ThreadLocalContextCore.java

@@ -17,7 +17,6 @@
 
 import java.util.HashMap;
 import java.util.Map;
-
 import io.seata.common.loader.LoadLevel;
 
 /**
@@ -28,13 +27,7 @@
 @LoadLevel(name = "ThreadLocalContextCore", order = Integer.MIN_VALUE)
 public class ThreadLocalContextCore implements ContextCore {
 
-    private ThreadLocal<Map<String, String>> threadLocal = new ThreadLocal<Map<String, String>>() {
-        @Override
-        protected Map<String, String> initialValue() {
-            return new HashMap<String, String>();
-        }
-
-    };
+    private ThreadLocal<Map<String, String>> threadLocal = ThreadLocal.withInitial(() -> new HashMap<>());
 
     @Override
     public String put(String key, String value) {

core/src/main/java/io/seata/core/logger/StackTraceLogger.java

@@ -21,7 +21,7 @@
 import org.slf4j.Logger;
 import java.util.concurrent.ThreadLocalRandom;
 
-import static io.seata.core.constants.DefaultValues.DEFAULT_LOG_EXCEPTION_RATE;
+import static io.seata.common.DefaultValues.DEFAULT_LOG_EXCEPTION_RATE;
 
 /**
  * @author jsbxyyx

core/src/main/java/io/seata/core/rpc/TransportProtocolType.java

@@ -26,10 +26,6 @@ public enum TransportProtocolType {
      */
     TCP("tcp"),
 
-    /**
-     * Udt transport protocol type.
-     */
-    UDT("udt"),
     /**
      * Unix domain socket transport protocol type.
      */
@@ -43,4 +39,20 @@ public enum TransportProtocolType {
     TransportProtocolType(String name) {
         this.name = name;
     }
+
+    /**
+     * Gets type.
+     *
+     * @param name the name
+     * @return the type
+     */
+    public static TransportProtocolType getType(String name) {
+        name = name.trim().replace('-', '_');
+        for (TransportProtocolType b : TransportProtocolType.values()) {
+            if (b.name().equalsIgnoreCase(name)) {
+                return b;
+            }
+        }
+        throw new IllegalArgumentException("unknown type:" + name);
+    }
 }

core/src/main/java/io/seata/core/rpc/TransportServerType.java

@@ -38,4 +38,19 @@ public enum TransportServerType {
     TransportServerType(String name) {
         this.name = name;
     }
+
+    /**
+     * Gets type.
+     *
+     * @param name the name
+     * @return the type
+     */
+    public static TransportServerType getType(String name) {
+        for (TransportServerType b : TransportServerType.values()) {
+            if (b.name().equalsIgnoreCase(name)) {
+                return b;
+            }
+        }
+        throw new IllegalArgumentException("unknown type:" + name);
+    }
 }