From f03f374397b9710d16c9d5fd3842ab06cbd944ce Mon Sep 17 00:00:00 2001 From: Albumen Kevin Date: Fri, 30 Dec 2022 15:52:26 +0800 Subject: [PATCH] Enhance Zookeeper ServiceDiscovery update (#11220) --- .github/workflows/build-and-test-pr.yml | 4 +- .../zookeeper/ZookeeperServiceDiscovery.java | 81 ++++++------------- ...ookeeperServiceDiscoveryChangeWatcher.java | 67 ++++++--------- .../zookeeper/util/CuratorFrameworkUtils.java | 59 -------------- 4 files changed, 49 insertions(+), 162 deletions(-) diff --git a/.github/workflows/build-and-test-pr.yml b/.github/workflows/build-and-test-pr.yml index 446b1e6193..d01b0f0ece 100644 --- a/.github/workflows/build-and-test-pr.yml +++ b/.github/workflows/build-and-test-pr.yml @@ -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 diff --git a/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscovery.java b/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscovery.java index f0f2a6fc98..ef7b6daafd 100644 --- a/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscovery.java +++ b/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscovery.java @@ -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 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 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 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; - } } diff --git a/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscoveryChangeWatcher.java b/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscoveryChangeWatcher.java index 681019c58a..f4e5168431 100644 --- a/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscoveryChangeWatcher.java +++ b/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscoveryChangeWatcher.java @@ -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 listeners = new ConcurrentHashSet<>(); +public class ZookeeperServiceDiscoveryChangeWatcher implements ServiceCacheListener { + private final Set listeners = new ConcurrentHashSet<>(); + + private final ServiceCache 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 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 instanceList = zookeeperServiceDiscovery.getInstances(serviceName); - notifier.notify(new ServiceInstancesChangedEvent(serviceName, instanceList)); - } - } + List 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 getCacheInstance() { + return cacheInstance; } public Set 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); } } diff --git a/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/util/CuratorFrameworkUtils.java b/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/util/CuratorFrameworkUtils.java index 93fe405931..178757b5b9 100644 --- a/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/util/CuratorFrameworkUtils.java +++ b/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/util/CuratorFrameworkUtils.java @@ -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(); - } - } - } - - } }