Enhance Zookeeper ServiceDiscovery update (#11220)
This commit is contained in:
GitHub
parent
2c6a933d3a
commit
f03f374397
|
|
@ -198,12 +198,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 -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 -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 -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 -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue