Merge branch 'apache-3.1' into apache-3.2
# Conflicts: # .github/workflows/build-and-test-pr.yml # dubbo-common/src/main/java/org/apache/dubbo/common/constants/LoggerCodeConstants.java
This commit is contained in:
commit
3375f727e1
|
|
@ -199,12 +199,12 @@ jobs:
|
|||
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
|
||||
run: |
|
||||
source ${{ github.workspace }}/.tmp/decrypted-sonarcloud-token
|
||||
./mvnw --batch-mode --no-snapshot-updates -e --no-transfer-progress --fail-fast clean test verify org.sonarsource.scanner.maven:sonar-maven-plugin:sonar -Pjacoco,jdk15ge-simple,'!jdk15ge',jacoco089 -Dsonar.host.url=https://sonarcloud.io -Dsonar.organization=apache -Dsonar.projectKey=apache_dubbo -Dmaven.wagon.httpconnectionManager.ttlSeconds=120 -Dmaven.wagon.http.retryHandler.count=5 -DskipTests=false -DskipIntegrationTests=false -Dcheckstyle.skip=false -Dcheckstyle_unix.skip=false -Drat.skip=false -Dmaven.javadoc.skip=true -DembeddedZookeeperPath=${{ github.workspace }}/.tmp/zookeeper -Dsonar.coverage.jacoco.xmlReportPaths=dubbo-test/dubbo-dependencies-all/target/site/jacoco-aggregate/jacoco.xml -Dsonar.login=${SONAR_TOKEN}
|
||||
./mvnw --batch-mode --no-snapshot-updates -e --no-transfer-progress --fail-fast clean test verify org.sonarsource.scanner.maven:sonar-maven-plugin:sonar -Pjacoco,jdk15ge-simple,'!jdk15ge',jacoco089 -Dsonar.host.url=https://sonarcloud.io -Dsonar.organization=apache -Dsonar.projectKey=apache_dubbo -DtrimStackTrace=false -Dmaven.wagon.httpconnectionManager.ttlSeconds=120 -Dmaven.wagon.http.retryHandler.count=5 -DskipTests=false -DskipIntegrationTests=false -Dcheckstyle.skip=false -Dcheckstyle_unix.skip=false -Drat.skip=false -Dmaven.javadoc.skip=true -DembeddedZookeeperPath=${{ github.workspace }}/.tmp/zookeeper -Dsonar.coverage.jacoco.xmlReportPaths=dubbo-test/dubbo-dependencies-all/target/site/jacoco-aggregate/jacoco.xml -Dsonar.login=${SONAR_TOKEN}
|
||||
- name: "Test with Maven without SonarCloud Scan"
|
||||
if: ${{ github.repository != 'apache/dubbo' }}
|
||||
timeout-minutes: 70
|
||||
run: |
|
||||
./mvnw --batch-mode --no-snapshot-updates -e --no-transfer-progress --fail-fast clean test verify -Dmaven.wagon.httpconnectionManager.ttlSeconds=120 -Pjacoco,jdk15ge-simple,'!jdk15ge',jacoco089 -Dmaven.wagon.http.retryHandler.count=5 -DskipTests=false -DskipIntegrationTests=false -Dcheckstyle.skip=false -Dcheckstyle_unix.skip=false -Drat.skip=false -Dmaven.javadoc.skip=true -DembeddedZookeeperPath=${{ github.workspace }}/.tmp/zookeeper
|
||||
./mvnw --batch-mode --no-snapshot-updates -e --no-transfer-progress --fail-fast clean test verify -Dmaven.wagon.httpconnectionManager.ttlSeconds=120 -Pjacoco,jdk15ge-simple,'!jdk15ge',jacoco089 -DtrimStackTrace=false -Dmaven.wagon.http.retryHandler.count=5 -DskipTests=false -DskipIntegrationTests=false -Dcheckstyle.skip=false -Dcheckstyle_unix.skip=false -Drat.skip=false -Dmaven.javadoc.skip=true -DembeddedZookeeperPath=${{ github.workspace }}/.tmp/zookeeper
|
||||
- name: "Upload coverage to Codecov"
|
||||
uses: codecov/codecov-action@v3
|
||||
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ public interface LoggerCodeConstants {
|
|||
|
||||
String COMMON_CLASS_NOT_FOUND = "0-7";
|
||||
|
||||
String COMMON_FAILED_REFLECT = "0-8";
|
||||
String COMMON_REFLECTIVE_OPERATION_FAILED = "0-8";
|
||||
|
||||
String COMMON_FAILED_NOTIFY_EVENT = "0-9";
|
||||
|
||||
|
|
@ -126,10 +126,19 @@ public interface LoggerCodeConstants {
|
|||
|
||||
String REGISTRY_UNABLE_ACCESS_KUBERNETES = "1-22";
|
||||
|
||||
/**
|
||||
* Absent. Original '1-23' is changed to '81-3'.
|
||||
*/
|
||||
String REGISTRY_FAILED_DOWNLOAD_FILE = "1-23";
|
||||
|
||||
/**
|
||||
* Absent. Original '1-24' is changed to '81-1'.
|
||||
*/
|
||||
String REGISTRY_FAILED_START_ZOOKEEPER = "1-24";
|
||||
|
||||
/**
|
||||
* Absent. Original '1-25' is changed to '81-2'.
|
||||
*/
|
||||
String REGISTRY_FAILED_STOP_ZOOKEEPER = "1-25";
|
||||
|
||||
String REGISTRY_FAILED_GENERATE_CERT_ISTIO = "1-26";
|
||||
|
|
@ -167,7 +176,7 @@ public interface LoggerCodeConstants {
|
|||
|
||||
String REGISTRY_ISTIO_EXCEPTION = "1-41";
|
||||
|
||||
// cluster module 2-x
|
||||
// Cluster module 2-x
|
||||
String CLUSTER_FAILED_SITE_SELECTION = "2-1";
|
||||
|
||||
String CLUSTER_NO_VALID_PROVIDER = "2-2";
|
||||
|
|
@ -266,7 +275,7 @@ public interface LoggerCodeConstants {
|
|||
|
||||
String PROTOCOL_FAILED_DECODE = "4-20";
|
||||
|
||||
// config module
|
||||
// Config module
|
||||
String CONFIG_FAILED_CONNECT_REGISTRY = "5-1";
|
||||
|
||||
String CONFIG_FAILED_SHUTDOWN_HOOK = "5-2";
|
||||
|
|
@ -303,6 +312,9 @@ public interface LoggerCodeConstants {
|
|||
|
||||
String CONFIG_FAILED_NOTIFY_EVENT = "5-18";
|
||||
|
||||
/**
|
||||
* Absent. Changed to 81-4.
|
||||
*/
|
||||
String CONFIG_ZOOKEEPER_SERVER_ERROR = "5-19";
|
||||
|
||||
String CONFIG_STOP_DUBBO_ERROR = "5-20";
|
||||
|
|
@ -347,7 +359,7 @@ public interface LoggerCodeConstants {
|
|||
|
||||
String CONFIG_DUBBO_BEAN_NOT_FOUND = "5-40";
|
||||
|
||||
// transport module
|
||||
// Transport module
|
||||
String TRANSPORT_FAILED_CONNECT_PROVIDER = "6-1";
|
||||
|
||||
String TRANSPORT_CLIENT_CONNECT_TIMEOUT = "6-2";
|
||||
|
|
@ -395,6 +407,15 @@ public interface LoggerCodeConstants {
|
|||
|
||||
String QOS_PERMISSION_DENY_EXCEPTION = "7-7";
|
||||
|
||||
// Testing module (8[X], where [X] is number of the module to be tested.)
|
||||
String TESTING_REGISTRY_FAILED_TO_START_ZOOKEEPER = "81-1";
|
||||
|
||||
String TESTING_REGISTRY_FAILED_TO_STOP_ZOOKEEPER = "81-2";
|
||||
|
||||
String TESTING_REGISTRY_FAILED_TO_DOWNLOAD_ZK_FILE = "81-3";
|
||||
|
||||
String TESTING_INIT_ZOOKEEPER_SERVER_ERROR = "81-4";
|
||||
|
||||
// Internal unknown error.
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -58,7 +58,7 @@ import java.util.concurrent.ConcurrentSkipListMap;
|
|||
import java.util.function.Consumer;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_FAILED_REFLECT;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_REFLECTIVE_OPERATION_FAILED;
|
||||
import static org.apache.dubbo.common.utils.ClassUtils.isAssignableFrom;
|
||||
|
||||
/**
|
||||
|
|
@ -526,7 +526,7 @@ public class PojoUtils {
|
|||
} catch (Exception e) {
|
||||
String exceptionDescription = "Failed to set pojo " + dest.getClass().getSimpleName() + " property " + name
|
||||
+ " value " + value.getClass() + ", cause: " + e.getMessage();
|
||||
logger.error(COMMON_FAILED_REFLECT, "", "", exceptionDescription, e);
|
||||
logger.error(COMMON_REFLECTIVE_OPERATION_FAILED, "", "", exceptionDescription, e);
|
||||
throw new RuntimeException(exceptionDescription, e);
|
||||
}
|
||||
} else if (field != null) {
|
||||
|
|
|
|||
|
|
@ -61,7 +61,7 @@ import java.util.concurrent.ConcurrentHashMap;
|
|||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_FAILED_OVERRIDE_FIELD;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_FAILED_REFLECT;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_REFLECTIVE_OPERATION_FAILED;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_UNEXPECTED_EXCEPTION;
|
||||
import static org.apache.dubbo.common.utils.ClassUtils.isSimpleType;
|
||||
import static org.apache.dubbo.common.utils.ReflectUtils.findMethodByMethodSignature;
|
||||
|
|
@ -507,7 +507,7 @@ public abstract class AbstractConfig implements Serializable {
|
|||
}
|
||||
}
|
||||
} catch (Throwable e) {
|
||||
logger.error(COMMON_FAILED_REFLECT, "", "", e.getMessage(), e);
|
||||
logger.error(COMMON_REFLECTIVE_OPERATION_FAILED, "", "", e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -32,7 +32,7 @@ import java.util.stream.Stream;
|
|||
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.$INVOKE;
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.$INVOKE_ASYNC;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_FAILED_REFLECT;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_REFLECTIVE_OPERATION_FAILED;
|
||||
|
||||
public class ReflectionMethodDescriptor implements MethodDescriptor {
|
||||
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(ReflectionMethodDescriptor.class);
|
||||
|
|
@ -58,7 +58,7 @@ public class ReflectionMethodDescriptor implements MethodDescriptor {
|
|||
try {
|
||||
returnTypesResult = ReflectUtils.getReturnTypes(method);
|
||||
} catch (Throwable throwable) {
|
||||
logger.error(COMMON_FAILED_REFLECT, "", "",
|
||||
logger.error(COMMON_REFLECTIVE_OPERATION_FAILED, "", "",
|
||||
"fail to get return types. Method name: " + methodName + " Declaring class:" + method.getDeclaringClass()
|
||||
.getName(), throwable);
|
||||
returnTypesResult = new Type[]{returnClass, returnClass};
|
||||
|
|
|
|||
|
|
@ -21,7 +21,6 @@ import org.apache.dubbo.common.logger.Logger;
|
|||
import org.junit.jupiter.api.Assertions;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_UNEXPECTED_EXCEPTION;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.Mockito.doThrow;
|
||||
|
|
@ -32,7 +31,7 @@ class FailsafeLoggerTest {
|
|||
@Test
|
||||
void testFailSafeForLoggingMethod() {
|
||||
Logger failLogger = mock(Logger.class);
|
||||
FailsafeErrorTypeAwareLogger failsafeLogger = new FailsafeErrorTypeAwareLogger(failLogger);
|
||||
FailsafeLogger failsafeLogger = new FailsafeLogger(failLogger);
|
||||
|
||||
doThrow(new RuntimeException()).when(failLogger).error(anyString());
|
||||
doThrow(new RuntimeException()).when(failLogger).warn(anyString());
|
||||
|
|
@ -40,8 +39,8 @@ class FailsafeLoggerTest {
|
|||
doThrow(new RuntimeException()).when(failLogger).debug(anyString());
|
||||
doThrow(new RuntimeException()).when(failLogger).trace(anyString());
|
||||
|
||||
failsafeLogger.error(COMMON_UNEXPECTED_EXCEPTION, "", "", "error");
|
||||
failsafeLogger.warn(COMMON_UNEXPECTED_EXCEPTION, "", "", "warn");
|
||||
failsafeLogger.error("error");
|
||||
failsafeLogger.warn("warn");
|
||||
failsafeLogger.info("info");
|
||||
failsafeLogger.debug("debug");
|
||||
failsafeLogger.trace("info");
|
||||
|
|
@ -52,14 +51,14 @@ class FailsafeLoggerTest {
|
|||
doThrow(new RuntimeException()).when(failLogger).debug(any(Throwable.class));
|
||||
doThrow(new RuntimeException()).when(failLogger).trace(any(Throwable.class));
|
||||
|
||||
failsafeLogger.error(COMMON_UNEXPECTED_EXCEPTION, "", "", "error", new Exception("error"));
|
||||
failsafeLogger.warn(COMMON_UNEXPECTED_EXCEPTION, "", "", "warn", new Exception("warn"));
|
||||
failsafeLogger.error(new Exception("error"));
|
||||
failsafeLogger.warn(new Exception("warn"));
|
||||
failsafeLogger.info(new Exception("info"));
|
||||
failsafeLogger.debug(new Exception("debug"));
|
||||
failsafeLogger.trace(new Exception("trace"));
|
||||
|
||||
failsafeLogger.error(COMMON_UNEXPECTED_EXCEPTION, "", "", "error", new Exception("error"));
|
||||
failsafeLogger.warn(COMMON_UNEXPECTED_EXCEPTION, "", "", "warn", new Exception("warn"));
|
||||
failsafeLogger.error("error", new Exception("error"));
|
||||
failsafeLogger.warn("warn", new Exception("warn"));
|
||||
failsafeLogger.info("info", new Exception("info"));
|
||||
failsafeLogger.debug("debug", new Exception("debug"));
|
||||
failsafeLogger.trace("trace", new Exception("trace"));
|
||||
|
|
@ -68,9 +67,9 @@ class FailsafeLoggerTest {
|
|||
@Test
|
||||
void testSuccessLogger() {
|
||||
Logger successLogger = mock(Logger.class);
|
||||
FailsafeErrorTypeAwareLogger failsafeLogger = new FailsafeErrorTypeAwareLogger(successLogger);
|
||||
failsafeLogger.error(COMMON_UNEXPECTED_EXCEPTION, "", "", "error");
|
||||
failsafeLogger.warn(COMMON_UNEXPECTED_EXCEPTION, "", "", "warn");
|
||||
FailsafeLogger failsafeLogger = new FailsafeLogger(successLogger);
|
||||
failsafeLogger.error("error");
|
||||
failsafeLogger.warn("warn");
|
||||
failsafeLogger.info("info");
|
||||
failsafeLogger.debug("debug");
|
||||
failsafeLogger.trace("info");
|
||||
|
|
@ -81,14 +80,14 @@ class FailsafeLoggerTest {
|
|||
verify(successLogger).debug(anyString());
|
||||
verify(successLogger).trace(anyString());
|
||||
|
||||
failsafeLogger.error(COMMON_UNEXPECTED_EXCEPTION, "", "", "error", new Exception("error"));
|
||||
failsafeLogger.warn(COMMON_UNEXPECTED_EXCEPTION, "", "", "warn", new Exception("warn"));
|
||||
failsafeLogger.error(new Exception("error"));
|
||||
failsafeLogger.warn(new Exception("warn"));
|
||||
failsafeLogger.info(new Exception("info"));
|
||||
failsafeLogger.debug(new Exception("debug"));
|
||||
failsafeLogger.trace(new Exception("trace"));
|
||||
|
||||
failsafeLogger.error(COMMON_UNEXPECTED_EXCEPTION, "", "", "error", new Exception("error"));
|
||||
failsafeLogger.warn(COMMON_UNEXPECTED_EXCEPTION, "", "", "warn", new Exception("warn"));
|
||||
failsafeLogger.error("error", new Exception("error"));
|
||||
failsafeLogger.warn("warn", new Exception("warn"));
|
||||
failsafeLogger.info("info", new Exception("info"));
|
||||
failsafeLogger.debug("debug", new Exception("debug"));
|
||||
failsafeLogger.trace("trace", new Exception("trace"));
|
||||
|
|
@ -98,10 +97,10 @@ class FailsafeLoggerTest {
|
|||
void testGetLogger() {
|
||||
Assertions.assertThrows(RuntimeException.class, () -> {
|
||||
Logger failLogger = mock(Logger.class);
|
||||
FailsafeErrorTypeAwareLogger failsafeLogger = new FailsafeErrorTypeAwareLogger(failLogger);
|
||||
FailsafeLogger failsafeLogger = new FailsafeLogger(failLogger);
|
||||
|
||||
doThrow(new RuntimeException()).when(failLogger).error(anyString());
|
||||
failsafeLogger.getLogger().error("should get error");
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -29,8 +29,8 @@ import java.lang.reflect.Method;
|
|||
import java.util.Properties;
|
||||
import java.util.UUID;
|
||||
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.CONFIG_ZOOKEEPER_SERVER_ERROR;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_STOP_ZOOKEEPER;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TESTING_INIT_ZOOKEEPER_SERVER_ERROR;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TESTING_REGISTRY_FAILED_TO_STOP_ZOOKEEPER;
|
||||
|
||||
/**
|
||||
* from: https://github.com/spring-projects/spring-xd/blob/v1.3.1.RELEASE/spring-xd-dirt/src/main/java/org/springframework/xd/dirt/zookeeper/ZooKeeperUtils.java
|
||||
|
|
@ -187,7 +187,7 @@ public class EmbeddedZooKeeper implements SmartLifecycle {
|
|||
zkServerThread = null;
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
logger.warn(REGISTRY_FAILED_STOP_ZOOKEEPER, "", "", "Interrupted while waiting for embedded ZooKeeper to exit");
|
||||
logger.warn(TESTING_REGISTRY_FAILED_TO_STOP_ZOOKEEPER, "", "", "Interrupted while waiting for embedded ZooKeeper to exit");
|
||||
// abandoning zk thread
|
||||
zkServerThread = null;
|
||||
}
|
||||
|
|
@ -240,7 +240,7 @@ public class EmbeddedZooKeeper implements SmartLifecycle {
|
|||
if (errorHandler != null) {
|
||||
errorHandler.handleError(e);
|
||||
} else {
|
||||
logger.error(CONFIG_ZOOKEEPER_SERVER_ERROR, "ZooKeeper server error", "", "Exception running embedded ZooKeeper.", e);
|
||||
logger.error(TESTING_INIT_ZOOKEEPER_SERVER_ERROR, "ZooKeeper server error", "", "Exception running embedded ZooKeeper.", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -31,8 +31,9 @@ import org.apache.dubbo.rpc.model.ApplicationModel;
|
|||
|
||||
import org.apache.curator.framework.CuratorFramework;
|
||||
import org.apache.curator.framework.api.CuratorWatcher;
|
||||
import org.apache.zookeeper.KeeperException;
|
||||
import org.apache.curator.x.discovery.ServiceCache;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
|
@ -144,13 +145,17 @@ public class ZookeeperServiceDiscovery extends AbstractServiceDiscovery {
|
|||
@Override
|
||||
public void removeServiceInstancesChangedListener(ServiceInstancesChangedListener listener) throws IllegalArgumentException {
|
||||
listener.getServiceNames().forEach(serviceName -> {
|
||||
String servicePath = buildServicePath(serviceName);
|
||||
ZookeeperServiceDiscoveryChangeWatcher watcher = watcherCaches.get(servicePath);
|
||||
ZookeeperServiceDiscoveryChangeWatcher watcher = watcherCaches.get(serviceName);
|
||||
if (watcher != null) {
|
||||
watcher.getListeners().remove(listener);
|
||||
if (watcher.getListeners().isEmpty()) {
|
||||
watcher.stopWatching();
|
||||
watcherCaches.remove(servicePath);
|
||||
watcherCaches.remove(serviceName);
|
||||
try {
|
||||
watcher.getCacheInstance().close();
|
||||
} catch (IOException e) {
|
||||
logger.error(REGISTRY_ZOOKEEPER_EXCEPTION, "curator stop watch failed", "",
|
||||
"Curator Stop service discovery watch failed. Service Name: " + serviceName);
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
|
@ -166,66 +171,26 @@ public class ZookeeperServiceDiscovery extends AbstractServiceDiscovery {
|
|||
}
|
||||
|
||||
protected void registerServiceWatcher(String serviceName, ServiceInstancesChangedListener listener) {
|
||||
String path = buildServicePath(serviceName);
|
||||
try {
|
||||
curatorFramework.create().creatingParentsIfNeeded().forPath(path);
|
||||
} catch (KeeperException.NodeExistsException e) {
|
||||
// ignored
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug(e);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
throw new IllegalStateException("registerServiceWatcher create path=" + path + " fail.", e);
|
||||
}
|
||||
|
||||
CountDownLatch latch = new CountDownLatch(1);
|
||||
ZookeeperServiceDiscoveryChangeWatcher watcher = watcherCaches.computeIfAbsent(path, key -> {
|
||||
ZookeeperServiceDiscoveryChangeWatcher tmpWatcher = new ZookeeperServiceDiscoveryChangeWatcher(this, serviceName, path, latch);
|
||||
|
||||
ZookeeperServiceDiscoveryChangeWatcher watcher = watcherCaches.computeIfAbsent(serviceName, name -> {
|
||||
ServiceCache<ZookeeperInstance> serviceCache = serviceDiscovery.serviceCacheBuilder()
|
||||
.name(name)
|
||||
.build();
|
||||
ZookeeperServiceDiscoveryChangeWatcher newer = new ZookeeperServiceDiscoveryChangeWatcher(this, serviceCache, name, latch);
|
||||
serviceCache.addListener(newer);
|
||||
|
||||
try {
|
||||
curatorFramework.getChildren().usingWatcher(tmpWatcher).forPath(path);
|
||||
} catch (KeeperException.NoNodeException e) {
|
||||
// ignored
|
||||
if (logger.isErrorEnabled()) {
|
||||
logger.error(REGISTRY_ZOOKEEPER_EXCEPTION, "", "", e.getMessage());
|
||||
}
|
||||
serviceCache.start();
|
||||
} catch (Exception e) {
|
||||
throw new IllegalStateException(e.getMessage(), e);
|
||||
throw new RpcException(REGISTRY_EXCEPTION, "Failed subscribe service: " + name, e);
|
||||
}
|
||||
return tmpWatcher;
|
||||
|
||||
return newer;
|
||||
});
|
||||
|
||||
watcher.addListener(listener);
|
||||
listener.onEvent(new ServiceInstancesChangedEvent(serviceName, this.getInstances(serviceName)));
|
||||
latch.countDown();
|
||||
}
|
||||
|
||||
/**
|
||||
* 1. re-register, taken care by curator ServiceDiscovery
|
||||
* 2. re-subscribe, register curator watcher and notify the latest provider list
|
||||
*/
|
||||
public void recover() {
|
||||
watcherCaches.forEach((path, watcher) -> {
|
||||
CountDownLatch latch = new CountDownLatch(1);
|
||||
Set<ServiceInstancesChangedListener> listeners = watcher.getListeners();
|
||||
try {
|
||||
watcher.setLatch(latch);
|
||||
curatorFramework.getChildren().usingWatcher(watcher).forPath(path);
|
||||
} catch (Exception e) {
|
||||
logger.error(REGISTRY_ZOOKEEPER_EXCEPTION, "", "", "Trying to recover from new zkClient session failed, path is " + path + ", error msg: " + e.getMessage());
|
||||
}
|
||||
|
||||
List<ServiceInstance> instances = this.getInstances(watcher.getServiceName());
|
||||
for (ServiceInstancesChangedListener listener : listeners) {
|
||||
listener.onEvent(new ServiceInstancesChangedEvent(watcher.getServiceName(), instances));
|
||||
}
|
||||
latch.countDown();
|
||||
});
|
||||
}
|
||||
|
||||
public void reRegisterWatcher(ZookeeperServiceDiscoveryChangeWatcher watcher) throws Exception {
|
||||
curatorFramework.getChildren().usingWatcher(watcher).forPath(watcher.getPath());
|
||||
}
|
||||
|
||||
private String buildServicePath(String serviceName) {
|
||||
return rootPath + "/" + serviceName;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -25,17 +25,17 @@ import org.apache.dubbo.registry.client.event.ServiceInstancesChangedEvent;
|
|||
import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener;
|
||||
import org.apache.dubbo.rpc.model.ScopeModelUtil;
|
||||
|
||||
import org.apache.curator.framework.CuratorFramework;
|
||||
import org.apache.curator.framework.api.CuratorWatcher;
|
||||
import org.apache.zookeeper.WatchedEvent;
|
||||
import org.apache.curator.framework.state.ConnectionState;
|
||||
import org.apache.curator.x.discovery.ServiceCache;
|
||||
import org.apache.curator.x.discovery.details.ServiceCacheListener;
|
||||
import org.apache.zookeeper.Watcher;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
|
||||
import static org.apache.zookeeper.Watcher.Event.EventType.NodeChildrenChanged;
|
||||
import static org.apache.zookeeper.Watcher.Event.EventType.NodeDataChanged;
|
||||
|
||||
/**
|
||||
* Zookeeper {@link ServiceDiscovery} Change {@link CuratorWatcher watcher} only interests in
|
||||
* {@link Watcher.Event.EventType#NodeChildrenChanged} and {@link Watcher.Event.EventType#NodeDataChanged} event types,
|
||||
|
|
@ -43,29 +43,26 @@ import static org.apache.zookeeper.Watcher.Event.EventType.NodeDataChanged;
|
|||
*
|
||||
* @since 2.7.5
|
||||
*/
|
||||
public class ZookeeperServiceDiscoveryChangeWatcher implements CuratorWatcher {
|
||||
private Set<ServiceInstancesChangedListener> listeners = new ConcurrentHashSet<>();
|
||||
public class ZookeeperServiceDiscoveryChangeWatcher implements ServiceCacheListener {
|
||||
private final Set<ServiceInstancesChangedListener> listeners = new ConcurrentHashSet<>();
|
||||
|
||||
private final ServiceCache<ZookeeperInstance> cacheInstance;
|
||||
|
||||
private final ZookeeperServiceDiscovery zookeeperServiceDiscovery;
|
||||
|
||||
private final RegistryNotifier notifier;
|
||||
|
||||
private volatile boolean keepWatching = true;
|
||||
|
||||
private final String serviceName;
|
||||
|
||||
private final String path;
|
||||
|
||||
private CountDownLatch latch;
|
||||
private final CountDownLatch latch;
|
||||
|
||||
public ZookeeperServiceDiscoveryChangeWatcher(ZookeeperServiceDiscovery zookeeperServiceDiscovery,
|
||||
ServiceCache<ZookeeperInstance> cacheInstance,
|
||||
String serviceName,
|
||||
String path,
|
||||
CountDownLatch latch) {
|
||||
this.zookeeperServiceDiscovery = zookeeperServiceDiscovery;
|
||||
this.cacheInstance = cacheInstance;
|
||||
this.serviceName = serviceName;
|
||||
this.path = path;
|
||||
this.latch = latch;
|
||||
this.notifier = new RegistryNotifier(zookeeperServiceDiscovery.getUrl(), zookeeperServiceDiscovery.getDelay(),
|
||||
ScopeModelUtil.getFrameworkModel(zookeeperServiceDiscovery.getUrl().getScopeModel()).getBeanFactory()
|
||||
.getBean(FrameworkExecutorRepository.class).getServiceDiscoveryAddressNotificationExecutor()) {
|
||||
|
|
@ -74,51 +71,35 @@ public class ZookeeperServiceDiscoveryChangeWatcher implements CuratorWatcher {
|
|||
listeners.forEach(listener -> listener.onEvent((ServiceInstancesChangedEvent) rawAddresses));
|
||||
}
|
||||
};
|
||||
this.latch = latch;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void process(WatchedEvent event) throws Exception {
|
||||
public void cacheChanged() {
|
||||
try {
|
||||
latch.await();
|
||||
} catch (InterruptedException e) {
|
||||
} catch (InterruptedException ignore) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
|
||||
Watcher.Event.EventType eventType = event.getType();
|
||||
|
||||
if (NodeChildrenChanged.equals(eventType) || NodeDataChanged.equals(eventType)) {
|
||||
if (shouldKeepWatching()) {
|
||||
zookeeperServiceDiscovery.reRegisterWatcher(this);
|
||||
List<ServiceInstance> instanceList = zookeeperServiceDiscovery.getInstances(serviceName);
|
||||
notifier.notify(new ServiceInstancesChangedEvent(serviceName, instanceList));
|
||||
}
|
||||
}
|
||||
List<ServiceInstance> instanceList = zookeeperServiceDiscovery.getInstances(serviceName);
|
||||
notifier.notify(new ServiceInstancesChangedEvent(serviceName, instanceList));
|
||||
}
|
||||
|
||||
public String getPath() {
|
||||
return path;
|
||||
@Override
|
||||
public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
|
||||
// ignore: taken care by curator ServiceDiscovery
|
||||
}
|
||||
|
||||
public void addListener(ServiceInstancesChangedListener listener) {
|
||||
listeners.add(listener);
|
||||
public ServiceCache<ZookeeperInstance> getCacheInstance() {
|
||||
return cacheInstance;
|
||||
}
|
||||
|
||||
public Set<ServiceInstancesChangedListener> getListeners() {
|
||||
return listeners;
|
||||
}
|
||||
|
||||
public boolean shouldKeepWatching() {
|
||||
return keepWatching;
|
||||
}
|
||||
|
||||
public void stopWatching() {
|
||||
this.keepWatching = false;
|
||||
}
|
||||
|
||||
public void setLatch(CountDownLatch latch) {
|
||||
this.latch = latch;
|
||||
}
|
||||
|
||||
public String getServiceName() {
|
||||
return serviceName;
|
||||
public void addListener(ServiceInstancesChangedListener listener) {
|
||||
listeners.add(listener);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -17,8 +17,6 @@
|
|||
package org.apache.dubbo.registry.zookeeper.util;
|
||||
|
||||
import org.apache.dubbo.common.URL;
|
||||
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
|
||||
import org.apache.dubbo.common.logger.LoggerFactory;
|
||||
import org.apache.dubbo.common.utils.StringUtils;
|
||||
import org.apache.dubbo.registry.client.DefaultServiceInstance;
|
||||
import org.apache.dubbo.registry.client.ServiceInstance;
|
||||
|
|
@ -31,8 +29,6 @@ import org.apache.curator.framework.CuratorFramework;
|
|||
import org.apache.curator.framework.CuratorFrameworkFactory;
|
||||
import org.apache.curator.framework.api.ACLProvider;
|
||||
import org.apache.curator.framework.imps.CuratorFrameworkState;
|
||||
import org.apache.curator.framework.state.ConnectionState;
|
||||
import org.apache.curator.framework.state.ConnectionStateListener;
|
||||
import org.apache.curator.retry.ExponentialBackoffRetry;
|
||||
import org.apache.curator.x.discovery.ServiceDiscovery;
|
||||
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
|
||||
|
|
@ -47,9 +43,6 @@ import java.util.stream.Collectors;
|
|||
|
||||
import static org.apache.curator.x.discovery.ServiceInstance.builder;
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.PATH_SEPARATOR;
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.SESSION_KEY;
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ZOOKEEPER_EXCEPTION;
|
||||
import static org.apache.dubbo.registry.zookeeper.ZookeeperServiceDiscovery.DEFAULT_GROUP;
|
||||
import static org.apache.dubbo.registry.zookeeper.util.CuratorFrameworkParams.BASE_SLEEP_TIME;
|
||||
import static org.apache.dubbo.registry.zookeeper.util.CuratorFrameworkParams.BLOCK_UNTIL_CONNECTED_UNIT;
|
||||
|
|
@ -95,8 +88,6 @@ public abstract class CuratorFrameworkUtils {
|
|||
}
|
||||
CuratorFramework curatorFramework = builder.build();
|
||||
|
||||
curatorFramework.getConnectionStateListenable().addListener(new CuratorConnectionStateListener(connectionURL, serviceDiscovery));
|
||||
|
||||
curatorFramework.start();
|
||||
curatorFramework.blockUntilConnected(BLOCK_UNTIL_CONNECTED_WAIT.getParameterValue(connectionURL),
|
||||
BLOCK_UNTIL_CONNECTED_UNIT.getParameterValue(connectionURL));
|
||||
|
|
@ -169,54 +160,4 @@ public abstract class CuratorFrameworkUtils {
|
|||
}
|
||||
return group;
|
||||
}
|
||||
|
||||
private static class CuratorConnectionStateListener implements ConnectionStateListener {
|
||||
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(CuratorConnectionStateListener.class);
|
||||
private final long UNKNOWN_SESSION_ID = -1L;
|
||||
protected final int DEFAULT_CONNECTION_TIMEOUT_MS = 30 * 1000;
|
||||
protected final int DEFAULT_SESSION_TIMEOUT_MS = 60 * 1000;
|
||||
|
||||
private long lastSessionId;
|
||||
private final int timeout;
|
||||
private final int sessionExpireMs;
|
||||
private final ZookeeperServiceDiscovery serviceDiscovery;
|
||||
|
||||
public CuratorConnectionStateListener(URL url, ZookeeperServiceDiscovery serviceDiscovery) {
|
||||
this.timeout = url.getParameter(TIMEOUT_KEY, DEFAULT_CONNECTION_TIMEOUT_MS);
|
||||
this.sessionExpireMs = url.getParameter(SESSION_KEY, DEFAULT_SESSION_TIMEOUT_MS);
|
||||
this.serviceDiscovery = serviceDiscovery;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stateChanged(CuratorFramework client, ConnectionState state) {
|
||||
long sessionId = UNKNOWN_SESSION_ID;
|
||||
try {
|
||||
sessionId = client.getZookeeperClient().getZooKeeper().getSessionId();
|
||||
} catch (Exception e) {
|
||||
logger.warn(REGISTRY_ZOOKEEPER_EXCEPTION, "", "", "Curator client state changed, but failed to get the related zk session instance.");
|
||||
}
|
||||
|
||||
if (state == ConnectionState.LOST) {
|
||||
logger.warn(REGISTRY_ZOOKEEPER_EXCEPTION, "", "", "Curator zookeeper session " + Long.toHexString(lastSessionId) + " expired.");
|
||||
} else if (state == ConnectionState.SUSPENDED) {
|
||||
logger.warn(REGISTRY_ZOOKEEPER_EXCEPTION, "", "", "Curator zookeeper connection of session " + Long.toHexString(sessionId) + " timed out. " +
|
||||
"connection timeout value is " + timeout + ", session expire timeout value is " + sessionExpireMs);
|
||||
} else if (state == ConnectionState.CONNECTED) {
|
||||
lastSessionId = sessionId;
|
||||
logger.info("Curator zookeeper client instance initiated successfully, session id is " + Long.toHexString(sessionId));
|
||||
} else if (state == ConnectionState.RECONNECTED) {
|
||||
if (lastSessionId == sessionId && sessionId != UNKNOWN_SESSION_ID) {
|
||||
logger.warn(REGISTRY_ZOOKEEPER_EXCEPTION, "", "", "Curator zookeeper connection recovered from connection lose, " +
|
||||
"reuse the old session " + Long.toHexString(sessionId));
|
||||
serviceDiscovery.recover();
|
||||
} else {
|
||||
logger.warn(REGISTRY_ZOOKEEPER_EXCEPTION, "", "", "New session created after old session lost, " +
|
||||
"old session " + Long.toHexString(lastSessionId) + ", new session " + Long.toHexString(sessionId));
|
||||
lastSessionId = sessionId;
|
||||
serviceDiscovery.recover();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -44,7 +44,7 @@ import java.lang.reflect.Type;
|
|||
import static org.apache.dubbo.common.constants.CommonConstants.$INVOKE;
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.$INVOKE_ASYNC;
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.GENERIC_PARAMETER_DESC;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_FAILED_REFLECT;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_REFLECTIVE_OPERATION_FAILED;
|
||||
import static org.apache.dubbo.rpc.Constants.GENERIC_KEY;
|
||||
|
||||
/**
|
||||
|
|
@ -196,7 +196,7 @@ public class GenericImplFilter implements Filter, Filter.Listener {
|
|||
}
|
||||
field.set(targetException, exception.getExceptionMessage());
|
||||
} catch (Throwable e) {
|
||||
logger.warn(COMMON_FAILED_REFLECT, "", "", e.getMessage(), e);
|
||||
logger.warn(COMMON_REFLECTIVE_OPERATION_FAILED, "", "", e.getMessage(), e);
|
||||
}
|
||||
appResponse.setException(targetException);
|
||||
} else if (lastException != null) {
|
||||
|
|
|
|||
|
|
@ -38,7 +38,7 @@ import static org.apache.dubbo.common.constants.CommonConstants.GENERIC_PARAMETE
|
|||
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_ATTACHMENT_KEY;
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_ATTACHMENT_KEY_LOWER;
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_FAILED_REFLECT;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_REFLECTIVE_OPERATION_FAILED;
|
||||
import static org.apache.dubbo.rpc.Constants.$ECHO;
|
||||
import static org.apache.dubbo.rpc.Constants.$ECHO_PARAMETER_DESC;
|
||||
import static org.apache.dubbo.rpc.Constants.ASYNC_KEY;
|
||||
|
|
@ -67,7 +67,7 @@ public class RpcUtils {
|
|||
}
|
||||
}
|
||||
} catch (Throwable t) {
|
||||
logger.warn(COMMON_FAILED_REFLECT, "", "", t.getMessage(), t);
|
||||
logger.warn(COMMON_REFLECTIVE_OPERATION_FAILED, "", "", t.getMessage(), t);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
|
@ -97,7 +97,7 @@ public class RpcUtils {
|
|||
}
|
||||
}
|
||||
} catch (Throwable t) {
|
||||
logger.warn(COMMON_FAILED_REFLECT, "", "", t.getMessage(), t);
|
||||
logger.warn(COMMON_REFLECTIVE_OPERATION_FAILED, "", "", t.getMessage(), t);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -56,7 +56,7 @@ import java.nio.charset.StandardCharsets;
|
|||
import java.util.Map;
|
||||
import java.util.concurrent.Executor;
|
||||
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_FAILED_REFLECT;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_REFLECTIVE_OPERATION_FAILED;
|
||||
|
||||
|
||||
/**
|
||||
|
|
@ -256,7 +256,7 @@ public class TripleClientStream extends AbstractStream implements ClientStream {
|
|||
}
|
||||
});
|
||||
} else {
|
||||
LOGGER.error(COMMON_FAILED_REFLECT, "", "", "Triple convertNoLowerCaseHeader error, obj is not String");
|
||||
LOGGER.error(COMMON_REFLECTIVE_OPERATION_FAILED, "", "", "Triple convertNoLowerCaseHeader error, obj is not String");
|
||||
}
|
||||
return attachments;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ import java.util.concurrent.Future;
|
|||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_DOWNLOAD_FILE;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TESTING_REGISTRY_FAILED_TO_DOWNLOAD_ZK_FILE;
|
||||
|
||||
/**
|
||||
* Download zookeeper binary archive.
|
||||
|
|
@ -169,7 +169,7 @@ public class DownloadZookeeperInitializer extends ZookeeperInitializer {
|
|||
|
||||
@Override
|
||||
public void onThrowable(Throwable t) {
|
||||
logger.warn(REGISTRY_FAILED_DOWNLOAD_FILE, "", "", "Failed to download the file, download url: " + url);
|
||||
logger.warn(TESTING_REGISTRY_FAILED_TO_DOWNLOAD_ZK_FILE, "", "", "Failed to download the file, download url: " + url);
|
||||
super.onThrowable(t);
|
||||
}
|
||||
});
|
||||
|
|
|
|||
|
|
@ -29,8 +29,8 @@ import java.io.InputStream;
|
|||
import java.io.InputStreamReader;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_START_ZOOKEEPER;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_STOP_ZOOKEEPER;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TESTING_REGISTRY_FAILED_TO_START_ZOOKEEPER;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TESTING_REGISTRY_FAILED_TO_STOP_ZOOKEEPER;
|
||||
|
||||
/**
|
||||
* The abstract implementation of {@link Processor} is to provide some common methods on Unix OS.
|
||||
|
|
@ -50,7 +50,7 @@ public abstract class ZookeeperUnixProcessor implements Processor {
|
|||
try {
|
||||
process.destroy();
|
||||
} catch (Throwable cause) {
|
||||
logger.warn(REGISTRY_FAILED_STOP_ZOOKEEPER, "", "", String.format("Failed to kill the process, with client port %s !", clientPort), cause);
|
||||
logger.warn(TESTING_REGISTRY_FAILED_TO_STOP_ZOOKEEPER, "", "", String.format("Failed to kill the process, with client port %s !", clientPort), cause);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -64,7 +64,7 @@ public abstract class ZookeeperUnixProcessor implements Processor {
|
|||
try (final BufferedReader reader = new BufferedReader(new InputStreamReader(errorStream))) {
|
||||
String line;
|
||||
while ((line = reader.readLine()) != null) {
|
||||
logger.error(REGISTRY_FAILED_START_ZOOKEEPER,"","",line);
|
||||
logger.error(TESTING_REGISTRY_FAILED_TO_START_ZOOKEEPER,"","",line);
|
||||
}
|
||||
} catch (IOException e) {
|
||||
/* eat quietly */
|
||||
|
|
|
|||
Loading…
Reference in New Issue