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 862f5e37e8 [ISSUE #7743] Isolate destination heartbeats, enforce
deadline, and resolve rebase conflicts (#7762)
862f5e37e8 is described below
commit 862f5e37e894bd8cef85ef540a8629f313e9ecdf
Author: Harsh Raj <[email protected]>
AuthorDate: Thu Oct 22 11:24:52 2026 +0530
[ISSUE #7743] Isolate destination heartbeats, enforce deadline, and resolve
rebase conflicts (#7762)
Co-authored-by: Harsh <[email protected]>
---
.../register/client/beat/HeartbeatListener.java | 190 +++++-
.../client/beat/HeartbeatBoundPortTest.java | 1 +
.../client/beat/HeartbeatListenerTest.java | 635 +++++++++++++--------
3 files changed, 588 insertions(+), 238 deletions(-)
diff --git
a/shenyu-register-center/shenyu-register-client-beat/src/main/java/org/apache/shenyu/register/client/beat/HeartbeatListener.java
b/shenyu-register-center/shenyu-register-client-beat/src/main/java/org/apache/shenyu/register/client/beat/HeartbeatListener.java
index 2fd6fd48bf..b289545e1f 100644
---
a/shenyu-register-center/shenyu-register-client-beat/src/main/java/org/apache/shenyu/register/client/beat/HeartbeatListener.java
+++
b/shenyu-register-center/shenyu-register-client-beat/src/main/java/org/apache/shenyu/register/client/beat/HeartbeatListener.java
@@ -22,6 +22,7 @@ import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.LoadingCache;
import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.MoreExecutors;
import org.apache.commons.lang3.StringUtils;
import org.apache.shenyu.common.concurrent.ShenyuThreadFactory;
import org.apache.shenyu.common.config.ShenyuConfig;
@@ -32,8 +33,8 @@ import org.apache.shenyu.common.utils.BoundServerPort;
import org.apache.shenyu.common.utils.GsonUtils;
import org.apache.shenyu.common.utils.IpUtils;
import org.apache.shenyu.common.utils.SystemInfoUtils;
-import org.apache.shenyu.register.common.utils.RegisterUtils;
import org.apache.shenyu.register.common.dto.InstanceBeatInfoDTO;
+import org.apache.shenyu.register.common.utils.RegisterUtils;
import org.checkerframework.checker.nullness.qual.NonNull;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.slf4j.Logger;
@@ -43,11 +44,19 @@ import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.event.EventListener;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.OptionalInt;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
public class HeartbeatListener {
@@ -57,8 +66,26 @@ public class HeartbeatListener {
private static final int CHECK_PERIOD = 5;
+ private static final int MAX_BEAT_THREADS = 8;
+
+ private static final String HEARTBEAT_TIMEOUT_MS = "heartbeatTimeoutMs";
+
+ private static final String HEARTBEAT_DIRECT_EXECUTION =
"heartbeatDirectExecution";
+
+ private static final long DEFAULT_DEADLINE_MS = 4000L;
+
+ private final boolean directExecution;
+
private ScheduledThreadPoolExecutor executor;
+ private ExecutorService beatExecutor;
+
+ private BeatSender beatSender;
+
+ private final long deadlineMs;
+
+ private final Map<String, AtomicBoolean> inFlight = new
ConcurrentHashMap<>();
+
private final ShenyuConfig shenyuConfig;
private String username;
@@ -82,7 +109,14 @@ public class HeartbeatListener {
private volatile boolean closed;
public HeartbeatListener(final ShenyuBootstrapHeartBeatConfig config,
final ShenyuConfig shenyuConfig) {
- executor = new ScheduledThreadPoolExecutor(1,
ShenyuThreadFactory.create("scheduled-instance-task", false));
+ this(config, shenyuConfig, null, null);
+ }
+
+ public HeartbeatListener(final ShenyuBootstrapHeartBeatConfig config,
+ final ShenyuConfig shenyuConfig,
+ final ExecutorService beatExecutor,
+ final BeatSender beatSender) {
+ this.executor = new ScheduledThreadPoolExecutor(1,
ShenyuThreadFactory.create("scheduled-instance-task", false));
this.shenyuConfig = shenyuConfig;
this.username = config.getProps().getProperty(Constants.USER_NAME);
this.password = config.getProps().getProperty(Constants.PASS_WORD);
@@ -93,6 +127,24 @@ public class HeartbeatListener {
this.password = AesUtils.cbcEncrypt(secretKey, secretIv, password);
}
this.serverList =
Lists.newArrayList(Splitter.on(",").split(config.getServerLists()));
+ String timeoutProp =
config.getProps().getProperty(HEARTBEAT_TIMEOUT_MS);
+ this.deadlineMs = parseDeadlineMs(timeoutProp);
+ String directProp =
config.getProps().getProperty(HEARTBEAT_DIRECT_EXECUTION);
+ this.directExecution = Boolean.parseBoolean(directProp);
+ if (Objects.nonNull(beatExecutor)) {
+ this.beatExecutor = beatExecutor;
+ } else if (this.directExecution) {
+ this.beatExecutor = MoreExecutors.newDirectExecutorService();
+ } else {
+ int poolSize = Math.min(Math.max(1, this.serverList.size()),
MAX_BEAT_THREADS);
+ this.beatExecutor = new ThreadPoolExecutor(
+ poolSize, poolSize,
+ 60L, TimeUnit.SECONDS,
+ new ArrayBlockingQueue<>(this.serverList.size()),
+ ShenyuThreadFactory.create("instance-beat-worker", false),
+ new ThreadPoolExecutor.AbortPolicy());
+ }
+ this.beatSender =
Optional.ofNullable(beatSender).orElseGet(DefaultBeatSender::new);
this.accessToken = Caffeine.newBuilder()
//see
org.apache.shenyu.admin.config.properties.JwtProperties#expiredSeconds
.expireAfterWrite(24L, TimeUnit.HOURS)
@@ -100,7 +152,7 @@ public class HeartbeatListener {
@Override
public @Nullable String load(@NonNull final String server)
{
try {
- Optional<?> login =
RegisterUtils.doLogin(username, password, server.concat(Constants.LOGIN_PATH));
+ Optional<?> login =
HeartbeatListener.this.beatSender.login(username, password,
server.concat(Constants.LOGIN_PATH));
return login.map(String::valueOf).orElse(null);
} catch (Exception e) {
LOG.error("Login admin url :{} is fail, will
retry. cause: {} ", server, e.getMessage());
@@ -111,6 +163,25 @@ public class HeartbeatListener {
executor.scheduleAtFixedRate(this::reportHeartbeat, INITIAL_DELAY,
CHECK_PERIOD, TimeUnit.SECONDS);
}
+ /**
+ * Parse the heartbeat timeout deadline in milliseconds.
+ * Values <= 0 (e.g. -1) disable the per-destination deadline entirely.
+ *
+ * @param timeoutProp timeout property string
+ * @return deadline in milliseconds, or <= 0 if disabled
+ */
+ private long parseDeadlineMs(final String timeoutProp) {
+ if (StringUtils.isBlank(timeoutProp)) {
+ return DEFAULT_DEADLINE_MS;
+ }
+ try {
+ return Long.parseLong(timeoutProp.trim());
+ } catch (NumberFormatException e) {
+ LOG.warn("Invalid heartbeatTimeoutMs property value '{}', falling
back to default {} ms", timeoutProp, DEFAULT_DEADLINE_MS);
+ return DEFAULT_DEADLINE_MS;
+ }
+ }
+
/**
* Record the application listener port. The management server is ignored,
and nothing is reported
* until a positive bound port is observed. Shutdown wins over a late
initialization event.
@@ -154,30 +225,123 @@ public class HeartbeatListener {
private void sendHeartbeat(final InstanceBeatInfoDTO instanceBeatInfoDTO) {
for (String server : serverList) {
- String concat = server.concat(Constants.BEAT_URI_PATH);
+ AtomicBoolean serverInFlight = null;
+ if (!directExecution) {
+ serverInFlight = inFlight.computeIfAbsent(server, k -> new
AtomicBoolean(false));
+ if (!serverInFlight.compareAndSet(false, true)) {
+ LOG.debug("Previous heartbeat to admin url :{} is still in
flight, skipping.", server);
+ continue;
+ }
+ }
+ final AtomicBoolean finalInFlight = serverInFlight;
+ Future<?> future;
try {
- String accessToken = this.accessToken.get(server);
- if (StringUtils.isBlank(accessToken)) {
- throw new NullPointerException("accessToken is null");
+ future = beatExecutor.submit(() -> {
+ try {
+ sendHeartbeatToServer(server, instanceBeatInfoDTO);
+ } finally {
+ if (Objects.nonNull(finalInFlight)) {
+ finalInFlight.set(false);
+ }
+ }
+ });
+ } catch (RejectedExecutionException e) {
+ if (Objects.nonNull(finalInFlight)) {
+ finalInFlight.set(false);
+ }
+ LOG.warn("Heartbeat dispatch rejected for admin url: {}",
server, e);
+ continue;
+ }
+ if (deadlineMs > 0 && Objects.nonNull(executor) &&
!executor.isShutdown()) {
+ try {
+ executor.schedule(() -> {
+ if (!future.isDone()) {
+ // Note: RegisterUtils.doHeartBeat relies on
OkHttp. Thread interruption does not immediately
+ // abort active socket writes on the wire, so
cancel(true) is best-effort cancellation.
+ future.cancel(true);
+ if (Objects.nonNull(finalInFlight)) {
+ finalInFlight.set(false);
+ }
+ LOG.warn("Heartbeat to admin url :{} timed out
after {} ms deadline.", server, deadlineMs);
+ }
+ }, deadlineMs, TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ LOG.warn("Scheduler rejected timeout task for admin url:
{}", server);
}
-
RegisterUtils.doHeartBeat(GsonUtils.getInstance().toJson(instanceBeatInfoDTO),
concat, Constants.HEARTBEAT, accessToken);
- } catch (Exception e) {
- LOG.error("HeartBeat admin url :{} is fail, will retry.",
server, e);
- // This is a periodic reporter, not a failback registration:
retry on the next tick.
}
}
}
+ private void sendHeartbeatToServer(final String server, final
InstanceBeatInfoDTO instanceBeatInfoDTO) {
+ String concat = server.concat(Constants.BEAT_URI_PATH);
+ try {
+ String token = this.accessToken.get(server);
+ if (StringUtils.isBlank(token)) {
+ throw new NullPointerException("accessToken is null");
+ }
+
beatSender.doHeartBeat(GsonUtils.getInstance().toJson(instanceBeatInfoDTO),
concat, Constants.HEARTBEAT, token);
+ } catch (Exception e) {
+ LOG.error("HeartBeat admin url :{} is fail, will retry.", server,
e);
+ // This is a periodic reporter, not a failback registration: retry
on the next tick.
+ }
+ }
+
@EventListener(ContextClosedEvent.class)
public void onShutdown() {
closed = true;
- executor.shutdown();
+ if (Objects.nonNull(executor)) {
+ executor.shutdown();
+ }
+ if (Objects.nonNull(beatExecutor)) {
+ beatExecutor.shutdown();
+ }
try {
- if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
+ if (Objects.nonNull(executor) && !executor.awaitTermination(5,
TimeUnit.SECONDS)) {
executor.shutdownNow();
}
+ if (Objects.nonNull(beatExecutor) &&
!beatExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
+ beatExecutor.shutdownNow();
+ }
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
+
+ public interface BeatSender {
+
+ /**
+ * Login to admin.
+ *
+ * @param username username
+ * @param password password
+ * @param url url
+ * @return login result
+ * @throws Exception exception
+ */
+ Optional<Object> login(String username, String password, String url)
throws Exception;
+
+ /**
+ * Send heartbeat to admin.
+ *
+ * @param json json payload
+ * @param url url
+ * @param type type
+ * @param token token
+ * @throws Exception exception
+ */
+ void doHeartBeat(String json, String url, String type, String token)
throws Exception;
+ }
+
+ public static class DefaultBeatSender implements BeatSender {
+
+ @Override
+ public Optional<Object> login(final String username, final String
password, final String url) throws Exception {
+ return RegisterUtils.doLogin(username, password, url);
+ }
+
+ @Override
+ public void doHeartBeat(final String json, final String url, final
String type, final String token) throws Exception {
+ RegisterUtils.doHeartBeat(json, url, type, token);
+ }
+ }
}
diff --git
a/shenyu-register-center/shenyu-register-client-beat/src/test/java/org/apache/shenyu/register/client/beat/HeartbeatBoundPortTest.java
b/shenyu-register-center/shenyu-register-client-beat/src/test/java/org/apache/shenyu/register/client/beat/HeartbeatBoundPortTest.java
index 642ee875e0..88c4723199 100644
---
a/shenyu-register-center/shenyu-register-client-beat/src/test/java/org/apache/shenyu/register/client/beat/HeartbeatBoundPortTest.java
+++
b/shenyu-register-center/shenyu-register-client-beat/src/test/java/org/apache/shenyu/register/client/beat/HeartbeatBoundPortTest.java
@@ -114,6 +114,7 @@ class HeartbeatBoundPortTest {
Properties props = new Properties();
props.setProperty(Constants.USER_NAME, "admin");
props.setProperty(Constants.PASS_WORD, "123456");
+ props.setProperty("heartbeatDirectExecution", "true");
ShenyuBootstrapHeartBeatConfig config = new
ShenyuBootstrapHeartBeatConfig();
config.setServerLists("http://127.0.0.1:1";);
config.setProps(props);
diff --git
a/shenyu-register-center/shenyu-register-client-beat/src/test/java/org/apache/shenyu/register/client/beat/HeartbeatListenerTest.java
b/shenyu-register-center/shenyu-register-client-beat/src/test/java/org/apache/shenyu/register/client/beat/HeartbeatListenerTest.java
index 2c9c33a374..0d9ab4112f 100644
---
a/shenyu-register-center/shenyu-register-client-beat/src/test/java/org/apache/shenyu/register/client/beat/HeartbeatListenerTest.java
+++
b/shenyu-register-center/shenyu-register-client-beat/src/test/java/org/apache/shenyu/register/client/beat/HeartbeatListenerTest.java
@@ -27,29 +27,41 @@ import
org.apache.shenyu.register.common.utils.RegisterUtils;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.MockedStatic;
-import org.mockito.MockedConstruction;
import org.mockito.ArgumentCaptor;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.boot.web.context.WebServerApplicationContext;
import org.springframework.boot.web.context.WebServerInitializedEvent;
import org.springframework.boot.web.server.WebServer;
+import java.io.IOException;
import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.Map;
import java.util.Optional;
import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.Mockito.never;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
class HeartbeatListenerTest {
@@ -73,6 +85,7 @@ class HeartbeatListenerTest {
props.setProperty(Constants.PASS_WORD, "123456");
props.setProperty(Constants.AES_SECRET_KEY, "");
props.setProperty(Constants.AES_SECRET_IV, "");
+ props.setProperty("heartbeatDirectExecution", "true");
ShenyuBootstrapHeartBeatConfig config = new
ShenyuBootstrapHeartBeatConfig();
config.setServerLists("http://localhost:9095,http://localhost:9096";);
@@ -101,18 +114,19 @@ class HeartbeatListenerTest {
Mockito.verify(executors.constructed().get(0)).scheduleAtFixedRate(periodic.capture(),
Mockito.eq(0L), Mockito.eq(5L), Mockito.eq(TimeUnit.SECONDS));
assertDoesNotThrow(periodic.getValue()::run);
assertDoesNotThrow(periodic.getValue()::run);
- register.verify(() -> RegisterUtils.doHeartBeat(anyString(),
Mockito.eq("http://localhost:9095"; + Constants.BEAT_URI_PATH),
+ register.verify(() -> RegisterUtils.doHeartBeat(anyString(),
Mockito.eq("http://localhost:9095".concat(Constants.BEAT_URI_PATH)),
Mockito.eq(Constants.HEARTBEAT),
Mockito.eq("mock-token")), Mockito.times(2));
- register.verify(() -> RegisterUtils.doHeartBeat(anyString(),
Mockito.eq("http://localhost:9096"; + Constants.BEAT_URI_PATH),
+ register.verify(() -> RegisterUtils.doHeartBeat(anyString(),
Mockito.eq("http://localhost:9096".concat(Constants.BEAT_URI_PATH)),
Mockito.eq(Constants.HEARTBEAT),
Mockito.eq("mock-token")), Mockito.times(2));
}
}
@Test
- void testPeriodicHeartbeatReportsBootstrapInstanceType() {
+ void testPeriodicHeartbeatPublishesBootstrapInstanceType() throws
Exception {
try (MockedConstruction<ScheduledThreadPoolExecutor> executors =
Mockito.mockConstruction(ScheduledThreadPoolExecutor.class);
- MockedStatic<SystemInfoUtils> ignored =
Mockito.mockStatic(SystemInfoUtils.class);
+ MockedStatic<SystemInfoUtils> system =
Mockito.mockStatic(SystemInfoUtils.class);
MockedStatic<RegisterUtils> register =
Mockito.mockStatic(RegisterUtils.class)) {
+
system.when(SystemInfoUtils::getSystemInfo).thenReturn("system-info");
register.when(() -> RegisterUtils.doLogin(anyString(),
anyString(), anyString())).thenReturn(Optional.of("mock-token"));
heartbeatListener = new HeartbeatListener(config, shenyuConfig);
bind(heartbeatListener, null, 8080);
@@ -120,214 +134,29 @@ class HeartbeatListenerTest {
Mockito.verify(executors.constructed().get(0)).scheduleAtFixedRate(periodic.capture(),
Mockito.eq(0L), Mockito.eq(5L), Mockito.eq(TimeUnit.SECONDS));
periodic.getValue().run();
ArgumentCaptor<String> payload =
ArgumentCaptor.forClass(String.class);
- register.verify(() -> RegisterUtils.doHeartBeat(payload.capture(),
anyString(), anyString(), anyString()), Mockito.atLeastOnce());
- InstanceBeatInfoDTO reported =
GsonUtils.getInstance().fromJson(payload.getAllValues().get(0),
InstanceBeatInfoDTO.class);
- assertEquals(InstanceTypeConstants.BOOTSTRAP_INSTANCE_TYPE,
reported.getInstanceType());
- assertEquals("8080", reported.getInstancePort());
- }
- }
-
- @Test
- void testDoesNotReportBeforeTheServerBinds() {
- try (MockedConstruction<ScheduledThreadPoolExecutor> executors =
Mockito.mockConstruction(ScheduledThreadPoolExecutor.class);
- MockedStatic<SystemInfoUtils> ignored =
Mockito.mockStatic(SystemInfoUtils.class);
- MockedStatic<RegisterUtils> register =
Mockito.mockStatic(RegisterUtils.class)) {
- register.when(() -> RegisterUtils.doLogin(anyString(),
anyString(), anyString())).thenReturn(Optional.of("mock-token"));
- heartbeatListener = new HeartbeatListener(config, shenyuConfig);
- ArgumentCaptor<Runnable> periodic =
ArgumentCaptor.forClass(Runnable.class);
-
Mockito.verify(executors.constructed().get(0)).scheduleAtFixedRate(periodic.capture(),
Mockito.eq(0L), Mockito.eq(5L), Mockito.eq(TimeUnit.SECONDS));
- periodic.getValue().run();
- register.verify(() -> RegisterUtils.doHeartBeat(anyString(),
anyString(), anyString(), anyString()), never());
- assertEquals(1, executors.constructed().size());
- }
- }
-
- @Test
- void testReportsBoundPortInsteadOfConfiguredZeroAndIgnoresManagement() {
- try (MockedConstruction<ScheduledThreadPoolExecutor> executors =
Mockito.mockConstruction(ScheduledThreadPoolExecutor.class);
- MockedStatic<SystemInfoUtils> ignored =
Mockito.mockStatic(SystemInfoUtils.class);
- MockedStatic<RegisterUtils> register =
Mockito.mockStatic(RegisterUtils.class)) {
- register.when(() -> RegisterUtils.doLogin(anyString(),
anyString(), anyString())).thenReturn(Optional.of("mock-token"));
- final HeartbeatListener first = new HeartbeatListener(config,
shenyuConfig);
- final HeartbeatListener second = new HeartbeatListener(config,
shenyuConfig);
- bind(first, "management", 19090);
- bind(first, null, 0);
- bind(first, null, 41234);
- bind(second, null, 41235);
- ArgumentCaptor<Runnable> periodic =
ArgumentCaptor.forClass(Runnable.class);
-
Mockito.verify(executors.constructed().get(0)).scheduleAtFixedRate(periodic.capture(),
Mockito.eq(0L), Mockito.eq(5L), Mockito.eq(TimeUnit.SECONDS));
-
Mockito.verify(executors.constructed().get(1)).scheduleAtFixedRate(periodic.capture(),
Mockito.eq(0L), Mockito.eq(5L), Mockito.eq(TimeUnit.SECONDS));
- periodic.getAllValues().get(0).run();
- periodic.getAllValues().get(1).run();
- ArgumentCaptor<String> payload =
ArgumentCaptor.forClass(String.class);
- register.verify(() -> RegisterUtils.doHeartBeat(payload.capture(),
anyString(), anyString(), anyString()), Mockito.atLeast(2));
- InstanceBeatInfoDTO reportedFirst =
GsonUtils.getInstance().fromJson(payload.getAllValues().get(0),
InstanceBeatInfoDTO.class);
- InstanceBeatInfoDTO reportedSecond =
GsonUtils.getInstance().fromJson(payload.getAllValues().get(2),
InstanceBeatInfoDTO.class);
- assertEquals("41234", reportedFirst.getInstancePort());
- assertEquals("41235", reportedSecond.getInstancePort());
- assertNotEquals(reportedFirst.getInstanceIp() + ":" +
reportedFirst.getInstancePort(),
- reportedSecond.getInstanceIp() + ":" +
reportedSecond.getInstancePort());
- assertEquals(2, executors.constructed().size());
- }
- }
-
- @Test
- void testShutdownBeforeInitializationDoesNotReport() {
- try (MockedConstruction<ScheduledThreadPoolExecutor> executors =
Mockito.mockConstruction(ScheduledThreadPoolExecutor.class);
- MockedStatic<RegisterUtils> register =
Mockito.mockStatic(RegisterUtils.class)) {
- register.when(() -> RegisterUtils.doLogin(anyString(),
anyString(), anyString())).thenReturn(Optional.of("mock-token"));
- heartbeatListener = new HeartbeatListener(config, shenyuConfig);
- ArgumentCaptor<Runnable> periodic =
ArgumentCaptor.forClass(Runnable.class);
-
Mockito.verify(executors.constructed().get(0)).scheduleAtFixedRate(periodic.capture(),
Mockito.eq(0L), Mockito.eq(5L), Mockito.eq(TimeUnit.SECONDS));
- heartbeatListener.onShutdown();
- bind(heartbeatListener, null, 9195, true);
- periodic.getValue().run();
- register.verify(() -> RegisterUtils.doHeartBeat(anyString(),
anyString(), anyString(), anyString()), never());
- Mockito.verify(executors.constructed().get(0), Mockito.times(1))
- .scheduleAtFixedRate(Mockito.any(), Mockito.eq(0L),
Mockito.eq(5L), Mockito.eq(TimeUnit.SECONDS));
- assertEquals(1, executors.constructed().size());
- }
- }
-
- private void bind(final HeartbeatListener listener, final String
namespace, final int port) {
- bind(listener, namespace, port, false);
- }
-
- private void bind(final HeartbeatListener listener, final String
namespace, final int port, final boolean lenient) {
- WebServer webServer = Mockito.mock(WebServer.class);
- WebServerApplicationContext applicationContext =
Mockito.mock(WebServerApplicationContext.class);
- WebServerInitializedEvent event =
Mockito.mock(WebServerInitializedEvent.class);
- if (lenient) {
- Mockito.lenient().when(webServer.getPort()).thenReturn(port);
-
Mockito.lenient().when(applicationContext.getServerNamespace()).thenReturn(namespace);
- Mockito.lenient().when(event.getWebServer()).thenReturn(webServer);
-
Mockito.lenient().when(event.getApplicationContext()).thenReturn(applicationContext);
- } else {
- Mockito.when(webServer.getPort()).thenReturn(port);
-
Mockito.when(applicationContext.getServerNamespace()).thenReturn(namespace);
- Mockito.when(event.getWebServer()).thenReturn(webServer);
-
Mockito.when(event.getApplicationContext()).thenReturn(applicationContext);
- }
- listener.onWebServerInitialized(event);
- }
-
- @Test
- void testHeartbeatListenerCreation() {
-
- try (MockedStatic<RegisterUtils> registerUtilsMockedStatic =
Mockito.mockStatic(RegisterUtils.class)) {
- registerUtilsMockedStatic.when(() ->
RegisterUtils.doLogin(anyString(), anyString(), anyString()))
- .thenReturn(Optional.of("mock-token"));
-
- assertDoesNotThrow(() -> {
- heartbeatListener = new HeartbeatListener(config,
shenyuConfig);
- });
-
- assertNotNull(heartbeatListener);
+ register.verify(() -> RegisterUtils.doHeartBeat(payload.capture(),
Mockito.eq("http://localhost:9095".concat(Constants.BEAT_URI_PATH)),
+ Mockito.eq(Constants.HEARTBEAT),
Mockito.eq("mock-token")));
+ InstanceBeatInfoDTO beatInfo =
GsonUtils.getInstance().fromJson(payload.getValue(), InstanceBeatInfoDTO.class);
+ assertEquals(InstanceTypeConstants.BOOTSTRAP_INSTANCE_TYPE,
beatInfo.getInstanceType());
+ assertEquals("8080", beatInfo.getInstancePort());
+ assertEquals("shenyu", beatInfo.getNamespaceId());
+ assertEquals("system-info", beatInfo.getInstanceInfo());
}
}
@Test
- void testHeartbeatListenerWithEncryption() {
-
- Properties props = new Properties();
- props.setProperty(Constants.USER_NAME, "admin");
- props.setProperty(Constants.PASS_WORD, "123456");
- props.setProperty(Constants.AES_SECRET_KEY, "2095132720951327");
- props.setProperty(Constants.AES_SECRET_IV, "6859932669599326");
-
- ShenyuBootstrapHeartBeatConfig configWithEncryption = new
ShenyuBootstrapHeartBeatConfig();
- configWithEncryption.setServerLists("http://localhost:9095";);
- configWithEncryption.setProps(props);
-
- try (MockedStatic<RegisterUtils> registerUtilsMockedStatic =
Mockito.mockStatic(RegisterUtils.class)) {
- registerUtilsMockedStatic.when(() ->
RegisterUtils.doLogin(anyString(), anyString(), anyString()))
- .thenReturn(Optional.of("mock-token"));
-
- assertDoesNotThrow(() -> {
- heartbeatListener = new
HeartbeatListener(configWithEncryption, shenyuConfig);
- });
-
- assertNotNull(heartbeatListener);
- }
- }
-
- @Test
- void testSendHeartbeatSuccess() throws Exception {
-
- try (MockedStatic<RegisterUtils> registerUtilsMockedStatic =
Mockito.mockStatic(RegisterUtils.class)) {
- registerUtilsMockedStatic.when(() ->
RegisterUtils.doLogin(anyString(), anyString(), anyString()))
- .thenReturn(Optional.of("mock-token"));
-
- registerUtilsMockedStatic.when(() ->
RegisterUtils.doHeartBeat(anyString(), anyString(), anyString(), anyString()))
- .then(invocation -> null);
-
- heartbeatListener = new HeartbeatListener(config, shenyuConfig);
-
- // Use reflection to access private method sendHeartbeat
- java.lang.reflect.Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
-
org.apache.shenyu.register.common.dto.InstanceBeatInfoDTO.class);
- sendHeartbeatMethod.setAccessible(true);
-
- org.apache.shenyu.register.common.dto.InstanceBeatInfoDTO beatInfo
=
- new
org.apache.shenyu.register.common.dto.InstanceBeatInfoDTO();
- beatInfo.setInstancePort("8080");
- beatInfo.setInstanceIp("127.0.0.1");
- beatInfo.setNamespaceId("shenyu");
-
- assertDoesNotThrow(() -> {
- try {
- sendHeartbeatMethod.invoke(heartbeatListener, beatInfo);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
-
- // Should be called for both servers in serverList
- registerUtilsMockedStatic.verify(() ->
RegisterUtils.doHeartBeat(anyString(), anyString(), anyString(), anyString()),
- Mockito.times(2));
- }
- }
-
- @Test
- void testSendHeartbeatWithLoginFailure() throws Exception {
-
- try (MockedStatic<RegisterUtils> registerUtilsMockedStatic =
Mockito.mockStatic(RegisterUtils.class)) {
- registerUtilsMockedStatic.when(() ->
RegisterUtils.doLogin(anyString(), anyString(), anyString()))
- .thenReturn(Optional.empty());
-
- heartbeatListener = new HeartbeatListener(config, shenyuConfig);
-
- // Use reflection to access private method sendHeartbeat
- java.lang.reflect.Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
-
org.apache.shenyu.register.common.dto.InstanceBeatInfoDTO.class);
- sendHeartbeatMethod.setAccessible(true);
-
- org.apache.shenyu.register.common.dto.InstanceBeatInfoDTO beatInfo
=
- new
org.apache.shenyu.register.common.dto.InstanceBeatInfoDTO();
-
- assertDoesNotThrow(() ->
sendHeartbeatMethod.invoke(heartbeatListener, beatInfo));
- assertDoesNotThrow(() ->
sendHeartbeatMethod.invoke(heartbeatListener, beatInfo));
- registerUtilsMockedStatic.verify(() ->
RegisterUtils.doHeartBeat(anyString(), anyString(), anyString(), anyString()),
- Mockito.never());
- }
- }
-
- @Test
- void testOnShutdown() throws Exception {
-
+ void testHeartbeatStop() throws Exception {
try (MockedStatic<RegisterUtils> registerUtilsMockedStatic =
Mockito.mockStatic(RegisterUtils.class)) {
registerUtilsMockedStatic.when(() ->
RegisterUtils.doLogin(anyString(), anyString(), anyString()))
.thenReturn(Optional.of("mock-token"));
heartbeatListener = new HeartbeatListener(config, shenyuConfig);
- // Use reflection to access the executor
Field executorField =
HeartbeatListener.class.getDeclaredField("executor");
executorField.setAccessible(true);
ScheduledThreadPoolExecutor executor =
(ScheduledThreadPoolExecutor) executorField.get(heartbeatListener);
- assertNotNull(executor);
- assertTrue(!executor.isShutdown());
+ assertFalse(executor.isShutdown());
heartbeatListener.onShutdown();
@@ -337,14 +166,12 @@ class HeartbeatListenerTest {
@Test
void testConfigurationValues() throws Exception {
-
try (MockedStatic<RegisterUtils> registerUtilsMockedStatic =
Mockito.mockStatic(RegisterUtils.class)) {
registerUtilsMockedStatic.when(() ->
RegisterUtils.doLogin(anyString(), anyString(), anyString()))
.thenReturn(Optional.of("mock-token"));
heartbeatListener = new HeartbeatListener(config, shenyuConfig);
- // Use reflection to check private fields
Field usernameField =
HeartbeatListener.class.getDeclaredField("username");
usernameField.setAccessible(true);
String username = (String) usernameField.get(heartbeatListener);
@@ -367,7 +194,6 @@ class HeartbeatListenerTest {
@Test
void testHeartbeatWithMultipleServersOneFailure() throws Exception {
-
try (MockedStatic<RegisterUtils> registerUtilsMockedStatic =
Mockito.mockStatic(RegisterUtils.class)) {
registerUtilsMockedStatic.when(() ->
RegisterUtils.doLogin(anyString(), anyString(), anyString()))
.thenReturn(Optional.of("mock-token"));
@@ -383,8 +209,7 @@ class HeartbeatListenerTest {
heartbeatListener = new HeartbeatListener(config, shenyuConfig);
- // Use reflection to access private method sendHeartbeat
- java.lang.reflect.Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
+ Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
org.apache.shenyu.register.common.dto.InstanceBeatInfoDTO.class);
sendHeartbeatMethod.setAccessible(true);
@@ -402,32 +227,392 @@ class HeartbeatListenerTest {
}
@Test
- void testHeartbeatWithAllServersFailure() throws Exception {
+ void testHeartbeatRunsOnDedicatedWorkerPool() throws Exception {
+ CountDownLatch latch = new CountDownLatch(2);
+ AtomicInteger executedCount = new AtomicInteger(0);
+
+ HeartbeatListener.BeatSender beatSender =
mock(HeartbeatListener.BeatSender.class);
+ when(beatSender.login(anyString(), anyString(),
anyString())).thenReturn(Optional.of("mock-token"));
+ Mockito.doAnswer(invocation -> {
+ executedCount.incrementAndGet();
+ latch.countDown();
+ assertNotEquals("main", Thread.currentThread().getName());
+ return null;
+ }).when(beatSender).doHeartBeat(anyString(), anyString(), anyString(),
anyString());
- try (MockedStatic<RegisterUtils> registerUtilsMockedStatic =
Mockito.mockStatic(RegisterUtils.class)) {
- registerUtilsMockedStatic.when(() ->
RegisterUtils.doLogin(anyString(), anyString(), anyString()))
- .thenReturn(Optional.of("mock-token"));
+ Properties props = new Properties();
+ props.setProperty(Constants.USER_NAME, "admin");
+ props.setProperty(Constants.PASS_WORD, "123456");
+ ShenyuBootstrapHeartBeatConfig poolConfig = new
ShenyuBootstrapHeartBeatConfig();
+
poolConfig.setServerLists("http://localhost:9095,http://localhost:9096";);
+ poolConfig.setProps(props);
- // All server calls fail
- registerUtilsMockedStatic.when(() ->
RegisterUtils.doHeartBeat(anyString(), anyString(), anyString(), anyString()))
- .thenThrow(new RuntimeException("Connection failed"));
+ heartbeatListener = new HeartbeatListener(poolConfig, shenyuConfig,
null, beatSender);
- heartbeatListener = new HeartbeatListener(config, shenyuConfig);
+ Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
InstanceBeatInfoDTO.class);
+ sendHeartbeatMethod.setAccessible(true);
+ InstanceBeatInfoDTO beatInfo = new InstanceBeatInfoDTO();
- // Use reflection to access private method sendHeartbeat
- java.lang.reflect.Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
-
org.apache.shenyu.register.common.dto.InstanceBeatInfoDTO.class);
- sendHeartbeatMethod.setAccessible(true);
+ sendHeartbeatMethod.invoke(heartbeatListener, beatInfo);
- org.apache.shenyu.register.common.dto.InstanceBeatInfoDTO beatInfo
=
- new
org.apache.shenyu.register.common.dto.InstanceBeatInfoDTO();
+ assertTrue(latch.await(2, TimeUnit.SECONDS));
+ assertEquals(2, executedCount.get());
+ heartbeatListener.onShutdown();
+ }
- // Should throw RuntimeException when all servers fail
+ @Test
+ void testHeartbeatTimeoutDeadlineEnforced() throws Exception {
+ CountDownLatch beatStarted = new CountDownLatch(1);
+ CountDownLatch beatInterrupted = new CountDownLatch(1);
+
+ HeartbeatListener.BeatSender beatSender =
mock(HeartbeatListener.BeatSender.class);
+ when(beatSender.login(anyString(), anyString(),
anyString())).thenReturn(Optional.of("mock-token"));
+ Mockito.doAnswer(invocation -> {
+ beatStarted.countDown();
try {
- sendHeartbeatMethod.invoke(heartbeatListener, beatInfo);
- } catch (Exception e) {
- assertTrue(e.getCause() instanceof RuntimeException);
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ beatInterrupted.countDown();
}
+ return null;
+ }).when(beatSender).doHeartBeat(anyString(), anyString(), anyString(),
anyString());
+
+ Properties props = new Properties();
+ props.setProperty(Constants.USER_NAME, "admin");
+ props.setProperty(Constants.PASS_WORD, "123456");
+ props.setProperty("heartbeatTimeoutMs", "100");
+ ShenyuBootstrapHeartBeatConfig timeoutConfig = new
ShenyuBootstrapHeartBeatConfig();
+ timeoutConfig.setServerLists("http://localhost:9095";);
+ timeoutConfig.setProps(props);
+
+ heartbeatListener = new HeartbeatListener(timeoutConfig, shenyuConfig,
null, beatSender);
+
+ Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
InstanceBeatInfoDTO.class);
+ sendHeartbeatMethod.setAccessible(true);
+ InstanceBeatInfoDTO beatInfo = new InstanceBeatInfoDTO();
+
+ sendHeartbeatMethod.invoke(heartbeatListener, beatInfo);
+
+ assertTrue(beatStarted.await(1, TimeUnit.SECONDS));
+ assertTrue(beatInterrupted.await(2, TimeUnit.SECONDS));
+ heartbeatListener.onShutdown();
+ }
+
+ @Test
+ void testSlowAdminDoesNotBlockHealthyAdmin() throws Exception {
+ CountDownLatch slowStarted = new CountDownLatch(1);
+ CountDownLatch healthyCompleted = new CountDownLatch(1);
+
+ HeartbeatListener.BeatSender beatSender =
mock(HeartbeatListener.BeatSender.class);
+ when(beatSender.login(anyString(), anyString(),
anyString())).thenReturn(Optional.of("mock-token"));
+
+ Mockito.doAnswer(invocation -> {
+ slowStarted.countDown();
+ Thread.sleep(5000);
+ return null;
+ }).when(beatSender).doHeartBeat(anyString(),
eq("http://localhost:9095".concat(Constants.BEAT_URI_PATH)), anyString(),
anyString());
+
+ Mockito.doAnswer(invocation -> {
+ healthyCompleted.countDown();
+ return null;
+ }).when(beatSender).doHeartBeat(anyString(),
eq("http://localhost:9096".concat(Constants.BEAT_URI_PATH)), anyString(),
anyString());
+
+ Properties props = new Properties();
+ props.setProperty(Constants.USER_NAME, "admin");
+ props.setProperty(Constants.PASS_WORD, "123456");
+ props.setProperty("heartbeatTimeoutMs", "500");
+ ShenyuBootstrapHeartBeatConfig isolateConfig = new
ShenyuBootstrapHeartBeatConfig();
+
isolateConfig.setServerLists("http://localhost:9095,http://localhost:9096";);
+ isolateConfig.setProps(props);
+
+ heartbeatListener = new HeartbeatListener(isolateConfig, shenyuConfig,
null, beatSender);
+
+ Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
InstanceBeatInfoDTO.class);
+ sendHeartbeatMethod.setAccessible(true);
+ InstanceBeatInfoDTO beatInfo = new InstanceBeatInfoDTO();
+
+ sendHeartbeatMethod.invoke(heartbeatListener, beatInfo);
+
+ assertTrue(slowStarted.await(1, TimeUnit.SECONDS));
+ assertTrue(healthyCompleted.await(1, TimeUnit.SECONDS));
+ heartbeatListener.onShutdown();
+ }
+
+ @Test
+ void testConcurrentHeartbeatToSameAdminSuppressed() throws Exception {
+ CountDownLatch holdingBeat = new CountDownLatch(1);
+ CountDownLatch releaseBeat = new CountDownLatch(1);
+ AtomicInteger totalDispatches = new AtomicInteger(0);
+
+ HeartbeatListener.BeatSender beatSender =
mock(HeartbeatListener.BeatSender.class);
+ when(beatSender.login(anyString(), anyString(),
anyString())).thenReturn(Optional.of("mock-token"));
+ Mockito.doAnswer(invocation -> {
+ totalDispatches.incrementAndGet();
+ holdingBeat.countDown();
+ releaseBeat.await(2, TimeUnit.SECONDS);
+ return null;
+ }).when(beatSender).doHeartBeat(anyString(), anyString(), anyString(),
anyString());
+
+ Properties props = new Properties();
+ props.setProperty(Constants.USER_NAME, "admin");
+ props.setProperty(Constants.PASS_WORD, "123456");
+ props.setProperty("heartbeatTimeoutMs", "5000");
+ ShenyuBootstrapHeartBeatConfig singleServerConfig = new
ShenyuBootstrapHeartBeatConfig();
+ singleServerConfig.setServerLists("http://localhost:9095";);
+ singleServerConfig.setProps(props);
+
+ heartbeatListener = new HeartbeatListener(singleServerConfig,
shenyuConfig, null, beatSender);
+
+ Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
InstanceBeatInfoDTO.class);
+ sendHeartbeatMethod.setAccessible(true);
+ InstanceBeatInfoDTO beatInfo = new InstanceBeatInfoDTO();
+
+ sendHeartbeatMethod.invoke(heartbeatListener, beatInfo);
+ assertTrue(holdingBeat.await(1, TimeUnit.SECONDS));
+
+ // Second tick while first is still in flight: must be skipped
+ sendHeartbeatMethod.invoke(heartbeatListener, beatInfo);
+
+ assertEquals(1, totalDispatches.get());
+
+ releaseBeat.countDown();
+ heartbeatListener.onShutdown();
+ }
+
+ @Test
+ void testTokenLoadingDoesNotBlockOtherServers() throws Exception {
+ CountDownLatch slowLoginStarted = new CountDownLatch(1);
+ CountDownLatch fastLoginCompleted = new CountDownLatch(1);
+
+ HeartbeatListener.BeatSender beatSender =
mock(HeartbeatListener.BeatSender.class);
+ Mockito.doAnswer(invocation -> {
+ slowLoginStarted.countDown();
+ Thread.sleep(2000);
+ return Optional.of("slow-token");
+ }).when(beatSender).login(anyString(), anyString(),
eq("http://localhost:9095".concat(Constants.LOGIN_PATH)));
+
+ Mockito.doAnswer(invocation -> {
+ fastLoginCompleted.countDown();
+ return Optional.of("fast-token");
+ }).when(beatSender).login(anyString(), anyString(),
eq("http://localhost:9096".concat(Constants.LOGIN_PATH)));
+
+ Properties props = new Properties();
+ props.setProperty(Constants.USER_NAME, "admin");
+ props.setProperty(Constants.PASS_WORD, "123456");
+ ShenyuBootstrapHeartBeatConfig loginConfig = new
ShenyuBootstrapHeartBeatConfig();
+
loginConfig.setServerLists("http://localhost:9095,http://localhost:9096";);
+ loginConfig.setProps(props);
+
+ heartbeatListener = new HeartbeatListener(loginConfig, shenyuConfig,
null, beatSender);
+
+ Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
InstanceBeatInfoDTO.class);
+ sendHeartbeatMethod.setAccessible(true);
+ InstanceBeatInfoDTO beatInfo = new InstanceBeatInfoDTO();
+
+ sendHeartbeatMethod.invoke(heartbeatListener, beatInfo);
+
+ assertTrue(slowLoginStarted.await(1, TimeUnit.SECONDS));
+ assertTrue(fastLoginCompleted.await(1, TimeUnit.SECONDS));
+ heartbeatListener.onShutdown();
+ }
+
+ @Test
+ void testDirectExecutionConfigurationRespected() throws Exception {
+ Properties props = new Properties();
+ props.setProperty(Constants.USER_NAME, "admin");
+ props.setProperty(Constants.PASS_WORD, "123456");
+ props.setProperty("heartbeatDirectExecution", "true");
+
+ ShenyuBootstrapHeartBeatConfig directCfg = new
ShenyuBootstrapHeartBeatConfig();
+ directCfg.setServerLists("http://localhost:9095";);
+ directCfg.setProps(props);
+
+ heartbeatListener = new HeartbeatListener(directCfg, shenyuConfig);
+
+ Field directField =
HeartbeatListener.class.getDeclaredField("directExecution");
+ directField.setAccessible(true);
+ boolean direct = (boolean) directField.get(heartbeatListener);
+ assertTrue(direct);
+
+ Field executorField =
HeartbeatListener.class.getDeclaredField("beatExecutor");
+ executorField.setAccessible(true);
+ Object exec = executorField.get(heartbeatListener);
+ assertNotNull(exec);
+ }
+
+ @Test
+ void testNegativeDeadlineDisablesTimeout() throws Exception {
+ Properties props = new Properties();
+ props.setProperty(Constants.USER_NAME, "admin");
+ props.setProperty(Constants.PASS_WORD, "123456");
+ props.setProperty("heartbeatTimeoutMs", "-1");
+
+ ShenyuBootstrapHeartBeatConfig noTimeoutCfg = new
ShenyuBootstrapHeartBeatConfig();
+ noTimeoutCfg.setServerLists("http://localhost:9095";);
+ noTimeoutCfg.setProps(props);
+
+ heartbeatListener = new HeartbeatListener(noTimeoutCfg, shenyuConfig);
+
+ Field deadlineField =
HeartbeatListener.class.getDeclaredField("deadlineMs");
+ deadlineField.setAccessible(true);
+ long deadline = (long) deadlineField.get(heartbeatListener);
+ assertEquals(-1L, deadline);
+ }
+
+ @Test
+ void testMalformedHeartbeatTimeoutMsFallback() throws Exception {
+ Properties props = new Properties();
+ props.setProperty(Constants.USER_NAME, "admin");
+ props.setProperty(Constants.PASS_WORD, "123456");
+ props.setProperty("heartbeatTimeoutMs", "invalid");
+
+ ShenyuBootstrapHeartBeatConfig cfg = new
ShenyuBootstrapHeartBeatConfig();
+ cfg.setServerLists("http://localhost:9095";);
+ cfg.setProps(props);
+
+ heartbeatListener = new HeartbeatListener(cfg, shenyuConfig);
+
+ Field deadlineField =
HeartbeatListener.class.getDeclaredField("deadlineMs");
+ deadlineField.setAccessible(true);
+ long deadline = (long) deadlineField.get(heartbeatListener);
+ assertEquals(4000L, deadline);
+ }
+
+ @Test
+ void testCancelledQueuedTaskClearsInFlightFlag() throws Exception {
+ CountDownLatch firstBeatStarted = new CountDownLatch(1);
+ CountDownLatch releaseFirstBeat = new CountDownLatch(1);
+ CountDownLatch secondBeatExecuted = new CountDownLatch(1);
+
+ HeartbeatListener.BeatSender beatSender =
mock(HeartbeatListener.BeatSender.class);
+ when(beatSender.login(anyString(), anyString(),
anyString())).thenReturn(Optional.of("mock-token"));
+
+ Mockito.doAnswer(invocation -> {
+ firstBeatStarted.countDown();
+ releaseFirstBeat.await(2, TimeUnit.SECONDS);
+ return null;
+ }).when(beatSender).doHeartBeat(anyString(),
eq("http://localhost:9095".concat(Constants.BEAT_URI_PATH)), anyString(),
anyString());
+
+ Mockito.doAnswer(invocation -> {
+ secondBeatExecuted.countDown();
+ return null;
+ }).when(beatSender).doHeartBeat(anyString(),
eq("http://localhost:9096".concat(Constants.BEAT_URI_PATH)), anyString(),
anyString());
+
+ Properties props = new Properties();
+ props.setProperty(Constants.USER_NAME, "admin");
+ props.setProperty(Constants.PASS_WORD, "123456");
+ props.setProperty("heartbeatTimeoutMs", "100");
+ ShenyuBootstrapHeartBeatConfig queueConfig = new
ShenyuBootstrapHeartBeatConfig();
+
queueConfig.setServerLists("http://localhost:9095,http://localhost:9096";);
+ queueConfig.setProps(props);
+
+ ExecutorService singleThreadExecutor =
Executors.newSingleThreadExecutor();
+ try {
+ heartbeatListener = new HeartbeatListener(queueConfig,
shenyuConfig, singleThreadExecutor, beatSender);
+ Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
InstanceBeatInfoDTO.class);
+ sendHeartbeatMethod.setAccessible(true);
+ InstanceBeatInfoDTO beatInfo = new InstanceBeatInfoDTO();
+
+ sendHeartbeatMethod.invoke(heartbeatListener, beatInfo);
+
+ assertTrue(firstBeatStarted.await(1, TimeUnit.SECONDS));
+
+ // Wait until 9096's deadline (100ms) expires and cancels its
queued task
+ Thread.sleep(250);
+
+ Field inFlightField =
HeartbeatListener.class.getDeclaredField("inFlight");
+ inFlightField.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ Map<String, AtomicBoolean> inFlightMap = (Map<String,
AtomicBoolean>) inFlightField.get(heartbeatListener);
+ assertNotNull(inFlightMap.get("http://localhost:9096";));
+ assertFalse(inFlightMap.get("http://localhost:9096";).get(),
+ "Cancelled queued task must clear the in-flight flag for
that server");
+
+ // Release the worker thread
+ releaseFirstBeat.countDown();
+ Thread.sleep(100);
+
+ // Subsequent heartbeat to 9096 must now proceed and execute
+ sendHeartbeatMethod.invoke(heartbeatListener, beatInfo);
+ assertTrue(secondBeatExecuted.await(1, TimeUnit.SECONDS),
+ "Subsequent heartbeat must execute after queued task was
cancelled");
+ } finally {
+ releaseFirstBeat.countDown();
+ singleThreadExecutor.shutdownNow();
}
}
+
+ @Test
+ void testRejectedTaskClearsInFlightFlag() throws Exception {
+ CountDownLatch workerHolding = new CountDownLatch(1);
+ CountDownLatch releaseWorker = new CountDownLatch(1);
+ CountDownLatch heartbeatExecuted = new CountDownLatch(1);
+
+ HeartbeatListener.BeatSender beatSender =
mock(HeartbeatListener.BeatSender.class);
+ when(beatSender.login(anyString(), anyString(),
anyString())).thenReturn(Optional.of("mock-token"));
+ Mockito.doAnswer(invocation -> {
+ heartbeatExecuted.countDown();
+ return null;
+ }).when(beatSender).doHeartBeat(anyString(), anyString(), anyString(),
anyString());
+
+ ThreadPoolExecutor rejectingPool = new ThreadPoolExecutor(
+ 1, 1, 0, TimeUnit.SECONDS,
+ new SynchronousQueue<>(),
+ new ThreadPoolExecutor.AbortPolicy());
+
+ // Fill the single worker thread so any new submit will be rejected
+ rejectingPool.submit(() -> {
+ workerHolding.countDown();
+ releaseWorker.await(2, TimeUnit.SECONDS);
+ return null;
+ });
+ assertTrue(workerHolding.await(1, TimeUnit.SECONDS));
+
+ Properties props = new Properties();
+ props.setProperty(Constants.USER_NAME, "admin");
+ props.setProperty(Constants.PASS_WORD, "123456");
+ ShenyuBootstrapHeartBeatConfig configSingle = new
ShenyuBootstrapHeartBeatConfig();
+ configSingle.setServerLists("http://localhost:9095";);
+ configSingle.setProps(props);
+
+ try {
+ heartbeatListener = new HeartbeatListener(configSingle,
shenyuConfig, rejectingPool, beatSender);
+ Method sendHeartbeatMethod =
HeartbeatListener.class.getDeclaredMethod("sendHeartbeat",
InstanceBeatInfoDTO.class);
+ sendHeartbeatMethod.setAccessible(true);
+ InstanceBeatInfoDTO beatInfo = new InstanceBeatInfoDTO();
+
+ // Submit will be rejected, but must not throw or leave flag stuck
+ assertDoesNotThrow(() ->
sendHeartbeatMethod.invoke(heartbeatListener, beatInfo));
+
+ Field inFlightField =
HeartbeatListener.class.getDeclaredField("inFlight");
+ inFlightField.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ Map<String, AtomicBoolean> inFlightMap = (Map<String,
AtomicBoolean>) inFlightField.get(heartbeatListener);
+ assertNotNull(inFlightMap.get("http://localhost:9095";));
+ assertFalse(inFlightMap.get("http://localhost:9095";).get(),
+ "Rejected task must clear in-flight flag immediately");
+
+ // Release worker so subsequent heartbeat can succeed
+ releaseWorker.countDown();
+ Thread.sleep(100);
+
+ sendHeartbeatMethod.invoke(heartbeatListener, beatInfo);
+ assertTrue(heartbeatExecuted.await(1, TimeUnit.SECONDS),
+ "Heartbeat must succeed once executor becomes available
again");
+ } finally {
+ releaseWorker.countDown();
+ rejectingPool.shutdownNow();
+ }
+ }
+
+ private void bind(final HeartbeatListener listener, final String
namespace, final int port) {
+ WebServerInitializedEvent event =
Mockito.mock(WebServerInitializedEvent.class);
+ WebServer webServer = Mockito.mock(WebServer.class);
+ WebServerApplicationContext applicationContext =
Mockito.mock(WebServerApplicationContext.class);
+ Mockito.when(event.getWebServer()).thenReturn(webServer);
+
Mockito.when(event.getApplicationContext()).thenReturn(applicationContext);
+ Mockito.when(webServer.getPort()).thenReturn(port);
+
Mockito.when(applicationContext.getServerNamespace()).thenReturn(namespace);
+ listener.onWebServerInitialized(event);
+ }
}
