Merge pull request #1364 from mercyblitz/finchley

[Feature] Synchronizing code to Spring Cloud Alibaba finchley branch

spring-cloud-alibaba-dependencies/pom.xml

@@ -89,6 +89,12 @@
                 <version>${sentinel.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>com.alibaba.csp</groupId>
+                <artifactId>sentinel-datasource-consul</artifactId>
+                <version>${sentinel.version}</version>
+            </dependency>
+
             <dependency>
                 <groupId>com.alibaba.csp</groupId>
                 <artifactId>sentinel-web-servlet</artifactId>

spring-cloud-alibaba-starters/spring-cloud-alibaba-sentinel-datasource/pom.xml

@@ -83,6 +83,12 @@
             <optional>true</optional>
         </dependency>
 
+        <dependency>
+            <groupId>com.alibaba.csp</groupId>
+            <artifactId>sentinel-datasource-consul</artifactId>
+            <optional>true</optional>
+        </dependency>
+
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>

spring-cloud-alibaba-starters/spring-cloud-alibaba-sentinel-datasource/src/main/java/com/alibaba/cloud/sentinel/datasource/config/ConsulDataSourceProperties.java

@@ -0,0 +1,97 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 com.alibaba.cloud.sentinel.datasource.config;
+
+import com.alibaba.cloud.sentinel.datasource.factorybean.ConsulDataSourceFactoryBean;
+
+import org.springframework.util.StringUtils;
+
+/**
+ * Consul Properties class Using by {@link DataSourcePropertiesConfiguration} and
+ * {@link ConsulDataSourceFactoryBean}.
+ *
+ * @author <a href="mailto:mengjindc@gmail.com">mengjin</a>
+ */
+public class ConsulDataSourceProperties extends AbstractDataSourceProperties {
+
+	public ConsulDataSourceProperties() {
+		super(ConsulDataSourceFactoryBean.class.getName());
+	}
+
+	/**
+	 * consul server host.
+	 */
+	private String host;
+
+	/**
+	 * consul server port.
+	 */
+	private int port = 8500;
+
+	/**
+	 * data key in Redis.
+	 */
+	private String ruleKey;
+
+	/**
+	 * Request of query will hang until timeout (in second) or get updated value.
+	 */
+	private int waitTimeoutInSecond = 1;
+
+	@Override
+	public void preCheck(String dataSourceName) {
+		if (StringUtils.isEmpty(host)) {
+			throw new IllegalArgumentException("ConsulDataSource server-host is empty");
+		}
+		if (StringUtils.isEmpty(ruleKey)) {
+			throw new IllegalArgumentException(
+					"ConsulDataSource ruleKey can not be empty");
+		}
+	}
+
+	public String getHost() {
+		return host;
+	}
+
+	public void setHost(String host) {
+		this.host = host;
+	}
+
+	public int getPort() {
+		return port;
+	}
+
+	public void setPort(int port) {
+		this.port = port;
+	}
+
+	public String getRuleKey() {
+		return ruleKey;
+	}
+
+	public void setRuleKey(String ruleKey) {
+		this.ruleKey = ruleKey;
+	}
+
+	public int getWaitTimeoutInSecond() {
+		return waitTimeoutInSecond;
+	}
+
+	public void setWaitTimeoutInSecond(int waitTimeoutInSecond) {
+		this.waitTimeoutInSecond = waitTimeoutInSecond;
+	}
+}

spring-cloud-alibaba-starters/spring-cloud-alibaba-sentinel-datasource/src/main/java/com/alibaba/cloud/sentinel/datasource/config/DataSourcePropertiesConfiguration.java

@@ -34,6 +34,7 @@ import org.springframework.util.ObjectUtils;
  * @see ZookeeperDataSourceProperties
  * @see FileDataSourceProperties
  * @see RedisDataSourceProperties
+ * @see ConsulDataSourceProperties
  */
 public class DataSourcePropertiesConfiguration {
 
@@ -47,9 +48,23 @@ public class DataSourcePropertiesConfiguration {
 
 	private RedisDataSourceProperties redis;
 
+	private ConsulDataSourceProperties consul;
+
 	public DataSourcePropertiesConfiguration() {
 	}
 
+	public DataSourcePropertiesConfiguration(ConsulDataSourceProperties consul) {
+		this.consul = consul;
+	}
+
+	public ConsulDataSourceProperties getConsul() {
+		return consul;
+	}
+
+	public void setConsul(ConsulDataSourceProperties consul) {
+		this.consul = consul;
+	}
+
 	public DataSourcePropertiesConfiguration(FileDataSourceProperties file) {
 		this.file = file;
 	}

spring-cloud-alibaba-starters/spring-cloud-alibaba-sentinel-datasource/src/main/java/com/alibaba/cloud/sentinel/datasource/config/RedisDataSourceProperties.java

@@ -24,7 +24,7 @@ import com.alibaba.cloud.sentinel.datasource.factorybean.RedisDataSourceFactoryB
 import org.springframework.util.StringUtils;
 
 /**
- * Zookeeper Properties class Using by {@link DataSourcePropertiesConfiguration} and
+ * Redis Properties class Using by {@link DataSourcePropertiesConfiguration} and
  * {@link RedisDataSourceFactoryBean}.
  *
  * @author <a href="mailto:wangiegie@gmail.com">lengleng</a>

spring-cloud-alibaba-starters/spring-cloud-alibaba-sentinel-datasource/src/main/java/com/alibaba/cloud/sentinel/datasource/factorybean/ConsulDataSourceFactoryBean.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 com.alibaba.cloud.sentinel.datasource.factorybean;
+
+import com.alibaba.csp.sentinel.datasource.Converter;
+import com.alibaba.csp.sentinel.datasource.consul.ConsulDataSource;
+
+import org.springframework.beans.factory.FactoryBean;
+
+/**
+ * A {@link FactoryBean} for creating {@link ConsulDataSource} instance.
+ *
+ * @author <a href="mailto:mengjindc@gmail.com">mengjin</a>
+ * @see ConsulDataSource
+ */
+public class ConsulDataSourceFactoryBean implements FactoryBean<ConsulDataSource> {
+
+	private String host;
+
+	private int port;
+
+	private String ruleKey;
+
+	private int waitTimeoutInSecond;
+
+	private Converter converter;
+
+	@Override
+	public ConsulDataSource getObject() throws Exception {
+		return new ConsulDataSource(host, port, ruleKey, waitTimeoutInSecond, converter);
+	}
+
+	@Override
+	public Class<?> getObjectType() {
+		return ConsulDataSource.class;
+	}
+
+	public String getHost() {
+		return host;
+	}
+
+	public void setHost(String host) {
+		this.host = host;
+	}
+
+	public int getPort() {
+		return port;
+	}
+
+	public void setPort(int port) {
+		this.port = port;
+	}
+
+	public String getRuleKey() {
+		return ruleKey;
+	}
+
+	public void setRuleKey(String ruleKey) {
+		this.ruleKey = ruleKey;
+	}
+
+	public int getWaitTimeoutInSecond() {
+		return waitTimeoutInSecond;
+	}
+
+	public void setWaitTimeoutInSecond(int waitTimeoutInSecond) {
+		this.waitTimeoutInSecond = waitTimeoutInSecond;
+	}
+
+	public Converter getConverter() {
+		return converter;
+	}
+
+	public void setConverter(Converter converter) {
+		this.converter = converter;
+	}
+}

spring-cloud-alibaba-starters/spring-cloud-alibaba-sentinel-datasource/src/main/resources/META-INF/sentinel-datasource.properties

@@ -3,3 +3,4 @@ file =com.alibaba.csp.sentinel.datasource.FileRefreshableDataSource
 apollo = com.alibaba.csp.sentinel.datasource.apollo.ApolloDataSource
 zk = com.alibaba.csp.sentinel.datasource.zookeeper.ZookeeperDataSource
 redis = com.alibaba.csp.sentinel.datasource.redis.RedisDataSource
+consul = com.alibaba.csp.sentinel.datasource.consul.ConsulDataSource

spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/src/main/java/com/alibaba/cloud/seata/feign/hystrix/SeataHystrixConcurrencyStrategy.java

@@ -16,23 +16,104 @@
 
 package com.alibaba.cloud.seata.feign.hystrix;
 
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
+import com.netflix.hystrix.HystrixThreadPoolKey;
+import com.netflix.hystrix.HystrixThreadPoolProperties;
 import com.netflix.hystrix.strategy.HystrixPlugins;
 import com.netflix.hystrix.strategy.concurrency.HystrixConcurrencyStrategy;
+import com.netflix.hystrix.strategy.concurrency.HystrixRequestVariable;
+import com.netflix.hystrix.strategy.concurrency.HystrixRequestVariableLifecycle;
+import com.netflix.hystrix.strategy.eventnotifier.HystrixEventNotifier;
+import com.netflix.hystrix.strategy.executionhook.HystrixCommandExecutionHook;
+import com.netflix.hystrix.strategy.metrics.HystrixMetricsPublisher;
+import com.netflix.hystrix.strategy.properties.HystrixPropertiesStrategy;
+import com.netflix.hystrix.strategy.properties.HystrixProperty;
 import io.seata.core.context.RootContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.web.context.request.RequestAttributes;
+import org.springframework.web.context.request.RequestContextHolder;
 
 /**
  * @author xiaojing
  */
 public class SeataHystrixConcurrencyStrategy extends HystrixConcurrencyStrategy {
 
+	private final Logger logger = LoggerFactory
+			.getLogger(SeataHystrixConcurrencyStrategy.class);
 	private HystrixConcurrencyStrategy delegate;
 
 	public SeataHystrixConcurrencyStrategy() {
-		this.delegate = HystrixPlugins.getInstance().getConcurrencyStrategy();
-		HystrixPlugins.reset();
-		HystrixPlugins.getInstance().registerConcurrencyStrategy(this);
+		try {
+			this.delegate = HystrixPlugins.getInstance().getConcurrencyStrategy();
+			if (this.delegate instanceof SeataHystrixConcurrencyStrategy) {
+				return;
+			}
+			HystrixCommandExecutionHook commandExecutionHook = HystrixPlugins
+					.getInstance().getCommandExecutionHook();
+			HystrixEventNotifier eventNotifier = HystrixPlugins.getInstance()
+					.getEventNotifier();
+			HystrixMetricsPublisher metricsPublisher = HystrixPlugins.getInstance()
+					.getMetricsPublisher();
+			HystrixPropertiesStrategy propertiesStrategy = HystrixPlugins.getInstance()
+					.getPropertiesStrategy();
+			logCurrentStateOfHystrixPlugins(eventNotifier, metricsPublisher,
+					propertiesStrategy);
+			HystrixPlugins.reset();
+			HystrixPlugins.getInstance().registerConcurrencyStrategy(this);
+			HystrixPlugins.getInstance()
+					.registerCommandExecutionHook(commandExecutionHook);
+			HystrixPlugins.getInstance().registerEventNotifier(eventNotifier);
+			HystrixPlugins.getInstance().registerMetricsPublisher(metricsPublisher);
+			HystrixPlugins.getInstance().registerPropertiesStrategy(propertiesStrategy);
+		}
+		catch (Exception ex) {
+			logger.error("Failed to register Seata Hystrix Concurrency Strategy", ex);
+		}
+	}
+
+	private void logCurrentStateOfHystrixPlugins(HystrixEventNotifier eventNotifier,
+			HystrixMetricsPublisher metricsPublisher,
+			HystrixPropertiesStrategy propertiesStrategy) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Current Hystrix plugins configuration is ["
+					+ "concurrencyStrategy [" + this.delegate + "]," + "eventNotifier ["
+					+ eventNotifier + "]," + "metricPublisher [" + metricsPublisher + "],"
+					+ "propertiesStrategy [" + propertiesStrategy + "]," + "]");
+			logger.debug("Registering Seata Hystrix Concurrency Strategy.");
+		}
+	}
+
+	@Override
+	public ThreadPoolExecutor getThreadPool(HystrixThreadPoolKey threadPoolKey,
+			HystrixProperty<Integer> corePoolSize,
+			HystrixProperty<Integer> maximumPoolSize,
+			HystrixProperty<Integer> keepAliveTime, TimeUnit unit,
+			BlockingQueue<Runnable> workQueue) {
+		return this.delegate.getThreadPool(threadPoolKey, corePoolSize, maximumPoolSize,
+				keepAliveTime, unit, workQueue);
+	}
+
+	@Override
+	public ThreadPoolExecutor getThreadPool(HystrixThreadPoolKey threadPoolKey,
+			HystrixThreadPoolProperties threadPoolProperties) {
+		return this.delegate.getThreadPool(threadPoolKey, threadPoolProperties);
+	}
+
+	@Override
+	public BlockingQueue<Runnable> getBlockingQueue(int maxQueueSize) {
+		return this.delegate.getBlockingQueue(maxQueueSize);
+	}
+
+	@Override
+	public <T> HystrixRequestVariable<T> getRequestVariable(
+			HystrixRequestVariableLifecycle<T> rv) {
+		return this.delegate.getRequestVariable(rv);
 	}
 
 	@Override
@@ -52,7 +133,8 @@ public class SeataHystrixConcurrencyStrategy extends HystrixConcurrencyStrategy
 			return wrappedCallable;
 		}
 
-		return new SeataContextCallable<>(wrappedCallable);
+		return new SeataContextCallable<>(wrappedCallable,
+				RequestContextHolder.getRequestAttributes());
 	}
 
 	private static class SeataContextCallable<K> implements Callable<K> {
@@ -61,19 +143,24 @@ public class SeataHystrixConcurrencyStrategy extends HystrixConcurrencyStrategy
 
 		private final String xid;
 
-		SeataContextCallable(Callable<K> actual) {
+		private final RequestAttributes requestAttributes;
+
+		SeataContextCallable(Callable<K> actual, RequestAttributes requestAttribute) {
 			this.actual = actual;
+			this.requestAttributes = requestAttribute;
 			this.xid = RootContext.getXID();
 		}
 
 		@Override
 		public K call() throws Exception {
 			try {
+				RequestContextHolder.setRequestAttributes(requestAttributes);
 				RootContext.bind(xid);
 				return actual.call();
 			}
 			finally {
 				RootContext.unbind();
+				RequestContextHolder.resetRequestAttributes();
 			}
 		}
 

spring-cloud-alibaba-starters/spring-cloud-starter-dubbo/src/main/java/com/alibaba/cloud/dubbo/metadata/repository/DubboServiceMetadataRepository.java

@@ -190,7 +190,7 @@ public class DubboServiceMetadataRepository
 	// //
 
 	private static <K, V> Map<K, V> getMap(Map<String, Map<K, V>> repository,
-										   String key) {
+			String key) {
 		return getOrDefault(repository, key, newHashMap());
 	}
 
@@ -343,7 +343,7 @@ public class DubboServiceMetadataRepository
 	}
 
 	private void addDubboMetadataServiceURLsMetadata(Map<String, String> metadata,
-													 List<URL> dubboMetadataServiceURLs) {
+			List<URL> dubboMetadataServiceURLs) {
 		String dubboMetadataServiceURLsJSON = jsonUtils.toJSON(dubboMetadataServiceURLs);
 		metadata.put(DUBBO_METADATA_SERVICE_URLS_PROPERTY_NAME,
 				dubboMetadataServiceURLsJSON);
@@ -390,7 +390,7 @@ public class DubboServiceMetadataRepository
 	}
 
 	public List<URL> findSubscribedDubboMetadataServiceURLs(String serviceName,
-															String group, String version, String protocol) {
+			String group, String version, String protocol) {
 		String serviceKey = URL.buildKey(serviceName, group, version);
 
 		List<URL> urls = null;
@@ -403,9 +403,11 @@ public class DubboServiceMetadataRepository
 			return emptyList();
 		}
 
-		return hasText(protocol) ? urls.stream()
-				.filter(url -> url.getProtocol().equalsIgnoreCase(protocol))
-				.collect(Collectors.toList()) : unmodifiableList(urls);
+		return hasText(protocol)
+				? urls.stream()
+						.filter(url -> url.getProtocol().equalsIgnoreCase(protocol))
+						.collect(Collectors.toList())
+				: unmodifiableList(urls);
 	}
 
 	/**
@@ -470,7 +472,7 @@ public class DubboServiceMetadataRepository
 	}
 
 	public Integer getDubboProtocolPort(ServiceInstance serviceInstance,
-										String protocol) {
+			String protocol) {
 		String protocolProperty = getDubboProtocolPropertyName(protocol);
 		Map<String, String> metadata = serviceInstance.getMetadata();
 		String protocolPort = metadata.get(protocolProperty);
@@ -478,7 +480,7 @@ public class DubboServiceMetadataRepository
 	}
 
 	public List<URL> getExportedURLs(String serviceInterface, String group,
-									 String version) {
+			String version) {
 		String serviceKey = URL.buildKey(serviceInterface, group, version);
 		return allExportedURLs.getOrDefault(serviceKey, Collections.emptyList());
 	}
@@ -535,7 +537,7 @@ public class DubboServiceMetadataRepository
 	 * @return {@link DubboRestServiceMetadata} if matched, or <code>null</code>
 	 */
 	public DubboRestServiceMetadata get(String serviceName,
-										RequestMetadata requestMetadata) {
+			RequestMetadata requestMetadata) {
 		return match(dubboRestServiceMetadataRepository, serviceName, requestMetadata);
 	}
 
@@ -552,7 +554,7 @@ public class DubboServiceMetadataRepository
 	}
 
 	private <T> T match(Map<String, Map<RequestMetadataMatcher, T>> repository,
-						String serviceName, RequestMetadata requestMetadata) {
+			String serviceName, RequestMetadata requestMetadata) {
 
 		Map<RequestMetadataMatcher, T> map = repository.get(serviceName);
 

spring-cloud-alibaba-starters/spring-cloud-starter-dubbo/src/main/java/com/alibaba/cloud/dubbo/registry/AbstractSpringCloudRegistry.java

@@ -98,10 +98,10 @@ public abstract class AbstractSpringCloudRegistry extends FailbackRegistry {
 	private final ConfigurableApplicationContext applicationContext;
 
 	public AbstractSpringCloudRegistry(URL url, DiscoveryClient discoveryClient,
-									   DubboServiceMetadataRepository dubboServiceMetadataRepository,
-									   DubboMetadataServiceProxy dubboMetadataConfigServiceProxy,
-									   JSONUtils jsonUtils, DubboGenericServiceFactory dubboGenericServiceFactory,
-									   ConfigurableApplicationContext applicationContext) {
+			DubboServiceMetadataRepository dubboServiceMetadataRepository,
+			DubboMetadataServiceProxy dubboMetadataConfigServiceProxy,
+			JSONUtils jsonUtils, DubboGenericServiceFactory dubboGenericServiceFactory,
+			ConfigurableApplicationContext applicationContext) {
 		super(url);
 		this.servicesLookupInterval = url
 				.getParameter(SERVICES_LOOKUP_INTERVAL_PARAM_NAME, 60L);
@@ -190,7 +190,7 @@ public abstract class AbstractSpringCloudRegistry extends FailbackRegistry {
 	 * @param listener {@link NotifyListener}
 	 */
 	private void registerServiceInstancesChangedEventListener(URL url,
-															  NotifyListener listener) {
+			NotifyListener listener) {
 		String listenerId = generateId(url);
 		if (registerListeners.add(listenerId)) {
 			applicationContext.addApplicationListener(
@@ -217,8 +217,8 @@ public abstract class AbstractSpringCloudRegistry extends FailbackRegistry {
 	}
 
 	protected void subscribeDubboServiceURL(URL url, NotifyListener listener,
-											String serviceName,
-											Function<String, Collection<ServiceInstance>> serviceInstancesFunction) {
+			String serviceName,
+			Function<String, Collection<ServiceInstance>> serviceInstancesFunction) {
 
 		if (logger.isInfoEnabled()) {
 			logger.info(
@@ -355,7 +355,7 @@ public abstract class AbstractSpringCloudRegistry extends FailbackRegistry {
 	}
 
 	private List<URL> getExportedURLs(DubboMetadataService dubboMetadataService,
-									  URL url) {
+			URL url) {
 		String serviceInterface = url.getServiceInterface();
 		String group = url.getParameter(GROUP_KEY);
 		String version = url.getParameter(VERSION_KEY);