natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Continue to notify address on metadata failure (#7796)

This commit is contained in:
ken.lj 2021-05-19 10:43:11 +08:00 committed by GitHub
parent 3531a7d681
commit 4bfeeb2b47
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 78 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/src/main/resources/dubbo-migration.yaml
View File

@ -18,5 +18,5 @@
#
key: demo-consumer
step: FORCE_APPLICATION
step: APPLICATION_FIRST
threshold: 0.1

2
dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/src/main/resources/spring/dubbo-consumer.xml
View File

@ -27,7 +27,7 @@
<!-- <dubbo:metadata-report address="zookeeper://127.0.0.1:2181"/>-->
<dubbo:registry address="zookeeper://127.0.0.1:2181?registry-type=service"/>
<dubbo:registry address="zookeeper://127.0.0.1:2181"/>
<dubbo:reference id="demoService" check="false"
interface="org.apache.dubbo.demo.DemoService"/>

8
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
View File

@ -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;
}

145
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationInvoker.java
View File

@ -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<T> implements MigrationClusterInvoker<T> {
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<T> implements MigrationClusterInvoker<T> {
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<T> implements MigrationClusterInvoker<T> {
}
}
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<MigrationAddressComparator> 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<PreMigratingConditionChecker> checkers = ExtensionLoader.getExtensionLoader(PreMigratingConditionChecker.class).getSupportedExtensionInstances();
if (CollectionUtils.isNotEmpty(checkers)) {
@ -332,44 +385,22 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
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<T> serviceDiscoveryInvoker, ClusterInvoker<T> invoker) {
private void setAvailableInvoker(ClusterInvoker<T> serviceDiscoveryInvoker, ClusterInvoker<T> 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<MigrationAddressComparator> 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<T> implements MigrationClusterInvoker<T> {
}
}
// protected synchronized void discardServiceDiscoveryInvokerAddress(ClusterInvoker<T> serviceDiscoveryInvoker) {
// if (this.invoker != null) {
// this.currentAvailableInvoker = this.invoker;
// updateConsumerModel(currentAvailableInvoker, serviceDiscoveryInvoker);
// }
// if (serviceDiscoveryInvoker != null && !serviceDiscoveryInvoker.isDestroyed()) {
// if (logger.isDebugEnabled()) {
// List<Invoker<T>> 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<T> implements MigrationClusterInvoker<T> {
}
}
//
// protected synchronized void discardInterfaceInvokerAddress(ClusterInvoker<T> invoker) {
// if (this.serviceDiscoveryInvoker != null) {
// this.currentAvailableInvoker = this.serviceDiscoveryInvoker;
// updateConsumerModel(currentAvailableInvoker, invoker);
// }
// if (invoker != null && !invoker.isDestroyed()) {
// if (logger.isDebugEnabled()) {
// List<Invoker<T>> 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<T> invoker) {
if (invoker == null) return;
DynamicDirectory<T> directory = (DynamicDirectory<T>) invoker.getDirectory();