This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new dd8b31240d fix(admin): preserve discovered upstreams across stale
health checks (#7882)
dd8b31240d is described below
commit dd8b31240d7397babad34d6dda2f6e6afe3c94bf
Author: cloudbase-ai <[email protected]>
AuthorDate: Wed Oct 21 10:17:17 2026 +0800
fix(admin): preserve discovered upstreams across stale health checks (#7882)
* fix(admin): preserve discovered upstreams across stale health checks
* fix(admin): address upstream snapshot review feedback
Signed-off-by: cloudbase-ai <[email protected]>
---------
Signed-off-by: cloudbase-ai <[email protected]>
Co-authored-by: cloudbase-ai
<[email protected]>
---
.../DiscoveryDataChangedEventSyncListener.java | 24 ++
.../admin/service/DiscoveryUpstreamService.java | 8 +
.../service/impl/DiscoveryUpstreamServiceImpl.java | 5 +
.../admin/service/impl/UpstreamCheckService.java | 156 ++++++++-----
.../DiscoveryDataChangedEventSyncListenerTest.java | 166 +++++++++++++
.../service/DiscoveryUpstreamServiceTest.java | 8 +
.../service/UpstreamCheckServiceRecoveryTest.java | 3 +-
.../service/UpstreamCheckServiceSyncTest.java | 257 ++++++++++++++++++++-
8 files changed, 563 insertions(+), 64 deletions(-)
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/DiscoveryDataChangedEventSyncListener.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/DiscoveryDataChangedEventSyncListener.java
index 2b7b7c5b93..30778fc842 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/DiscoveryDataChangedEventSyncListener.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/DiscoveryDataChangedEventSyncListener.java
@@ -25,6 +25,8 @@ import org.apache.shenyu.admin.discovery.parse.KeyValueParser;
import org.apache.shenyu.admin.listener.DataChangedEvent;
import org.apache.shenyu.admin.mapper.DiscoveryUpstreamMapper;
import org.apache.shenyu.admin.model.entity.DiscoveryUpstreamDO;
+import org.apache.shenyu.admin.model.enums.EventTypeEnum;
+import
org.apache.shenyu.admin.model.event.discovery.DiscoveryStreamUpdatedEvent;
import org.apache.shenyu.admin.transfer.DiscoveryTransfer;
import org.apache.shenyu.common.dto.DiscoverySyncData;
import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
@@ -36,6 +38,8 @@ import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.support.TransactionSynchronization;
+import
org.springframework.transaction.support.TransactionSynchronizationManager;
import java.sql.Timestamp;
import java.util.Collections;
@@ -135,10 +139,30 @@ public class DiscoveryDataChangedEventSyncListener
implements DataChangedEventLi
}
syncData.setUpstreamDataList(discoveryUpstreamMapper.selectByDiscoveryHandlerId(discoveryHandlerId).stream()
.map(DiscoveryTransfer.INSTANCE::mapToData).collect(Collectors.toList()));
+ if (TransactionSynchronizationManager.isActualTransactionActive()
+ &&
TransactionSynchronizationManager.isSynchronizationActive()) {
+ TransactionSynchronizationManager.registerSynchronization(new
TransactionSynchronization() {
+ @Override
+ public void afterCommit() {
+ refreshHealthSnapshot(syncData);
+ }
+ });
+ } else {
+ refreshHealthSnapshot(syncData);
+ }
DataChangedEvent dataChangedEvent = new
DataChangedEvent(ConfigGroupEnum.DISCOVER_UPSTREAM, DataEventTypeEnum.UPDATE,
Collections.singletonList(syncData));
eventPublisher.publishEvent(dataChangedEvent);
}
+ private void refreshHealthSnapshot(final DiscoverySyncData syncData) {
+ try {
+ eventPublisher.publishEvent(new
DiscoveryStreamUpdatedEvent(syncData, EventTypeEnum.UPDATE));
+ } catch (RuntimeException ex) {
+ // A local refresh failure must not prevent delivery of the
committed discovery snapshot.
+ LOG.warn("Failed to refresh health-check upstreams for selector
{}", syncData.getSelectorId(), ex);
+ }
+ }
+
private DiscoverySyncData buildProxySelectorData(final DiscoverySyncData
discoverySyncData, final String value) {
List<DiscoveryUpstreamData> discoveryUpstreamDTOS =
keyValueParser.parseValue(value);
discoveryUpstreamDTOS.forEach(discoveryUpstreamData -> {
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/DiscoveryUpstreamService.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/DiscoveryUpstreamService.java
index b99a51bda3..e1587419dc 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/DiscoveryUpstreamService.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/DiscoveryUpstreamService.java
@@ -114,6 +114,14 @@ public interface DiscoveryUpstreamService {
*/
List<DiscoveryUpstreamData> findBySelectorId(String selectorId);
+ /**
+ * Whether a selector has authoritative discovery state, including an
empty instance list.
+ *
+ * @param selectorId selector id
+ * @return whether a discovery handler exists
+ */
+ boolean hasDiscoveryHandler(String selectorId);
+
/**
* deleteBySelectorIdAndUrl.
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/DiscoveryUpstreamServiceImpl.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/DiscoveryUpstreamServiceImpl.java
index 0004cb43c0..792ac17d67 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/DiscoveryUpstreamServiceImpl.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/DiscoveryUpstreamServiceImpl.java
@@ -323,6 +323,11 @@ public class DiscoveryUpstreamServiceImpl implements
DiscoveryUpstreamService {
return
discoveryUpstreamDOS.stream().map(DiscoveryTransfer.INSTANCE::mapToData).collect(Collectors.toList());
}
+ @Override
+ public boolean hasDiscoveryHandler(final String selectorId) {
+ return
Objects.nonNull(discoveryHandlerMapper.selectBySelectorId(selectorId));
+ }
+
@Override
@Transactional(rollbackFor = Exception.class)
public void deleteBySelectorIdAndUrl(final String selectorId, final String
url) {
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/UpstreamCheckService.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/UpstreamCheckService.java
index b7715a6fec..c3ed957de3 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/UpstreamCheckService.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/UpstreamCheckService.java
@@ -20,6 +20,7 @@ package org.apache.shenyu.admin.service.impl;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.Striped;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.ListUtils;
import org.apache.commons.lang3.StringUtils;
@@ -49,7 +50,6 @@ import org.apache.shenyu.common.enums.ConfigGroupEnum;
import org.apache.shenyu.common.enums.DataEventTypeEnum;
import org.apache.shenyu.common.enums.PluginEnum;
import org.apache.shenyu.common.utils.GsonUtils;
-import org.apache.shenyu.common.utils.MapUtils;
import org.apache.shenyu.common.utils.UpstreamCheckUtils;
import org.apache.shenyu.register.common.config.ShenyuRegisterCenterConfig;
import org.slf4j.Logger;
@@ -73,6 +73,7 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
import java.util.stream.Collectors;
/**
@@ -85,6 +86,8 @@ public class UpstreamCheckService {
private static final Map<String, List<CommonUpstream>> UPSTREAM_MAP =
Maps.newConcurrentMap();
+ private static final Striped<Lock> UPSTREAM_UPDATE_LOCKS =
Striped.lazyWeakLock(1024);
+
private static final Set<Integer> PENDING_SYNC =
Sets.newConcurrentHashSet();
private static final Set<ZombieUpstream> ZOMBIE_SET =
Sets.newConcurrentHashSet();
@@ -194,7 +197,13 @@ public class UpstreamCheckService {
* @param selectorId the selector id
*/
public static void removeByKey(final String selectorId) {
- UPSTREAM_MAP.remove(selectorId);
+ Lock lock = UPSTREAM_UPDATE_LOCKS.get(selectorId);
+ lock.lock();
+ try {
+ UPSTREAM_MAP.remove(selectorId);
+ } finally {
+ lock.unlock();
+ }
}
/**
@@ -209,7 +218,7 @@ public class UpstreamCheckService {
}
Optional.ofNullable(submitJust(selectorId, commonUpstream))
- .ifPresent(upstreams -> executor.execute(() ->
updateHandler(selectorId, upstreams, upstreams)));
+ .ifPresent(upstreams -> executor.execute(() ->
updateHandler(selectorId, upstreams, upstreams, upstreams)));
}
private List<CommonUpstream> submitJust(final String selectorId, final
CommonUpstream commonUpstream) {
@@ -217,23 +226,23 @@ public class UpstreamCheckService {
return null;
}
- List<CommonUpstream> upstreams =
MapUtils.computeIfAbsent(UPSTREAM_MAP, selectorId, k -> new
CopyOnWriteArrayList<>());
- if (commonUpstream.isStatus()) {
- Optional<CommonUpstream> exists = upstreams.stream().filter(item
-> StringUtils.isNotBlank(item.getUpstreamUrl())
- &&
item.getUpstreamUrl().equals(commonUpstream.getUpstreamUrl())).findFirst();
- if (!exists.isPresent()) {
- upstreams.add(commonUpstream);
+ return UPSTREAM_MAP.compute(selectorId, (key, current) -> {
+ List<CommonUpstream> upstreams = Objects.isNull(current) ? new
CopyOnWriteArrayList<>() : new CopyOnWriteArrayList<>(current);
+ if (commonUpstream.isStatus()) {
+ Optional<CommonUpstream> exists =
upstreams.stream().filter(item -> StringUtils.isNotBlank(item.getUpstreamUrl())
+ &&
item.getUpstreamUrl().equals(commonUpstream.getUpstreamUrl())).findFirst();
+ if (!exists.isPresent()) {
+ upstreams.add(commonUpstream);
+ } else {
+
exists.get().setHealthCheckEnabled(commonUpstream.isHealthCheckEnabled());
+ }
+ PENDING_SYNC.add(commonUpstream.hashCode());
} else {
- CommonUpstream existUpstream = exists.get();
-
existUpstream.setHealthCheckEnabled(commonUpstream.isHealthCheckEnabled());
- LOG.info("upstream host {} is exists.",
commonUpstream.getUpstreamHost());
+ upstreams.removeIf(item -> item.equals(commonUpstream));
+ PENDING_SYNC.add(NumberUtils.INTEGER_ZERO);
}
- PENDING_SYNC.add(commonUpstream.hashCode());
- } else {
- upstreams.removeIf(item -> item.equals(commonUpstream));
- PENDING_SYNC.add(NumberUtils.INTEGER_ZERO);
- }
- return upstreams;
+ return upstreams;
+ });
}
/**
@@ -281,7 +290,13 @@ public class UpstreamCheckService {
if (!REGISTER_TYPE_HTTP.equalsIgnoreCase(registerType)) {
return;
}
- UPSTREAM_MAP.put(selectorId, toThreadSafeList(commonUpstreams));
+ Lock lock = UPSTREAM_UPDATE_LOCKS.get(selectorId);
+ lock.lock();
+ try {
+ UPSTREAM_MAP.put(selectorId, toThreadSafeList(commonUpstreams));
+ } finally {
+ lock.unlock();
+ }
}
private void scheduled() {
@@ -349,8 +364,9 @@ public class UpstreamCheckService {
}
private void check(final String selectorId, final List<CommonUpstream>
upstreamList) {
- final List<CompletableFuture<CommonUpstream>> checkFutures = new
ArrayList<>(upstreamList.size());
- for (CommonUpstream commonUpstream : upstreamList) {
+ final List<CommonUpstream> checkedUpstreams = new
ArrayList<>(upstreamList);
+ final List<CompletableFuture<CommonUpstream>> checkFutures = new
ArrayList<>(checkedUpstreams.size());
+ for (CommonUpstream commonUpstream : checkedUpstreams) {
checkFutures.add(CompletableFuture.supplyAsync(() -> {
if (!commonUpstream.isHealthCheckEnabled()) {
if (!commonUpstream.isStatus()) {
@@ -388,22 +404,44 @@ public class UpstreamCheckService {
.map(CompletableFuture::join)
.filter(Objects::nonNull)
.collect(Collectors.toList());
- updateHandler(selectorId, upstreamList, successList);
+ updateHandler(selectorId, checkedUpstreams, successList,
upstreamList);
}));
}
private void updateHandler(final String selectorId, final
List<CommonUpstream> upstreamList, final List<CommonUpstream> successList) {
+ updateHandler(selectorId, upstreamList, successList, null);
+ }
+
+ private void updateHandler(final String selectorId, final
List<CommonUpstream> upstreamList,
+ final List<CommonUpstream> successList, final
List<CommonUpstream> expectedUpstreams) {
//No node changes, including zombie node resurrection and live node
death
if (successList.size() == upstreamList.size() &&
PENDING_SYNC.isEmpty()) {
return;
}
- removePendingSync(successList);
- if (!successList.isEmpty()) {
- UPSTREAM_MAP.put(selectorId, toThreadSafeList(successList));
- updateSelectorHandler(selectorId, successList);
- } else {
- UPSTREAM_MAP.remove(selectorId);
- updateSelectorHandler(selectorId, new ArrayList<>());
+ // Serialize discovery replacement with publication, without holding a
ConcurrentHashMap bin lock during I/O.
+ Lock lock = UPSTREAM_UPDATE_LOCKS.get(selectorId);
+ lock.lock();
+ try {
+ List<CommonUpstream> current = UPSTREAM_MAP.get(selectorId);
+ // A registry replacement or removal supersedes an in-flight
health check.
+ if (Objects.nonNull(expectedUpstreams) && current !=
expectedUpstreams) {
+ return;
+ }
+ removePendingSync(successList);
+ List<CommonUpstream> alive = new ArrayList<>(successList);
+ if (Objects.nonNull(expectedUpstreams)) {
+ current.stream().filter(upstream ->
upstreamList.stream().noneMatch(checkedUpstream ->
+
checkedUpstream.getUpstreamUrl().equals(upstream.getUpstreamUrl())))
+ .filter(CommonUpstream::isStatus).forEach(alive::add);
+ }
+ Set<String> checkedFailedUrls =
upstreamList.stream().map(CommonUpstream::getUpstreamUrl).collect(Collectors.toSet());
+ successList.forEach(upstream ->
checkedFailedUrls.remove(upstream.getUpstreamUrl()));
+ List<CommonUpstream> reconciled =
updateSelectorHandler(selectorId, alive, checkedFailedUrls);
+ // A registration can update the cache during publication; retain
it for its queued update.
+ UPSTREAM_MAP.compute(selectorId, (key, latest) -> latest == current
+ ? (reconciled.isEmpty() ? null :
toThreadSafeList(reconciled)) : latest);
+ } finally {
+ lock.unlock();
}
}
@@ -416,18 +454,39 @@ public class UpstreamCheckService {
return upstreams instanceof CopyOnWriteArrayList ? upstreams : new
CopyOnWriteArrayList<>(upstreams);
}
- private void updateSelectorHandler(final String selectorId, final
List<CommonUpstream> aliveList) {
+ private List<CommonUpstream> updateSelectorHandler(final String
selectorId, final List<CommonUpstream> aliveList, final Set<String>
checkedFailedUrls) {
SelectorDO selectorDO = selectorMapper.selectById(selectorId);
if (Objects.isNull(selectorDO)) {
- return;
+ return aliveList;
}
PluginDO pluginDO = pluginMapper.selectById(selectorDO.getPluginId());
if (Objects.isNull(pluginDO)) {
- return;
+ return aliveList;
+ }
+
+ List<DiscoveryUpstreamData> discoveryUpstreamDataList =
discoveryUpstreamService.findBySelectorId(selectorId);
+ if (CollectionUtils.isEmpty(discoveryUpstreamDataList) &&
!discoveryUpstreamService.hasDiscoveryHandler(selectorId)) {
+ // Legacy registrations without discovery records still use their
checked upstreams.
+ discoveryUpstreamDataList =
aliveList.stream().map(DiscoveryTransfer.INSTANCE::mapToDiscoveryUpstreamData).collect(Collectors.toList());
}
+ Set<String> aliveUrls =
aliveList.stream().map(CommonUpstream::getUpstreamUrl).collect(Collectors.toSet());
+ discoveryUpstreamDataList.stream().filter(upstream ->
aliveUrls.contains(upstream.getUrl())).forEach(upstream -> {
+ LOG.info("change alive selectorId={}|url={}", selectorId,
upstream.getUrl());
+
discoveryUpstreamService.changeStatusBySelectorIdAndUrl(selectorId,
upstream.getUrl(), Boolean.TRUE);
+ upstream.setStatus(NumberUtils.INTEGER_ZERO);
+ });
+ // Unchecked instances retain their persisted status; a stale check is
not a full discovery snapshot.
+ discoveryUpstreamDataList = discoveryUpstreamDataList.stream()
+ .filter(upstream -> upstream.getStatus() == 0 &&
!checkedFailedUrls.contains(upstream.getUrl())).collect(Collectors.toList());
+ List<CommonUpstream> currentAlive =
discoveryUpstreamDataList.stream().map(upstream -> {
+ CommonUpstream common = aliveList.stream().filter(alive ->
alive.getUpstreamUrl().equals(upstream.getUrl())).findFirst()
+ .orElseGet(() ->
DiscoveryTransfer.INSTANCE.mapToCommonUpstream(upstream));
+ common.setStatus(true);
+ return common;
+ }).collect(Collectors.toList());
String pluginName = pluginDO.getName();
- String handler =
converterFactor.newInstance(pluginName).handler(selectorDO.getHandle(),
aliveList);
+ String handler =
converterFactor.newInstance(pluginName).handler(selectorDO.getHandle(),
currentAlive);
selectorDO.setHandle(handler);
selectorMapper.updateSelective(selectorDO);
@@ -440,28 +499,6 @@ public class UpstreamCheckService {
eventPublisher.publishEvent(new
DataChangedEvent(ConfigGroupEnum.SELECTOR, DataEventTypeEnum.UPDATE,
Collections.singletonList(selectorData)));
// publish discovery change event.
- List<DiscoveryUpstreamData> discoveryUpstreamDataList =
discoveryUpstreamService.findBySelectorId(selectorId);
-
- if (CollectionUtils.isEmpty(discoveryUpstreamDataList)) {
- discoveryUpstreamDataList =
aliveList.stream().map(DiscoveryTransfer.INSTANCE::mapToDiscoveryUpstreamData).collect(Collectors.toList());
- }
-
- discoveryUpstreamDataList.removeIf(u -> {
- for (CommonUpstream alive : aliveList) {
- if (alive.getUpstreamUrl().equals(u.getUrl())) {
- return false;
- }
- }
- return true;
- });
- // change live node status to TRUE
- discoveryUpstreamDataList.forEach(upstream -> {
- LOG.info("change alive selectorId={}|url={}", selectorId,
upstream.getUrl());
-
discoveryUpstreamService.changeStatusBySelectorIdAndUrl(selectorId,
upstream.getUrl(), Boolean.TRUE);
- // The snapshot was read before the database update; publish the
same healthy status.
- upstream.setStatus(NumberUtils.INTEGER_ZERO);
- });
-
DiscoverySyncData discoverySyncData = new DiscoverySyncData();
discoverySyncData.setUpstreamDataList(discoveryUpstreamDataList);
discoverySyncData.setPluginName(pluginName);
@@ -470,6 +507,7 @@ public class UpstreamCheckService {
discoverySyncData.setNamespaceId(selectorDO.getNamespaceId());
LOG.debug("UpstreamCacheManager update selectorId={}|json={}",
selectorId, GsonUtils.getGson().toJson(discoverySyncData));
eventPublisher.publishEvent(new
DataChangedEvent(ConfigGroupEnum.DISCOVER_UPSTREAM, DataEventTypeEnum.UPDATE,
Collections.singletonList(discoverySyncData)));
+ return currentAlive;
}
/**
@@ -499,7 +537,7 @@ public class UpstreamCheckService {
.collect(Collectors.toList()));
}
if (CollectionUtils.isNotEmpty(commonUpstreams)) {
- UPSTREAM_MAP.put(selectorDO.getId(), commonUpstreams);
+ replace(selectorDO.getId(), commonUpstreams);
PENDING_SYNC.add(NumberUtils.INTEGER_ZERO);
}
});
@@ -516,7 +554,11 @@ public class UpstreamCheckService {
LOG.info("onDiscoveryUpstreamUpdated plugin={}|list={}",
discoverySyncData.getPluginName(), discoverySyncData.getUpstreamDataList());
if
(PluginEnum.DIVIDE.getName().equals(discoverySyncData.getPluginName())) {
List<DiscoveryUpstreamData> upstreamDataList =
discoverySyncData.getUpstreamDataList();
- List<CommonUpstream> collect =
upstreamDataList.stream().map(DiscoveryTransfer.INSTANCE::mapToCommonUpstream).collect(Collectors.toList());
+ List<CommonUpstream> collect =
upstreamDataList.stream().map(upstream -> {
+ CommonUpstream common =
DiscoveryTransfer.INSTANCE.mapToCommonUpstream(upstream);
+ common.setStatus(upstream.getStatus() == 0);
+ return common;
+ }).collect(Collectors.toList());
List<CommonUpstream> commonUpstreams =
CommonUpstreamUtils.convertCommonUpstreamList(collect);
LOG.info("UpstreamCacheManager replace selectorId={}|json={}",
discoverySyncData.getSelectorId(), GsonUtils.getGson().toJson(commonUpstreams));
replace(discoverySyncData.getSelectorId(), commonUpstreams);
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/discovery/DiscoveryDataChangedEventSyncListenerTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/discovery/DiscoveryDataChangedEventSyncListenerTest.java
index 1a4286f533..945aecd5ba 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/discovery/DiscoveryDataChangedEventSyncListenerTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/discovery/DiscoveryDataChangedEventSyncListenerTest.java
@@ -17,13 +17,21 @@
package org.apache.shenyu.admin.discovery;
+import org.apache.shenyu.admin.config.properties.ClusterProperties;
import org.apache.shenyu.admin.discovery.listener.DiscoveryDataChangedEvent;
import org.apache.shenyu.admin.discovery.parse.KeyValueParser;
+import org.apache.shenyu.admin.listener.ClusterDataChangedEventForwarder;
import org.apache.shenyu.admin.listener.DataChangedEvent;
+import org.apache.shenyu.admin.listener.DataChangedEventDispatcher;
+import org.apache.shenyu.admin.listener.DataChangedListener;
import org.apache.shenyu.admin.mapper.DiscoveryUpstreamMapper;
import org.apache.shenyu.admin.model.entity.DiscoveryUpstreamDO;
+import org.apache.shenyu.admin.model.enums.EventTypeEnum;
+import
org.apache.shenyu.admin.model.event.discovery.DiscoveryStreamUpdatedEvent;
+import org.apache.shenyu.admin.service.manager.LoadServiceDocEntry;
import org.apache.shenyu.common.dto.DiscoverySyncData;
import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -34,11 +42,18 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.support.TransactionSynchronization;
+import
org.springframework.transaction.support.TransactionSynchronizationManager;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -47,7 +62,12 @@ import java.util.concurrent.TimeUnit;
import static
org.apache.shenyu.common.constant.Constants.SYS_DEFAULT_NAMESPACE_ID;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -76,6 +96,152 @@ public class DiscoveryDataChangedEventSyncListenerTest {
discoveryDataChangedEventSyncListener = new
DiscoveryDataChangedEventSyncListener(eventPublisher, discoveryUpstreamMapper,
keyValueParser, contextInfo, discoveryHandlerId);
}
+ @AfterEach
+ void clearTransaction() {
+ if (TransactionSynchronizationManager.isSynchronizationActive()) {
+ TransactionSynchronizationManager.clearSynchronization();
+ }
+ TransactionSynchronizationManager.setActualTransactionActive(false);
+ }
+
+ @Test
+ void testCommittedRegistryChangeRefreshesCompleteHealthSnapshot() {
+ prepareRegistryChange();
+ DiscoveryUpstreamDO old = new DiscoveryUpstreamDO();
+ old.setUpstreamUrl("127.0.0.1:8080");
+ DiscoveryUpstreamDO replacement = new DiscoveryUpstreamDO();
+ replacement.setUpstreamUrl("127.0.0.1:8081");
+
when(discoveryUpstreamMapper.selectByDiscoveryHandlerId("discoveryHandlerId")).thenReturn(List.of(old,
replacement));
+ TransactionSynchronizationManager.setActualTransactionActive(true);
+ TransactionSynchronizationManager.initSynchronization();
+
+ discoveryDataChangedEventSyncListener.onChange(new
DiscoveryDataChangedEvent("key", "value",
DiscoveryDataChangedEvent.Event.ADDED));
+
+ verify(eventPublisher,
never()).publishEvent(any(DiscoveryStreamUpdatedEvent.class));
+
TransactionSynchronizationManager.getSynchronizations().forEach(TransactionSynchronization::afterCommit);
+ ArgumentCaptor<DiscoveryStreamUpdatedEvent> captor =
ArgumentCaptor.forClass(DiscoveryStreamUpdatedEvent.class);
+ verify(eventPublisher).publishEvent(captor.capture());
+ DiscoverySyncData snapshot = captor.getValue().getDiscoverySyncData();
+ Assertions.assertEquals("selector-1", snapshot.getSelectorId());
+ Assertions.assertEquals(2, snapshot.getUpstreamDataList().size());
+ Assertions.assertEquals("127.0.0.1:8081",
snapshot.getUpstreamDataList().get(1).getUrl());
+ }
+
+ @Test
+ void testRolledBackRegistryChangeDoesNotRefreshHealthSnapshot() {
+ prepareRegistryChange();
+ TransactionSynchronizationManager.setActualTransactionActive(true);
+ TransactionSynchronizationManager.initSynchronization();
+
+ discoveryDataChangedEventSyncListener.onChange(new
DiscoveryDataChangedEvent("key", "value",
DiscoveryDataChangedEvent.Event.ADDED));
+ TransactionSynchronizationManager.getSynchronizations()
+ .forEach(synchronization ->
synchronization.afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK));
+
+ verify(eventPublisher,
never()).publishEvent(any(DiscoveryStreamUpdatedEvent.class));
+ }
+
+ @Test
+ void testHealthRefreshPrecedesCommittedGatewayDispatch() {
+ prepareRegistryChange();
+ List<String> delivered = new ArrayList<>();
+ TransactionSynchronizationManager.setActualTransactionActive(true);
+ TransactionSynchronizationManager.initSynchronization();
+ connectGatewayDispatcher(delivered);
+
+ discoveryDataChangedEventSyncListener.onChange(new
DiscoveryDataChangedEvent("key", "value",
DiscoveryDataChangedEvent.Event.ADDED));
+ Assertions.assertTrue(delivered.isEmpty());
+
TransactionSynchronizationManager.getSynchronizations().forEach(TransactionSynchronization::afterCommit);
+
+ Assertions.assertEquals(List.of("health", "gateway"), delivered);
+ }
+
+ @Test
+ void testHealthRefreshFailureDoesNotSuppressCommittedGatewayDispatch() {
+ prepareRegistryChange();
+ List<String> delivered = new ArrayList<>();
+ TransactionSynchronizationManager.setActualTransactionActive(true);
+ TransactionSynchronizationManager.initSynchronization();
+ connectGatewayDispatcher(delivered);
+ doThrow(new IllegalArgumentException("invalid discovery properties"))
+
.when(eventPublisher).publishEvent(any(DiscoveryStreamUpdatedEvent.class));
+
+ discoveryDataChangedEventSyncListener.onChange(new
DiscoveryDataChangedEvent("key", "value",
DiscoveryDataChangedEvent.Event.ADDED));
+
+ Assertions.assertDoesNotThrow(() ->
TransactionSynchronizationManager.getSynchronizations().forEach(TransactionSynchronization::afterCommit));
+ Assertions.assertEquals(Collections.singletonList("gateway"),
delivered);
+ }
+
+ @Test
+ void testHealthRefreshFailureDoesNotSuppressNonTransactionalGatewayEvent()
{
+ prepareRegistryChange();
+ doThrow(new IllegalArgumentException("invalid discovery properties"))
+
.when(eventPublisher).publishEvent(any(DiscoveryStreamUpdatedEvent.class));
+
+ Assertions.assertDoesNotThrow(() ->
discoveryDataChangedEventSyncListener.onChange(
+ new DiscoveryDataChangedEvent("key", "value",
DiscoveryDataChangedEvent.Event.UPDATED)));
+
+ verify(eventPublisher).publishEvent(any(DataChangedEvent.class));
+ }
+
+ @SuppressWarnings("unchecked")
+ private void connectGatewayDispatcher(final List<String> delivered) {
+ ApplicationContext context = mock(ApplicationContext.class);
+ DataChangedListener gateway = mock(DataChangedListener.class);
+
when(context.getBeansOfType(DataChangedListener.class)).thenReturn(Map.of("gateway",
gateway));
+
when(context.getBean(LoadServiceDocEntry.class)).thenReturn(mock(LoadServiceDocEntry.class));
+ ObjectProvider<ClusterDataChangedEventForwarder> forwarder =
mock(ObjectProvider.class);
+
when(context.getBeanProvider(ClusterDataChangedEventForwarder.class)).thenReturn(forwarder);
+ DataChangedEventDispatcher dispatcher = new
DataChangedEventDispatcher(context);
+ ReflectionTestUtils.setField(dispatcher, "clusterProperties", new
ClusterProperties());
+ dispatcher.afterPropertiesSet();
+ doAnswer(invocation -> {
+ delivered.add("gateway");
+ return null;
+
}).when(gateway).onDiscoveryUpstreamChanged(org.mockito.ArgumentMatchers.anyList(),
any());
+ doAnswer(invocation -> {
+ ApplicationEvent event = invocation.getArgument(0);
+ if (event instanceof DataChangedEvent) {
+ dispatcher.onApplicationEvent((DataChangedEvent) event);
+ } else if (event instanceof DiscoveryStreamUpdatedEvent) {
+ delivered.add("health");
+ }
+ return null;
+ }).when(eventPublisher).publishEvent(any(ApplicationEvent.class));
+ }
+
+ @Test
+ void testDeletingLastRegistryUpstreamRefreshesEmptySnapshot() {
+ prepareRegistryChange();
+
+ discoveryDataChangedEventSyncListener.onChange(new
DiscoveryDataChangedEvent("key", "value",
DiscoveryDataChangedEvent.Event.DELETED));
+
+ ArgumentCaptor<DiscoveryStreamUpdatedEvent> captor =
ArgumentCaptor.forClass(DiscoveryStreamUpdatedEvent.class);
+ org.mockito.InOrder order = inOrder(eventPublisher);
+ order.verify(eventPublisher).publishEvent(captor.capture());
+ order.verify(eventPublisher).publishEvent(any(DataChangedEvent.class));
+
Assertions.assertTrue(captor.getValue().getDiscoverySyncData().getUpstreamDataList().isEmpty());
+ Assertions.assertEquals(EventTypeEnum.UPDATE,
captor.getValue().getType());
+ }
+
+ @Test
+ void testRegistryUpdateRefreshesHealthSnapshotWithoutTransaction() {
+ prepareRegistryChange();
+
+ discoveryDataChangedEventSyncListener.onChange(new
DiscoveryDataChangedEvent("key", "value",
DiscoveryDataChangedEvent.Event.UPDATED));
+
+
verify(eventPublisher).publishEvent(any(DiscoveryStreamUpdatedEvent.class));
+ }
+
+ private void prepareRegistryChange() {
+ DiscoveryUpstreamData upstream = new DiscoveryUpstreamData();
+ upstream.setUrl("127.0.0.1:8081");
+
when(keyValueParser.parseValue("value")).thenReturn(Collections.singletonList(upstream));
+
when(contextInfo.getNamespaceId()).thenReturn(SYS_DEFAULT_NAMESPACE_ID);
+ when(contextInfo.getPluginName()).thenReturn("divide");
+ when(contextInfo.getSelectorId()).thenReturn("selector-1");
+
when(contextInfo.getDiscoveryHandlerId()).thenReturn("discoveryHandlerId");
+ }
+
@Test
public void testOnChange() {
final List<DiscoveryUpstreamData> discoveryUpstreamDTOS = new
ArrayList<>();
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/DiscoveryUpstreamServiceTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/DiscoveryUpstreamServiceTest.java
index 0cfd83239b..58628740da 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/DiscoveryUpstreamServiceTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/DiscoveryUpstreamServiceTest.java
@@ -127,6 +127,14 @@ public final class DiscoveryUpstreamServiceTest {
);
}
+ @Test
+ void
testHasDiscoveryHandlerDistinguishesEmptyDiscoveryFromLegacyRegistration() {
+
when(discoveryHandlerMapper.selectBySelectorId("managed")).thenReturn(new
DiscoveryHandlerDO());
+
+
org.junit.jupiter.api.Assertions.assertTrue(discoveryUpstreamService.hasDiscoveryHandler("managed"));
+
org.junit.jupiter.api.Assertions.assertFalse(discoveryUpstreamService.hasDiscoveryHandler("legacy"));
+ }
+
@Test
public void testCreateOrUpdate() {
when(discoveryUpstreamMapper.insert(any())).thenReturn(1);
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/UpstreamCheckServiceRecoveryTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/UpstreamCheckServiceRecoveryTest.java
index 0b327d35cd..f0b773ee51 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/UpstreamCheckServiceRecoveryTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/UpstreamCheckServiceRecoveryTest.java
@@ -121,7 +121,8 @@ class UpstreamCheckServiceRecoveryTest {
ReflectionTestUtils.invokeMethod(service, "checkZombie",
ZombieUpstream.transform(upstream, 1, SELECTOR_ID));
} else {
service.replace(SELECTOR_ID, Collections.singletonList(upstream));
- ReflectionTestUtils.invokeMethod(service, "check", SELECTOR_ID,
Collections.singletonList(upstream));
+ Map<?, ?> upstreams = (Map<?, ?>)
ReflectionTestUtils.getField(UpstreamCheckService.class, "UPSTREAM_MAP");
+ ReflectionTestUtils.invokeMethod(service, "check", SELECTOR_ID,
upstreams.get(SELECTOR_ID));
}
CompletionException exception =
assertThrows(CompletionException.class, () ->
ReflectionTestUtils.invokeMethod(service, "waitFinish"));
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/UpstreamCheckServiceSyncTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/UpstreamCheckServiceSyncTest.java
index b653b25d18..a87294f456 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/UpstreamCheckServiceSyncTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/UpstreamCheckServiceSyncTest.java
@@ -23,33 +23,57 @@ import
org.apache.shenyu.admin.mapper.SelectorConditionMapper;
import org.apache.shenyu.admin.mapper.SelectorMapper;
import org.apache.shenyu.admin.model.entity.PluginDO;
import org.apache.shenyu.admin.model.entity.SelectorDO;
+import org.apache.shenyu.admin.model.enums.EventTypeEnum;
+import
org.apache.shenyu.admin.model.event.discovery.DiscoveryStreamUpdatedEvent;
import org.apache.shenyu.admin.service.converter.DivideSelectorHandleConverter;
import org.apache.shenyu.admin.service.converter.SelectorHandleConverterFactor;
import org.apache.shenyu.admin.service.impl.UpstreamCheckService;
import org.apache.shenyu.common.dto.DiscoverySyncData;
import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
+import org.apache.shenyu.common.dto.convert.selector.CommonUpstream;
import org.apache.shenyu.common.dto.convert.selector.DivideUpstream;
import org.apache.shenyu.common.enums.ConfigGroupEnum;
import org.apache.shenyu.common.enums.PluginEnum;
import org.apache.shenyu.common.utils.GsonUtils;
+import org.apache.shenyu.common.utils.UpstreamCheckUtils;
import org.apache.shenyu.register.common.config.ShenyuRegisterCenterConfig;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
+import org.mockito.MockedStatic;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.test.util.ReflectionTestUtils;
+import java.sql.Timestamp;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
/**
@@ -81,14 +105,127 @@ class UpstreamCheckServiceSyncTest {
void setUp() {
SelectorHandleConverterFactor converters = new
SelectorHandleConverterFactor(
Collections.singletonMap(PluginEnum.DIVIDE.getName(), new
DivideSelectorHandleConverter()));
+ ShenyuRegisterCenterConfig config = new ShenyuRegisterCenterConfig();
+ config.setRegisterType("http");
+ config.getProps().setProperty("checked", "true");
service = new UpstreamCheckService(selectorMapper, eventPublisher,
pluginMapper, selectorConditionMapper,
- new ShenyuRegisterCenterConfig(), converters,
discoveryUpstreamService);
-
when(selectorMapper.selectById(SELECTOR_ID)).thenReturn(SelectorDO.builder().id(SELECTOR_ID)
+ config, converters, discoveryUpstreamService);
+
lenient().when(selectorMapper.selectById(SELECTOR_ID)).thenReturn(SelectorDO.builder().id(SELECTOR_ID)
.selectorName("/divide").namespaceId("namespace").pluginId("divide-id").handle("[]").build());
- when(pluginMapper.selectById("divide-id"))
+ lenient().when(pluginMapper.selectById("divide-id"))
.thenReturn(PluginDO.builder().id("divide-id").name(PluginEnum.DIVIDE.getName()).build());
}
+ @AfterEach
+ void tearDown() {
+ service.close();
+ UpstreamCheckService.removeByKey(SELECTOR_ID);
+ ((Collection<?>)
ReflectionTestUtils.getField(UpstreamCheckService.class,
"PENDING_SYNC")).clear();
+ ((Collection<?>)
ReflectionTestUtils.getField(UpstreamCheckService.class, "ZOMBIE_SET")).clear();
+ }
+
+ @Test
+ void testUncheckedHealthyDiscoveryUpstreamIsPreserved() {
+ DiscoveryUpstreamData offline =
DiscoveryUpstreamData.builder().url("127.0.0.1:8080").status(1).build();
+ DiscoveryUpstreamData replacement =
DiscoveryUpstreamData.builder().id("replacement")
+ .url("127.0.0.1:8081").protocol("http://";).dateCreated(new
Timestamp(1)).status(0).weight(80).props("{}").build();
+
when(discoveryUpstreamService.findBySelectorId(SELECTOR_ID)).thenReturn(new
ArrayList<>(List.of(offline, replacement)));
+
+ ReflectionTestUtils.invokeMethod(service, "updateSelectorHandler",
SELECTOR_ID, Collections.emptyList(), Collections.emptySet());
+
+ DiscoverySyncData snapshot = publishedSnapshot();
+ assertEquals(1, snapshot.getUpstreamDataList().size());
+ assertEquals("replacement",
snapshot.getUpstreamDataList().get(0).getId());
+ assertEquals(80, snapshot.getUpstreamDataList().get(0).getWeight());
+ verify(discoveryUpstreamService,
never()).changeStatusBySelectorIdAndUrl(SELECTOR_ID, replacement.getUrl(),
Boolean.TRUE);
+ ArgumentCaptor<SelectorDO> selectorCaptor =
ArgumentCaptor.forClass(SelectorDO.class);
+ verify(selectorMapper).updateSelective(selectorCaptor.capture());
+
assertTrue(selectorCaptor.getValue().getHandle().contains(replacement.getUrl()));
+ }
+
+ @Test
+ void testPartiallyStaleResultPreservesUncheckedUpstream() {
+ DiscoveryUpstreamData checked =
DiscoveryUpstreamData.builder().url("127.0.0.1:8080").protocol("http://";).status(0).build();
+ DiscoveryUpstreamData replacement =
DiscoveryUpstreamData.builder().url("127.0.0.1:8081").protocol("http://";).dateCreated(new
Timestamp(1)).status(0).build();
+
when(discoveryUpstreamService.findBySelectorId(SELECTOR_ID)).thenReturn(new
ArrayList<>(List.of(checked, replacement)));
+ DivideUpstream alive =
DivideUpstream.builder().upstreamUrl(checked.getUrl()).protocol("http://";).status(true).build();
+
+ ReflectionTestUtils.invokeMethod(service, "updateSelectorHandler",
SELECTOR_ID, Collections.singletonList(alive), Collections.emptySet());
+
+ assertEquals(2, publishedSnapshot().getUpstreamDataList().size());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testLateHealthCheckCannotOverwriteDiscoveryRefresh(final boolean
deleted) {
+ ScheduledThreadPoolExecutor executor =
mock(ScheduledThreadPoolExecutor.class);
+ List<Runnable> requests = new ArrayList<>();
+ doAnswer(invocation ->
requests.add(invocation.getArgument(0))).when(executor).execute(any(Runnable.class));
+ ReflectionTestUtils.setField(service, "invokeExecutor", executor);
+ CommonUpstream old =
DivideUpstream.builder().upstreamUrl("127.0.0.1:8080")
+
.protocol("http://";).status(true).healthCheckEnabled(true).build();
+ service.replace(SELECTOR_ID, Collections.singletonList(old));
+ Map<?, ?> upstreams = (Map<?, ?>)
ReflectionTestUtils.getField(UpstreamCheckService.class, "UPSTREAM_MAP");
+ ReflectionTestUtils.invokeMethod(service, "check", SELECTOR_ID,
upstreams.get(SELECTOR_ID));
+ CommonUpstream replacement =
DivideUpstream.builder().upstreamUrl("127.0.0.1:8081")
+ .protocol("http://";).status(true).build();
+ List<CommonUpstream> latest = deleted ? Collections.emptyList() :
Collections.singletonList(replacement);
+ service.replace(SELECTOR_ID, latest);
+
+ try (MockedStatic<UpstreamCheckUtils> checks =
mockStatic(UpstreamCheckUtils.class)) {
+ checks.when(() ->
UpstreamCheckUtils.checkUrl(old.getUpstreamUrl())).thenReturn(false);
+ requests.forEach(Runnable::run);
+ }
+ ReflectionTestUtils.invokeMethod(service, "waitFinish");
+
+ assertEquals(latest, upstreams.get(SELECTOR_ID));
+ verifyNoInteractions(eventPublisher);
+ }
+
+ @Test
+ void testAuthoritativeEmptySnapshotDoesNotResurrectCheckedUpstream() {
+
when(discoveryUpstreamService.hasDiscoveryHandler(SELECTOR_ID)).thenReturn(true);
+ DivideUpstream alive =
DivideUpstream.builder().upstreamUrl("127.0.0.1:8080").protocol("http://";).status(true).build();
+
+ ReflectionTestUtils.invokeMethod(service, "updateSelectorHandler",
SELECTOR_ID, Collections.singletonList(alive), Collections.emptySet());
+
+ assertTrue(publishedSnapshot().getUpstreamDataList().isEmpty());
+ verify(discoveryUpstreamService,
never()).changeStatusBySelectorIdAndUrl(SELECTOR_ID, alive.getUpstreamUrl(),
Boolean.TRUE);
+ }
+
+ @Test
+ void testReplacementDiscoveredByAnotherAdminRemainsInHealthCache() {
+ DiscoveryUpstreamData replacement =
DiscoveryUpstreamData.builder().url("127.0.0.1:8081").protocol("http://";)
+ .dateCreated(new Timestamp(1)).status(0).build();
+
when(discoveryUpstreamService.findBySelectorId(SELECTOR_ID)).thenReturn(Collections.singletonList(replacement));
+ CommonUpstream old =
DivideUpstream.builder().upstreamUrl("127.0.0.1:8080").protocol("http://";).status(false).build();
+ service.replace(SELECTOR_ID, Collections.singletonList(old));
+
+ ReflectionTestUtils.invokeMethod(service, "updateHandler",
SELECTOR_ID, Collections.singletonList(old), Collections.emptyList());
+
+ assertEquals(replacement.getUrl(),
publishedSnapshot().getUpstreamDataList().get(0).getUrl());
+ @SuppressWarnings("unchecked")
+ Map<String, List<CommonUpstream>> upstreams = (Map<String,
List<CommonUpstream>>) ReflectionTestUtils.getField(UpstreamCheckService.class,
"UPSTREAM_MAP");
+ assertEquals(1, upstreams.get(SELECTOR_ID).size());
+ assertEquals(replacement.getUrl(),
upstreams.get(SELECTOR_ID).get(0).getUpstreamUrl());
+ }
+
+ @Test
+ void testFailedCheckOverridesPersistedHealthyStatus() {
+ DiscoveryUpstreamData failed =
DiscoveryUpstreamData.builder().url("127.0.0.1:8080").protocol("http://";).status(0).build();
+ DiscoveryUpstreamData replacement =
DiscoveryUpstreamData.builder().url("127.0.0.1:8081").protocol("http://";)
+ .dateCreated(new Timestamp(1)).status(0).build();
+
when(discoveryUpstreamService.findBySelectorId(SELECTOR_ID)).thenReturn(List.of(failed,
replacement));
+ CommonUpstream old =
DivideUpstream.builder().upstreamUrl(failed.getUrl()).protocol("http://";).status(false).build();
+ service.replace(SELECTOR_ID, Collections.singletonList(old));
+
+ ReflectionTestUtils.invokeMethod(service, "updateHandler",
SELECTOR_ID, Collections.singletonList(old), Collections.emptyList());
+
+ List<DiscoveryUpstreamData> outgoing =
publishedSnapshot().getUpstreamDataList();
+ assertEquals(1, outgoing.size());
+ assertEquals(replacement.getUrl(), outgoing.get(0).getUrl());
+ }
+
@Test
void testRecoveredUpstreamPublishesCurrentStatus() {
DiscoveryUpstreamData recovered =
DiscoveryUpstreamData.builder().id("recovered")
@@ -97,7 +234,7 @@ class UpstreamCheckServiceSyncTest {
when(discoveryUpstreamService.findBySelectorId(SELECTOR_ID)).thenReturn(new
ArrayList<>(List.of(recovered, offline)));
DivideUpstream alive =
DivideUpstream.builder().upstreamUrl(recovered.getUrl()).protocol("http://";).status(true).build();
- ReflectionTestUtils.invokeMethod(service, "updateSelectorHandler",
SELECTOR_ID, Collections.singletonList(alive));
+ ReflectionTestUtils.invokeMethod(service, "updateSelectorHandler",
SELECTOR_ID, Collections.singletonList(alive), Collections.emptySet());
verify(discoveryUpstreamService).changeStatusBySelectorIdAndUrl(SELECTOR_ID,
recovered.getUrl(), Boolean.TRUE);
DiscoverySyncData snapshot = publishedSnapshot();
@@ -114,7 +251,7 @@ class UpstreamCheckServiceSyncTest {
DiscoveryUpstreamData offline =
DiscoveryUpstreamData.builder().url("127.0.0.1:8080").status(1).build();
when(discoveryUpstreamService.findBySelectorId(SELECTOR_ID)).thenReturn(new
ArrayList<>(Collections.singletonList(offline)));
- ReflectionTestUtils.invokeMethod(service, "updateSelectorHandler",
SELECTOR_ID, Collections.emptyList());
+ ReflectionTestUtils.invokeMethod(service, "updateSelectorHandler",
SELECTOR_ID, Collections.emptyList(), Collections.emptySet());
assertTrue(publishedSnapshot().getUpstreamDataList().isEmpty());
}
@@ -123,7 +260,7 @@ class UpstreamCheckServiceSyncTest {
void testUpstreamWithoutDiscoveryRecordPublishesHealthyStatus() {
DivideUpstream alive =
DivideUpstream.builder().upstreamUrl("127.0.0.1:8080").protocol("http://";).status(true).build();
- ReflectionTestUtils.invokeMethod(service, "updateSelectorHandler",
SELECTOR_ID, Collections.singletonList(alive));
+ ReflectionTestUtils.invokeMethod(service, "updateSelectorHandler",
SELECTOR_ID, Collections.singletonList(alive), Collections.emptySet());
DiscoverySyncData snapshot = publishedSnapshot();
assertEquals(1, snapshot.getUpstreamDataList().size());
@@ -131,6 +268,114 @@ class UpstreamCheckServiceSyncTest {
assertEquals(0, snapshot.getUpstreamDataList().get(0).getStatus());
}
+ @Test
+ void testRegistrationCanProceedDuringSlowEventDispatch() throws Exception {
+ CommonUpstream old =
DivideUpstream.builder().upstreamUrl("127.0.0.1:8080").protocol("http://";).status(false).build();
+ CommonUpstream added =
DivideUpstream.builder().upstreamUrl("127.0.0.1:8081").protocol("http://";).status(true).build();
+ service.replace(SELECTOR_ID, Collections.singletonList(old));
+ CountDownLatch publishing = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ DataChangedEvent event = invocation.getArgument(0);
+ if (event.getGroupKey() == ConfigGroupEnum.SELECTOR) {
+ publishing.countDown();
+ assertTrue(release.await(5, TimeUnit.SECONDS));
+ }
+ return null;
+ }).when(eventPublisher).publishEvent(any(DataChangedEvent.class));
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ final Future<?> check = executor.submit(() ->
ReflectionTestUtils.invokeMethod(service, "updateHandler", SELECTOR_ID,
+ Collections.singletonList(old), Collections.emptyList()));
+ assertTrue(publishing.await(5, TimeUnit.SECONDS));
+ Future<?> registration = executor.submit(() ->
ReflectionTestUtils.invokeMethod(service, "submitJust", SELECTOR_ID, added));
+ registration.get(2, TimeUnit.SECONDS);
+ release.countDown();
+ check.get(5, TimeUnit.SECONDS);
+ Map<?, ?> upstreams = (Map<?, ?>)
ReflectionTestUtils.getField(UpstreamCheckService.class, "UPSTREAM_MAP");
+ assertTrue(upstreams.containsKey(SELECTOR_ID));
+ assertTrue(((List<?>) upstreams.get(SELECTOR_ID)).contains(added));
+ } finally {
+ release.countDown();
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+
+ @Test
+ void testQueuedRegistrationCannotPublishSupersededList() {
+ ScheduledThreadPoolExecutor executor =
mock(ScheduledThreadPoolExecutor.class);
+ List<Runnable> updates = new ArrayList<>();
+ doAnswer(invocation ->
updates.add(invocation.getArgument(0))).when(executor).execute(any(Runnable.class));
+ ReflectionTestUtils.setField(service, "executor", executor);
+ CommonUpstream first =
DivideUpstream.builder().upstreamUrl("127.0.0.1:8080").protocol("http://";).status(true).build();
+ CommonUpstream second =
DivideUpstream.builder().upstreamUrl("127.0.0.1:8081").protocol("http://";).status(true).build();
+ service.submit(SELECTOR_ID, first);
+ service.submit(SELECTOR_ID, second);
+
+ updates.get(0).run();
+ verifyNoInteractions(eventPublisher);
+ updates.get(1).run();
+
+ assertEquals(2, publishedSnapshot().getUpstreamDataList().size());
+ }
+
+ @Test
+ void testReentrantDiscoveryRefreshDuringPublicationIsRetained() {
+ CommonUpstream old =
DivideUpstream.builder().upstreamUrl("127.0.0.1:8080").protocol("http://";).status(false).build();
+ CommonUpstream replacement =
DivideUpstream.builder().upstreamUrl("127.0.0.1:8081").protocol("http://";).status(true).build();
+ service.replace(SELECTOR_ID, Collections.singletonList(old));
+ doAnswer(invocation -> {
+ DataChangedEvent event = invocation.getArgument(0);
+ if (event.getGroupKey() == ConfigGroupEnum.DISCOVER_UPSTREAM) {
+ service.replace(SELECTOR_ID,
Collections.singletonList(replacement));
+ }
+ return null;
+ }).when(eventPublisher).publishEvent(any(DataChangedEvent.class));
+
+ ReflectionTestUtils.invokeMethod(service, "updateHandler",
SELECTOR_ID, Collections.singletonList(old), Collections.emptyList());
+
+ Map<?, ?> upstreams = (Map<?, ?>)
ReflectionTestUtils.getField(UpstreamCheckService.class, "UPSTREAM_MAP");
+ assertEquals(Collections.singletonList(replacement),
upstreams.get(SELECTOR_ID));
+ }
+
+ @Test
+ void testMixedHttpRegistrationDoesNotOverrideAuthoritativeDiscovery() {
+ CommonUpstream http =
DivideUpstream.builder().upstreamUrl("127.0.0.1:8080").protocol("http://";).status(true).build();
+ DiscoveryUpstreamData discovered =
DiscoveryUpstreamData.builder().url("127.0.0.1:8081").protocol("http://";)
+ .dateCreated(new Timestamp(1)).status(0).build();
+
when(discoveryUpstreamService.findBySelectorId(SELECTOR_ID)).thenReturn(Collections.singletonList(discovered));
+
+ ReflectionTestUtils.invokeMethod(service, "updateSelectorHandler",
SELECTOR_ID, Collections.singletonList(http), Collections.emptySet());
+
+ List<DiscoveryUpstreamData> outgoing =
publishedSnapshot().getUpstreamDataList();
+ assertEquals(1, outgoing.size());
+ assertEquals(discovered.getUrl(), outgoing.get(0).getUrl());
+ ArgumentCaptor<SelectorDO> selector =
ArgumentCaptor.forClass(SelectorDO.class);
+ verify(selectorMapper).updateSelective(selector.capture());
+
assertTrue(selector.getValue().getHandle().contains(discovered.getUrl()));
+
org.junit.jupiter.api.Assertions.assertFalse(selector.getValue().getHandle().contains(http.getUpstreamUrl()));
+ }
+
+ @Test
+ void testDiscoveryRefreshPreservesPersistedHealthState() {
+ DiscoveryUpstreamData healthy =
DiscoveryUpstreamData.builder().url("127.0.0.1:8080").protocol("http://";)
+ .dateCreated(new Timestamp(1)).status(0).build();
+ DiscoveryUpstreamData unhealthy =
DiscoveryUpstreamData.builder().url("127.0.0.1:8081").protocol("http://";)
+ .dateCreated(new Timestamp(1)).status(1).build();
+ DiscoverySyncData snapshot = new DiscoverySyncData();
+ snapshot.setSelectorId(SELECTOR_ID);
+ snapshot.setPluginName(PluginEnum.DIVIDE.getName());
+ snapshot.setUpstreamDataList(List.of(healthy, unhealthy));
+
+ service.onDiscoveryUpstreamUpdated(new
DiscoveryStreamUpdatedEvent(snapshot, EventTypeEnum.UPDATE));
+
+ @SuppressWarnings("unchecked")
+ Map<String, List<CommonUpstream>> upstreams = (Map<String,
List<CommonUpstream>>) ReflectionTestUtils.getField(UpstreamCheckService.class,
"UPSTREAM_MAP");
+ assertTrue(upstreams.get(SELECTOR_ID).get(0).isStatus());
+
org.junit.jupiter.api.Assertions.assertFalse(upstreams.get(SELECTOR_ID).get(1).isStatus());
+ }
+
private DiscoverySyncData publishedSnapshot() {
ArgumentCaptor<DataChangedEvent> captor =
ArgumentCaptor.forClass(DataChangedEvent.class);
verify(eventPublisher, times(2)).publishEvent(captor.capture());
