From 4bfeeb2b47284ae04f1fe340ac4c29c096dfb260 Mon Sep 17 00:00:00 2001 From: "ken.lj" Date: Wed, 19 May 2021 10:43:11 +0800 Subject: [PATCH] Continue to notify address on metadata failure (#7796) --- .../src/main/resources/dubbo-migration.yaml | 2 +- .../main/resources/spring/dubbo-consumer.xml | 2 +- .../ServiceInstancesChangedListener.java | 8 +- .../client/migration/MigrationInvoker.java | 145 +++++++++--------- 4 files changed, 78 insertions(+), 79 deletions(-) diff --git a/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/src/main/resources/dubbo-migration.yaml b/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/src/main/resources/dubbo-migration.yaml index b33d32c2a1..3f55a0bf36 100644 --- a/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/src/main/resources/dubbo-migration.yaml +++ b/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/src/main/resources/dubbo-migration.yaml @@ -18,5 +18,5 @@ # key: demo-consumer -step: FORCE_APPLICATION +step: APPLICATION_FIRST threshold: 0.1 \ No newline at end of file diff --git a/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/src/main/resources/spring/dubbo-consumer.xml b/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/src/main/resources/spring/dubbo-consumer.xml index 539a6458cc..481c9b42da 100644 --- a/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/src/main/resources/spring/dubbo-consumer.xml +++ b/dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/src/main/resources/spring/dubbo-consumer.xml @@ -27,7 +27,7 @@ - + diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java index 1517ed7dbd..b8bac8e86a 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java @@ -149,9 +149,9 @@ public class ServiceInstancesChangedListener implements ConditionalEventListener retryFuture = scheduler.schedule(new AddressRefreshRetryTask(retryPermission), 10000, TimeUnit.MILLISECONDS); logger.warn("Address refresh try task submitted."); } - logger.warn("Address refresh failed because of Metadata Server failure, wait for retry or new address refresh event."); - this.revisionToMetadata = newRevisionToMetadata; - return; +// logger.warn("Address refresh failed because of Metadata Server failure, wait for retry or new address refresh event."); +// this.revisionToMetadata = newRevisionToMetadata; +// return; } this.revisionToMetadata = newRevisionToMetadata; @@ -244,7 +244,7 @@ public class ServiceInstancesChangedListener implements ConditionalEventListener if (event instanceof RetryServiceInstancesChangedEvent) { RetryServiceInstancesChangedEvent retryEvent = (RetryServiceInstancesChangedEvent) event; logger.warn("Received address refresh retry event, " + retryEvent.getFailureRecordTime()); - if (retryEvent.getFailureRecordTime() < lastRefreshTime) { + if (retryEvent.getFailureRecordTime() < lastRefreshTime && !hasEmptyMetadata(revisionToMetadata)) { logger.warn("Ignore retry event, event time: " + retryEvent.getFailureRecordTime() + ", last refresh time: " + lastRefreshTime); return true; } diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationInvoker.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationInvoker.java index bc23bf7a36..f1f5ef383b 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationInvoker.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationInvoker.java @@ -17,6 +17,7 @@ package org.apache.dubbo.registry.client.migration; import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.config.ConfigurationUtils; import org.apache.dubbo.common.extension.ExtensionLoader; import org.apache.dubbo.common.logger.Logger; import org.apache.dubbo.common.logger.LoggerFactory; @@ -40,12 +41,14 @@ import org.apache.dubbo.rpc.model.ConsumerModel; import java.util.Set; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import static org.apache.dubbo.common.status.reporter.FrameworkStatusReporter.createConsumptionReport; import static org.apache.dubbo.rpc.cluster.Constants.REFER_KEY; public class MigrationInvoker implements MigrationClusterInvoker { private Logger logger = LoggerFactory.getLogger(MigrationInvoker.class); + private static final String MIGRATION_DELAY_KEY = "dubbo.application.migration.delay"; private URL url; private URL consumerUrl; @@ -169,12 +172,17 @@ public class MigrationInvoker implements MigrationClusterInvoker { if (!forceMigrate) { refreshServiceDiscoveryInvoker(); refreshInterfaceInvoker(); - setListener(invoker, () -> { - this.compareAddresses(serviceDiscoveryInvoker, invoker); - }); - setListener(serviceDiscoveryInvoker, () -> { - this.compareAddresses(serviceDiscoveryInvoker, invoker); - }); + // By the time the task gets scheduled, the address notifications are expected to be finished for both address + // types. Otherwise, migration task will just pick interface invoker. + if (!migrated) { + scheduler.schedule(new MigrationTask(), getDelay(), TimeUnit.MILLISECONDS); + setListener(invoker, () -> { + this.setAvailableInvoker(serviceDiscoveryInvoker, invoker); + }); + setListener(serviceDiscoveryInvoker, () -> { + this.setAvailableInvoker(serviceDiscoveryInvoker, invoker); + }); + } } else { refreshServiceDiscoveryInvoker(); setListener(serviceDiscoveryInvoker, () -> { @@ -189,6 +197,51 @@ public class MigrationInvoker implements MigrationClusterInvoker { } } + private int getDelay() { + int delay = 60000; + String delayStr = ConfigurationUtils.getProperty(MIGRATION_DELAY_KEY); + if (StringUtils.isEmpty(delayStr)) { + return delay; + } + + try { + delay = Integer.parseInt(delayStr); + } catch (Exception e) { + logger.warn("Invalid migration delay param " + delayStr); + } + return delay; + } + + private class MigrationTask implements Runnable { + @Override + public void run() { + if (migrated) { + return; + } + + Set detectors = ExtensionLoader.getExtensionLoader(MigrationAddressComparator.class).getSupportedExtensionInstances(); + if (CollectionUtils.isEmpty(detectors)) { + migrated = true; + destroyInterfaceInvoker(); + return; + } + + if (detectors.stream().allMatch(comparator -> comparator.shouldMigrate(serviceDiscoveryInvoker, invoker, rule))) { + destroyInterfaceInvoker(); + FrameworkStatusReporter.reportConsumptionStatus( + createConsumptionReport(consumerUrl.getServiceInterface(), consumerUrl.getVersion(), consumerUrl.getGroup(), "app_app") + ); + } else { + // by default + destroyServiceDiscoveryInvoker(); + FrameworkStatusReporter.reportConsumptionStatus( + createConsumptionReport(consumerUrl.getServiceInterface(), consumerUrl.getVersion(), consumerUrl.getGroup(), "app_interface") + ); + } + migrated = true; + } + } + private boolean checkMigratingConditionMatch(URL consumerUrl) { Set checkers = ExtensionLoader.getExtensionLoader(PreMigratingConditionChecker.class).getSupportedExtensionInstances(); if (CollectionUtils.isNotEmpty(checkers)) { @@ -332,44 +385,22 @@ public class MigrationInvoker implements MigrationClusterInvoker { private volatile boolean invokersChanged; /** - * Need to know which invoker change triggered this compare. + * Set the available invoker before migration is determined. Interface invoker goes first. + * @param serviceDiscoveryInvoker + * @param invoker */ - private synchronized void compareAddresses(ClusterInvoker serviceDiscoveryInvoker, ClusterInvoker invoker) { + private void setAvailableInvoker(ClusterInvoker serviceDiscoveryInvoker, ClusterInvoker invoker) { this.invokersChanged = true; - if (migrated) { - return; - } - if (serviceDiscoveryInvoker == null || serviceDiscoveryInvoker.isDestroyed()) { - currentAvailableInvoker = invoker; - return; - } else if (invoker == null || invoker.isDestroyed()) { - currentAvailableInvoker = serviceDiscoveryInvoker; - return; - } - Set detectors = ExtensionLoader.getExtensionLoader(MigrationAddressComparator.class).getSupportedExtensionInstances(); - if (detectors != null && detectors.stream().allMatch(migrationDetector -> migrationDetector.shouldMigrate(serviceDiscoveryInvoker, invoker, rule))) { - logger.info("serviceKey:" + invoker.getUrl().getServiceKey() + " switch to APP Level address"); - scheduler.submit(() -> { - if (invoker.getDirectory().isNotificationReceived()) { - destroyInterfaceInvoker(); - migrated = true; - FrameworkStatusReporter.reportConsumptionStatus( - createConsumptionReport(consumerUrl.getServiceInterface(), consumerUrl.getVersion(), consumerUrl.getGroup(), "app_app") - ); - } - }); - } else { - logger.info("serviceKey:" + serviceDiscoveryInvoker.getUrl().getServiceKey() + " switch to Service Level address"); - scheduler.submit(() -> { - if (serviceDiscoveryInvoker.getDirectory().isNotificationReceived()) { - destroyServiceDiscoveryInvoker(); - migrated = true; - FrameworkStatusReporter.reportConsumptionStatus( - createConsumptionReport(consumerUrl.getServiceInterface(), consumerUrl.getVersion(), consumerUrl.getGroup(), "app_interface") - ); - } - }); + if (currentAvailableInvoker == null) { + if (invoker != null && !invoker.isDestroyed() && invoker.hasProxyInvokers()) { + currentAvailableInvoker = invoker; + } else if (serviceDiscoveryInvoker != null && !serviceDiscoveryInvoker.isDestroyed() && serviceDiscoveryInvoker.hasProxyInvokers()) { + currentAvailableInvoker = serviceDiscoveryInvoker; + } + if (currentAvailableInvoker == null) { + currentAvailableInvoker = invoker; + } } } @@ -386,20 +417,6 @@ public class MigrationInvoker implements MigrationClusterInvoker { } } -// protected synchronized void discardServiceDiscoveryInvokerAddress(ClusterInvoker serviceDiscoveryInvoker) { -// if (this.invoker != null) { -// this.currentAvailableInvoker = this.invoker; -// updateConsumerModel(currentAvailableInvoker, serviceDiscoveryInvoker); -// } -// if (serviceDiscoveryInvoker != null && !serviceDiscoveryInvoker.isDestroyed()) { -// if (logger.isDebugEnabled()) { -// List> invokers = serviceDiscoveryInvoker.getDirectory().getAllInvokers(); -// logger.debug("Discarding instance addresses, total size " + (invokers == null ? 0 : invokers.size())); -// } -// serviceDiscoveryInvoker.getDirectory().discordAddresses(); -// } -// } - protected void refreshServiceDiscoveryInvoker() { clearListener(serviceDiscoveryInvoker); if (needRefresh(serviceDiscoveryInvoker)) { @@ -435,24 +452,6 @@ public class MigrationInvoker implements MigrationClusterInvoker { } } -// -// protected synchronized void discardInterfaceInvokerAddress(ClusterInvoker invoker) { -// if (this.serviceDiscoveryInvoker != null) { -// this.currentAvailableInvoker = this.serviceDiscoveryInvoker; -// updateConsumerModel(currentAvailableInvoker, invoker); -// } -// if (invoker != null && !invoker.isDestroyed()) { -// if (logger.isDebugEnabled()) { -// List> invokers = invoker.getDirectory().getAllInvokers(); -// logger.debug("Discarding interface addresses, total address size " + (invokers == null ? 0 : invokers.size())); -// } -// invoker.getDirectory().discordAddresses(); -//// if (invokerDestroyStatus == null) { -//// invokerDestroyStatus = executorService.schedule(new InvokerDestroyTask(), destroyInterval, TimeUnit.MILLISECONDS); -//// } -// } -// } - private void clearListener(ClusterInvoker invoker) { if (invoker == null) return; DynamicDirectory directory = (DynamicDirectory) invoker.getDirectory();