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 af9d695e87 fix(admin): pull API docs once per upstream event and skip
the lock when already refreshed (#7872)
af9d695e87 is described below
commit af9d695e8776f0ece545b0118fef8a39fc05191e
Author: Duansg <[email protected]>
AuthorDate: Tue Oct 20 00:07:42 2026 -0700
fix(admin): pull API docs once per upstream event and skip the lock when
already refreshed (#7872)
---
.../admin/listener/DataChangedEventDispatcher.java | 113 +++++++++++--------
.../manager/impl/PullSwaggerDocServiceImpl.java | 27 +++--
.../listener/DataChangedEventDispatcherTest.java | 124 ++++++++++++++++++++-
.../manager/impl/SwaggerDocManagerImplTest.java | 112 +++++++++++++++++++
4 files changed, 321 insertions(+), 55 deletions(-)
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/DataChangedEventDispatcher.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/DataChangedEventDispatcher.java
index 8fa4edb188..385a1ab3d1 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/DataChangedEventDispatcher.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/DataChangedEventDispatcher.java
@@ -96,53 +96,78 @@ public class DataChangedEventDispatcher implements
ApplicationListener<DataChang
if (!master) {
forwardEventToMaster(event);
}
- for (DataChangedListener listener : listeners) {
- if (!master && !(listener instanceof AbstractDataChangedListener))
{
- // push listeners run on the master only; the event was handed
over above
- continue;
+ boolean loadDoc = false;
+ Throwable listenerFailure = null;
+ try {
+ for (DataChangedListener listener : listeners) {
+ if (!master && !(listener instanceof
AbstractDataChangedListener)) {
+ // push listeners run on the master only; the event was
handed over above
+ continue;
+ }
+ final int size = event.getSource() instanceof
java.util.Collection
+ ? ((java.util.Collection<?>) event.getSource()).size()
: 1;
+ LOG.info("received DataChangedEvent, group={}, size={},
type={}",
+ event.getGroupKey(), size, event.getEventType());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("received DataChangedEvent payload: {}",
JsonUtils.toJson(event));
+ }
+ switch (event.getGroupKey()) {
+ case APP_AUTH:
+ listener.onAppAuthChanged((List<AppAuthData>)
event.getSource(),
+ event.getEventType(), event.getNamespaceId());
+ break;
+ case PLUGIN:
+ listener.onPluginChanged((List<PluginData>)
event.getSource(),
+ event.getEventType(), event.getNamespaceId());
+ break;
+ case RULE:
+ listener.onRuleChanged((List<RuleData>)
event.getSource(),
+ event.getEventType(), event.getNamespaceId());
+ break;
+ case SELECTOR:
+ listener.onSelectorChanged((List<SelectorData>)
event.getSource(),
+ event.getEventType(), event.getNamespaceId());
+ break;
+ case META_DATA:
+ listener.onMetaDataChanged((List<MetaData>)
event.getSource(), event.getEventType());
+ break;
+ case PROXY_SELECTOR:
+
listener.onProxySelectorChanged((List<ProxySelectorData>) event.getSource(),
event.getEventType());
+ break;
+ case AI_PROXY_API_KEY:
+
listener.onAiProxyApiKeyChanged((List<ProxyApiKeyData>) event.getSource(),
event.getEventType());
+ break;
+ case DISCOVER_UPSTREAM:
+
listener.onDiscoveryUpstreamChanged((List<DiscoverySyncData>) event.getSource(),
+ event.getEventType());
+ loadDoc = true;
+ break;
+ default:
+ throw new IllegalStateException("Unexpected value: " +
event.getGroupKey());
+ }
}
- final int size = event.getSource() instanceof java.util.Collection
- ? ((java.util.Collection<?>) event.getSource()).size() : 1;
- LOG.info("received DataChangedEvent, group={}, size={}, type={}",
- event.getGroupKey(), size, event.getEventType());
- if (LOG.isDebugEnabled()) {
- LOG.debug("received DataChangedEvent payload: {}",
JsonUtils.toJson(event));
+ } catch (Throwable ex) {
+ listenerFailure = ex;
+ throw ex;
+ } finally {
+ // load once per event, even if a later listener failed after the
upstream change was applied
+ if (loadDoc) {
+ loadDocOnUpstreamChanged(event, listenerFailure);
}
- switch (event.getGroupKey()) {
- case APP_AUTH:
- listener.onAppAuthChanged((List<AppAuthData>)
event.getSource(),
- event.getEventType(), event.getNamespaceId());
- break;
- case PLUGIN:
- listener.onPluginChanged((List<PluginData>)
event.getSource(),
- event.getEventType(), event.getNamespaceId());
- break;
- case RULE:
- listener.onRuleChanged((List<RuleData>) event.getSource(),
- event.getEventType(), event.getNamespaceId());
- break;
- case SELECTOR:
- listener.onSelectorChanged((List<SelectorData>)
event.getSource(),
- event.getEventType(), event.getNamespaceId());
- break;
- case META_DATA:
- listener.onMetaDataChanged((List<MetaData>)
event.getSource(), event.getEventType());
- break;
- case PROXY_SELECTOR:
- listener.onProxySelectorChanged((List<ProxySelectorData>)
event.getSource(), event.getEventType());
- break;
- case AI_PROXY_API_KEY:
- listener.onAiProxyApiKeyChanged((List<ProxyApiKeyData>)
event.getSource(), event.getEventType());
- break;
- case DISCOVER_UPSTREAM:
-
listener.onDiscoveryUpstreamChanged((List<DiscoverySyncData>) event.getSource(),
- event.getEventType());
-
applicationContext.getBean(LoadServiceDocEntry.class).loadDocOnUpstreamChanged(
- (List<DiscoverySyncData>) event.getSource(),
event.getEventType());
- break;
- default:
- throw new IllegalStateException("Unexpected value: " +
event.getGroupKey());
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private void loadDocOnUpstreamChanged(final DataChangedEvent event, final
Throwable listenerFailure) {
+ try {
+
applicationContext.getBean(LoadServiceDocEntry.class).loadDocOnUpstreamChanged(
+ (List<DiscoverySyncData>) event.getSource(),
event.getEventType());
+ } catch (Throwable ex) {
+ if (Objects.isNull(listenerFailure)) {
+ throw ex;
}
+ // keep the listener failure as the thrown exception
+ listenerFailure.addSuppressed(ex);
}
}
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/manager/impl/PullSwaggerDocServiceImpl.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/manager/impl/PullSwaggerDocServiceImpl.java
index bbdb392fa5..114f95e395 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/manager/impl/PullSwaggerDocServiceImpl.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/manager/impl/PullSwaggerDocServiceImpl.java
@@ -91,7 +91,6 @@ public class PullSwaggerDocServiceImpl implements
PullSwaggerDocService {
synchronized (interner.intern(instance.getClusterName())) {
tagVO = saveTagVOAndAcquireLock(instance);
if (!canPull(instance, tagVO)) {
- LOG.info("api document has been pulled and cannot be pulled
again，instance: {}", instance.getClusterName());
return;
}
}
@@ -127,16 +126,23 @@ public class PullSwaggerDocServiceImpl implements
PullSwaggerDocService {
}
private boolean canPull(final UpstreamInstance instance, final TagVO
tagVO) {
- boolean canPull = false;
- if (Objects.isNull(tagVO) || Objects.isNull(tagVO.getTagExt()) ||
StringUtils.isEmpty(tagVO.getTagExt().getDocLock())) {
- LOG.info("Unable to obtain lock for {}, retry after {} seconds.",
instance.getClusterName(), DOC_LOCK_EXPIRED_TIME / 1000);
+ if (Objects.isNull(tagVO) || Objects.isNull(tagVO.getTagExt())) {
+ return false;
+ }
+ if (!isRefreshDue(instance, tagVO.getTagExt())) {
+ LOG.info("api document has been pulled and cannot be pulled
again，instance: {}", instance.getClusterName());
return false;
}
- Long cacheLastStartUpTime = tagVO.getTagExt().getRefreshTime();
- if (Objects.isNull(cacheLastStartUpTime) || instance.getStartupTime()
> cacheLastStartUpTime + PULL_MIN_INTERVAL_TIME) {
- canPull = true;
+ if (StringUtils.isEmpty(tagVO.getTagExt().getDocLock())) {
+ LOG.info("Unable to obtain lock for {}, retry after {} seconds.",
instance.getClusterName(), DOC_LOCK_EXPIRED_TIME / 1000);
+ return false;
}
- return canPull;
+ return true;
+ }
+
+ private boolean isRefreshDue(final UpstreamInstance instance, final
TagDO.TagExt tagExt) {
+ Long lastRefreshTime = tagExt.getRefreshTime();
+ return Objects.isNull(lastRefreshTime) || instance.getStartupTime() >
lastRefreshTime + PULL_MIN_INTERVAL_TIME;
}
private TagVO saveTagVOAndAcquireLock(final UpstreamInstance instance) {
@@ -145,6 +151,11 @@ public class PullSwaggerDocServiceImpl implements
PullSwaggerDocService {
TagVO tagVO = tagVOList.get(0);
TagDO.TagExt tagExt = convertTagExt(tagVO.getExt());
tagVO.setTagExt(tagExt);
+ // already refreshed: skip without taking the lock, so nothing has
to be written back
+ if (!isRefreshDue(instance, tagExt)) {
+ tagExt.setDocLock(null);
+ return tagVO;
+ }
if (StringUtils.isNotEmpty(tagExt.getDocLock()) &&
NumberUtils.toLong(tagExt.getDocLock(), 0) > System.currentTimeMillis()) {
tagExt.setDocLock(null);
return tagVO;
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/DataChangedEventDispatcherTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/DataChangedEventDispatcherTest.java
index ba38da6df9..06a310413f 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/DataChangedEventDispatcherTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/DataChangedEventDispatcherTest.java
@@ -28,6 +28,9 @@ import org.apache.shenyu.common.enums.ConfigGroupEnum;
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.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
@@ -43,15 +46,19 @@ import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.stream.Stream;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
-import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -285,7 +292,7 @@ public final class DataChangedEventDispatcherTest {
}
/**
- * onApplicationEvent DISCOVER_UPSTREAM configGroupEnum test case — also
triggers loadDocOnUpstreamChanged.
+ * All discovery listeners receive the event, but documentation is loaded
only once.
*/
@Test
void onApplicationEventWithDiscoverUpstreamConfigGroupTest() {
@@ -297,7 +304,118 @@ public final class DataChangedEventDispatcherTest {
verify(nacosDataChangedListener,
times(1)).onDiscoveryUpstreamChanged(anyList(), any());
verify(websocketDataChangedListener,
times(1)).onDiscoveryUpstreamChanged(anyList(), any());
verify(zookeeperDataChangedListener,
times(1)).onDiscoveryUpstreamChanged(anyList(), any());
- verify(loadServiceDocEntry,
atLeastOnce()).loadDocOnUpstreamChanged(anyList(), any());
+ verify(loadServiceDocEntry,
times(1)).loadDocOnUpstreamChanged(anyList(), any());
+ }
+
+ /**
+ * Documentation is still loaded once when a later listener fails, and the
listener failure is rethrown.
+ */
+ @Test
+ void onApplicationEventDiscoverUpstreamListenerFailureStillLoadsDocTest() {
+ when(clusterProperties.isEnabled()).thenReturn(false);
+
when(applicationContext.getBeansOfType(DataChangedListener.class)).thenReturn(orderedListeners());
+ dataChangedEventDispatcher.afterPropertiesSet();
+ IllegalStateException listenerFailure = new
IllegalStateException("listener failed");
+
doThrow(listenerFailure).when(websocketDataChangedListener).onDiscoveryUpstreamChanged(anyList(),
any());
+ DataChangedEvent dataChangedEvent = new
DataChangedEvent(ConfigGroupEnum.DISCOVER_UPSTREAM, null, new ArrayList<>());
+
+ IllegalStateException thrown =
assertThrows(IllegalStateException.class,
+ () ->
dataChangedEventDispatcher.onApplicationEvent(dataChangedEvent));
+
+ assertSame(listenerFailure, thrown);
+ verify(nacosDataChangedListener,
times(1)).onDiscoveryUpstreamChanged(anyList(), any());
+ verify(loadServiceDocEntry,
times(1)).loadDocOnUpstreamChanged(anyList(), any());
+ }
+
+ /**
+ * A documentation failure does not replace the listener failure; it is
attached as suppressed.
+ */
+ @ParameterizedTest
+ @MethodSource("listenerAndDocFailures")
+ void
onApplicationEventDiscoverUpstreamListenerAndDocFailureKeepsListenerFailureTest(final
Throwable listenerFailure, final Throwable docFailure) {
+ when(clusterProperties.isEnabled()).thenReturn(false);
+
when(applicationContext.getBeansOfType(DataChangedListener.class)).thenReturn(orderedListeners());
+ dataChangedEventDispatcher.afterPropertiesSet();
+
doThrow(listenerFailure).when(websocketDataChangedListener).onDiscoveryUpstreamChanged(anyList(),
any());
+
doThrow(docFailure).when(loadServiceDocEntry).loadDocOnUpstreamChanged(anyList(),
any());
+ DataChangedEvent dataChangedEvent = new
DataChangedEvent(ConfigGroupEnum.DISCOVER_UPSTREAM, null, new ArrayList<>());
+
+ Throwable thrown = assertThrows(listenerFailure.getClass(),
+ () ->
dataChangedEventDispatcher.onApplicationEvent(dataChangedEvent));
+
+ assertSame(listenerFailure, thrown);
+ assertArrayEquals(new Throwable[]{docFailure}, thrown.getSuppressed());
+ verify(loadServiceDocEntry,
times(1)).loadDocOnUpstreamChanged(anyList(), any());
+ }
+
+ /**
+ * Without a listener failure, a documentation failure is propagated as
before.
+ */
+ @ParameterizedTest
+ @MethodSource("docFailures")
+ void onApplicationEventDiscoverUpstreamDocFailurePropagatesTest(final
Throwable docFailure) {
+ when(clusterProperties.isEnabled()).thenReturn(false);
+
doThrow(docFailure).when(loadServiceDocEntry).loadDocOnUpstreamChanged(anyList(),
any());
+ DataChangedEvent dataChangedEvent = new
DataChangedEvent(ConfigGroupEnum.DISCOVER_UPSTREAM, null, new ArrayList<>());
+
+ Throwable thrown = assertThrows(docFailure.getClass(),
+ () ->
dataChangedEventDispatcher.onApplicationEvent(dataChangedEvent));
+
+ assertSame(docFailure, thrown);
+ verify(zookeeperDataChangedListener,
times(1)).onDiscoveryUpstreamChanged(anyList(), any());
+ }
+
+ private static Stream<Arguments> listenerAndDocFailures() {
+ return Stream.of(
+ Arguments.of(new IllegalStateException("listener failed"), new
IllegalArgumentException("doc failed")),
+ Arguments.of(new LinkageError("listener failed"), new
IllegalArgumentException("doc failed")),
+ Arguments.of(new IllegalStateException("listener failed"), new
AssertionError("doc failed")),
+ Arguments.of(new LinkageError("listener failed"), new
AssertionError("doc failed")));
+ }
+
+ private static Stream<Throwable> docFailures() {
+ return Stream.of(new IllegalArgumentException("doc failed"), new
AssertionError("doc failed"));
+ }
+
+ private Map<String, DataChangedListener> orderedListeners() {
+ Map<String, DataChangedListener> listenerMap = new LinkedHashMap<>();
+ listenerMap.put("nacosDataChangedListener", nacosDataChangedListener);
+ listenerMap.put("websocketDataChangedListener",
websocketDataChangedListener);
+ listenerMap.put("zookeeperDataChangedListener",
zookeeperDataChangedListener);
+ return listenerMap;
+ }
+
+ /**
+ * Discovery events do not load documentation when no listeners are
registered.
+ */
+ @Test
+ void onApplicationEventWithoutListenersSkipsDocLoadingTest() {
+ when(clusterProperties.isEnabled()).thenReturn(false);
+
when(applicationContext.getBeansOfType(DataChangedListener.class)).thenReturn(Collections.emptyMap());
+ dataChangedEventDispatcher.afterPropertiesSet();
+ DataChangedEvent dataChangedEvent = new
DataChangedEvent(ConfigGroupEnum.DISCOVER_UPSTREAM, null, new ArrayList<>());
+
+ dataChangedEventDispatcher.onApplicationEvent(dataChangedEvent);
+
+ verify(loadServiceDocEntry,
never()).loadDocOnUpstreamChanged(anyList(), any());
+ }
+
+ /**
+ * Discovery events do not load documentation when all listeners are
skipped on a follower.
+ */
+ @Test
+ void onApplicationEventNotMasterWithoutCacheListenersSkipsDocLoadingTest()
{
+ when(clusterProperties.isEnabled()).thenReturn(true);
+ when(shenyuClusterSelectMasterService.isMaster()).thenReturn(false);
+ when(applicationContext.getBeansOfType(DataChangedListener.class))
+
.thenReturn(Collections.singletonMap("nacosDataChangedListener",
nacosDataChangedListener));
+ dataChangedEventDispatcher.afterPropertiesSet();
+ DataChangedEvent dataChangedEvent = new
DataChangedEvent(ConfigGroupEnum.DISCOVER_UPSTREAM, null, new ArrayList<>());
+
+ dataChangedEventDispatcher.onApplicationEvent(dataChangedEvent);
+
+ verify(nacosDataChangedListener,
never()).onDiscoveryUpstreamChanged(anyList(), any());
+ verify(loadServiceDocEntry,
never()).loadDocOnUpstreamChanged(anyList(), any());
}
/**
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/manager/impl/SwaggerDocManagerImplTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/manager/impl/SwaggerDocManagerImplTest.java
index aa247083b4..b9225faeb4 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/manager/impl/SwaggerDocManagerImplTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/manager/impl/SwaggerDocManagerImplTest.java
@@ -17,7 +17,15 @@
package org.apache.shenyu.admin.service.manager.impl;
+import com.sun.net.httpserver.HttpServer;
+import org.apache.http.HttpStatus;
import org.apache.shenyu.admin.model.bean.UpstreamInstance;
+import org.apache.shenyu.admin.model.entity.TagDO;
+import org.apache.shenyu.admin.model.vo.TagVO;
+import org.apache.shenyu.admin.service.TagService;
+import org.apache.shenyu.admin.service.manager.DocManager;
+import org.apache.shenyu.common.constant.AdminConstants;
+import org.apache.shenyu.common.utils.GsonUtils;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
@@ -26,10 +34,21 @@ import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Collections;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+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;
@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.LENIENT)
@@ -41,10 +60,103 @@ public class SwaggerDocManagerImplTest {
@Mock
private Set<UpstreamInstance> currentServices;
+ @Mock
+ private TagService tagService;
+
+ @Mock
+ private DocManager docManager;
+
@Test
public void testPullApiDocument() {
serviceDocManager.pullApiDocument(currentServices);
verify(currentServices).forEach(any());
}
+ @Test
+ public void testPullApiDocumentSkipsLockWhenAlreadyRefreshed() {
+ UpstreamInstance instance = new UpstreamInstance();
+ instance.setContextPath("/test");
+ instance.setStartupTime(1000L);
+ TagDO.TagExt tagExt = new TagDO.TagExt();
+ tagExt.setRefreshTime(1000L);
+ tagExt.setApiDocMd5("existing-md5");
+ TagVO tagVO = new TagVO();
+ tagVO.setId("tag-id");
+ tagVO.setExt(GsonUtils.getInstance().toJson(tagExt));
+ when(tagService.findByQuery(instance.getContextPath(),
AdminConstants.TAG_ROOT_PARENT_ID))
+ .thenReturn(Collections.singletonList(tagVO));
+
+ for (int i = 0; i < 100; i++) {
+ serviceDocManager.pullApiDocument(instance);
+ }
+
+ verify(tagService, never()).updateTagExt(any(), any());
+ verifyNoInteractions(docManager);
+ }
+
+ @Test
+ public void testPullApiDocumentReleasesExpiredLockAfterHttpFailure()
throws IOException {
+ UpstreamInstance instance = new UpstreamInstance();
+ instance.setContextPath("/test");
+ instance.setIp("127.0.0.1");
+ instance.setStartupTime(31001L);
+ TagDO.TagExt tagExt = new TagDO.TagExt();
+ tagExt.setRefreshTime(1000L);
+ tagExt.setDocLock("1");
+ TagVO tagVO = new TagVO();
+ tagVO.setId("tag-id");
+ tagVO.setExt(GsonUtils.getInstance().toJson(tagExt));
+ when(tagService.findByQuery(instance.getContextPath(),
AdminConstants.TAG_ROOT_PARENT_ID))
+ .thenReturn(Collections.singletonList(tagVO));
+ when(tagService.updateTagExt(eq(tagVO.getId()),
any(TagDO.TagExt.class))).thenAnswer(invocation -> {
+ TagDO.TagExt updatedTagExt = invocation.getArgument(1);
+ // Serialize each write to distinguish persisted changes from
in-memory mutations.
+ tagVO.setExt(GsonUtils.getInstance().toJson(updatedTagExt));
+ return 1;
+ });
+
+ AtomicInteger requests = new AtomicInteger();
+ HttpServer server = HttpServer.create(new
InetSocketAddress("127.0.0.1", 0), 0);
+ try {
+ server.createContext("/test/v3/api-docs", exchange -> {
+ requests.incrementAndGet();
+
exchange.sendResponseHeaders(HttpStatus.SC_SERVICE_UNAVAILABLE, -1);
+ exchange.close();
+ });
+ server.start();
+ instance.setPort(server.getAddress().getPort());
+
+ serviceDocManager.pullApiDocument(instance);
+
+ assertEquals(1, requests.get());
+ verify(tagService, times(2)).updateTagExt(eq(tagVO.getId()),
any(TagDO.TagExt.class));
+ TagDO.TagExt persistedTagExt =
GsonUtils.getInstance().fromJson(tagVO.getExt(), TagDO.TagExt.class);
+ assertNull(persistedTagExt.getDocLock());
+ assertEquals(tagExt.getRefreshTime(),
persistedTagExt.getRefreshTime());
+ verifyNoInteractions(docManager);
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ public void testPullApiDocumentDoesNotReleaseLockHeldByAnotherPull() {
+ UpstreamInstance instance = new UpstreamInstance();
+ instance.setContextPath("/test");
+ instance.setStartupTime(31001L);
+ TagDO.TagExt tagExt = new TagDO.TagExt();
+ tagExt.setRefreshTime(1000L);
+ tagExt.setDocLock(String.valueOf(Long.MAX_VALUE));
+ TagVO tagVO = new TagVO();
+ tagVO.setId("tag-id");
+ tagVO.setExt(GsonUtils.getInstance().toJson(tagExt));
+ when(tagService.findByQuery(instance.getContextPath(),
AdminConstants.TAG_ROOT_PARENT_ID))
+ .thenReturn(Collections.singletonList(tagVO));
+
+ serviceDocManager.pullApiDocument(instance);
+
+ verify(tagService, never()).updateTagExt(any(), any());
+ verifyNoInteractions(docManager);
+ }
+
}
