This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12161-801e330cbaf69345e644300f8cf00bb8878b4798 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 0a5c2cc2fc3e053192f3ed11e2966f250166c586 Author: Jast <[email protected]> AuthorDate: Tue Sep 22 08:50:49 2026 +0000 [Fix][Zeta] Render Zeta health metrics with a stable locale and bound REST fanout (#12161) Co-authored-by: zhangshenghang <[email protected]> --- docs/en/engines/zeta/hybrid-cluster-deployment.md | 8 ++ docs/en/engines/zeta/rest-api-v1.md | 3 + docs/en/engines/zeta/rest-api-v2.md | 3 + .../engines/zeta/separated-cluster-deployment.md | 8 ++ docs/zh/engines/zeta/hybrid-cluster-deployment.md | 8 ++ docs/zh/engines/zeta/rest-api-v1.md | 3 + docs/zh/engines/zeta/rest-api-v2.md | 3 + .../engines/zeta/separated-cluster-deployment.md | 8 ++ .../engine/common/config/EngineConfig.java | 12 ++ .../config/YamlSeaTunnelDomConfigProcessor.java | 9 ++ .../common/config/server/ServerConfigOptions.java | 9 ++ .../engine/server/SeaTunnelHealthMonitor.java | 18 ++- .../engine/server/rest/service/BaseService.java | 105 +++++++++++--- .../engine/server/SeaTunnelHealthMonitorTest.java | 64 +++++++++ .../rest/service/BaseServiceHealthMetricsTest.java | 158 +++++++++++++++++++++ 15 files changed, 392 insertions(+), 27 deletions(-) diff --git a/docs/en/engines/zeta/hybrid-cluster-deployment.md b/docs/en/engines/zeta/hybrid-cluster-deployment.md index ba6b43aa56..bf53fba998 100644 --- a/docs/en/engines/zeta/hybrid-cluster-deployment.md +++ b/docs/en/engines/zeta/hybrid-cluster-deployment.md @@ -137,6 +137,14 @@ seatunnel: state-cleanup-delay-ms: 60000 ``` +The `/system-monitoring-information` REST API asks every cluster member for its health metrics. All members share one deadline controlled by `health-metrics-timeout-seconds`, whose default value is `3` seconds. A member that does not answer within this deadline is reported with its address and a `timeout` marker instead of blocking the whole response, so the total latency of the API no longer grows with the number of unreachable members. + +```yaml +seatunnel: + engine: + health-metrics-timeout-seconds: 3 +``` + ### 4.5 Class Loader Cache Mode This configuration primarily addresses the issue of resource leakage caused by constantly creating and attempting to destroy the class loader. diff --git a/docs/en/engines/zeta/rest-api-v1.md b/docs/en/engines/zeta/rest-api-v1.md index 7d43d11c92..fff61c8f71 100644 --- a/docs/en/engines/zeta/rest-api-v1.md +++ b/docs/en/engines/zeta/rest-api-v1.md @@ -573,6 +573,9 @@ When we can't get the job info, the response will be: ] ``` +Each member is asked in parallel and awaited against one shared deadline (`seatunnel.engine.health-metrics-timeout-seconds`, default `3` seconds). A member that does not answer in time is reported as `{"host": "10.0.0.1", "port": 5801, "error": "timeout"}`; members whose request dispatch or response fails are reported with the corresponding `error` marker. + + </details> ------------------------------------------------------------------------------------------ diff --git a/docs/en/engines/zeta/rest-api-v2.md b/docs/en/engines/zeta/rest-api-v2.md index 8616c953fd..f49243e325 100644 --- a/docs/en/engines/zeta/rest-api-v2.md +++ b/docs/en/engines/zeta/rest-api-v2.md @@ -890,6 +890,9 @@ starting beyond the end of the result set also returns `400`, while a page start ] ``` +Each member is asked in parallel and awaited against one shared deadline (`seatunnel.engine.health-metrics-timeout-seconds`, default `3` seconds). A member that does not answer in time is reported as `{"host": "10.0.0.1", "port": 5801, "error": "timeout"}`; members whose request dispatch or response fails are reported with the corresponding `error` marker. + + </details> ------------------------------------------------------------------------------------------ diff --git a/docs/en/engines/zeta/separated-cluster-deployment.md b/docs/en/engines/zeta/separated-cluster-deployment.md index 43ebb91e54..e6ca1c0bb7 100644 --- a/docs/en/engines/zeta/separated-cluster-deployment.md +++ b/docs/en/engines/zeta/separated-cluster-deployment.md @@ -202,6 +202,14 @@ seatunnel: state-cleanup-delay-ms: 60000 ``` +The `/system-monitoring-information` REST API asks every cluster member for its health metrics. All members share one deadline controlled by `health-metrics-timeout-seconds`, whose default value is `3` seconds. A member that does not answer within this deadline is reported with its address and a `timeout` marker instead of blocking the whole response, so the total latency of the API no longer grows with the number of unreachable members. + +```yaml +seatunnel: + engine: + health-metrics-timeout-seconds: 3 +``` + ### 4.5 Class Loader Cache Mode This configuration mainly solves the problem of resource leakage caused by continuously creating and attempting to destroy class loaders. diff --git a/docs/zh/engines/zeta/hybrid-cluster-deployment.md b/docs/zh/engines/zeta/hybrid-cluster-deployment.md index 2378ff775c..f17b1d2483 100644 --- a/docs/zh/engines/zeta/hybrid-cluster-deployment.md +++ b/docs/zh/engines/zeta/hybrid-cluster-deployment.md @@ -136,6 +136,14 @@ seatunnel: state-cleanup-delay-ms: 60000 ``` +`/system-monitoring-information` REST API 会向每个集群成员收集健康指标。所有成员共享一个由 `health-metrics-timeout-seconds` 控制的统一截止时间，默认值为 `3` 秒。在截止时间内未应答的成员会以带地址和 `timeout` 标记的条目返回，而不是一直阻塞整个响应，因此该 API 的总耗时不会随失联成员数量而增长。 + +```yaml +seatunnel: + engine: + health-metrics-timeout-seconds: 3 +``` + ### 4.5 类加载器缓存模式 此配置主要解决不断创建和尝试销毁类加载器所导致的资源泄漏问题。 diff --git a/docs/zh/engines/zeta/rest-api-v1.md b/docs/zh/engines/zeta/rest-api-v1.md index d2f4e359a8..207a478a47 100644 --- a/docs/zh/engines/zeta/rest-api-v1.md +++ b/docs/zh/engines/zeta/rest-api-v1.md @@ -569,6 +569,9 @@ network: ] ``` +每个成员的请求会被并行发出，并共享一个统一截止时间（`seatunnel.engine.health-metrics-timeout-seconds`，默认 `3` 秒）。在截止时间内未应答的成员会以 `{"host": "10.0.0.1", "port": 5801, "error": "timeout"}` 的形式返回；请求分发或响应失败时也会带有对应的 `error` 标记。 + + </details> ------------------------------------------------------------------------------------------ diff --git a/docs/zh/engines/zeta/rest-api-v2.md b/docs/zh/engines/zeta/rest-api-v2.md index 9ef191f636..1e88085dce 100644 --- a/docs/zh/engines/zeta/rest-api-v2.md +++ b/docs/zh/engines/zeta/rest-api-v2.md @@ -861,6 +861,9 @@ seatunnel: ] ``` +每个成员的请求会被并行发出，并共享一个统一截止时间（`seatunnel.engine.health-metrics-timeout-seconds`，默认 `3` 秒）。在截止时间内未应答的成员会以 `{"host": "10.0.0.1", "port": 5801, "error": "timeout"}` 的形式返回；请求分发或响应失败时也会带有对应的 `error` 标记。 + + </details> ------------------------------------------------------------------------------------------ diff --git a/docs/zh/engines/zeta/separated-cluster-deployment.md b/docs/zh/engines/zeta/separated-cluster-deployment.md index 126f7c2ab3..7796bd2b72 100644 --- a/docs/zh/engines/zeta/separated-cluster-deployment.md +++ b/docs/zh/engines/zeta/separated-cluster-deployment.md @@ -207,6 +207,14 @@ seatunnel: state-cleanup-delay-ms: 60000 ``` +`/system-monitoring-information` REST API 会向每个集群成员收集健康指标。所有成员共享一个由 `health-metrics-timeout-seconds` 控制的统一截止时间，默认值为 `3` 秒。在截止时间内未应答的成员会以带地址和 `timeout` 标记的条目返回，而不是一直阻塞整个响应，因此该 API 的总耗时不会随失联成员数量而增长。 + +```yaml +seatunnel: + engine: + health-metrics-timeout-seconds: 3 +``` + ### 4.5 类加载器缓存模式 此配置主要解决不断创建和尝试销毁类加载器所导致的资源泄漏问题。 diff --git a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/EngineConfig.java b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/EngineConfig.java index 82b7592575..a3173207a4 100644 --- a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/EngineConfig.java +++ b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/EngineConfig.java @@ -107,6 +107,10 @@ public class EngineConfig { private HttpConfig httpConfig = ServerConfigOptions.MasterServerConfigOptions.HTTP.defaultValue(); + private int healthMetricsTimeoutSeconds = + ServerConfigOptions.MasterServerConfigOptions.HEALTH_METRICS_TIMEOUT_SECONDS + .defaultValue(); + /** * Stain trace sampling and persistence knobs used by source, transform, sink, and reporters. */ @@ -158,6 +162,14 @@ public class EngineConfig { this.printJobMetricsInfoInterval = printJobMetricsInfoInterval; } + public void setHealthMetricsTimeoutSeconds(int healthMetricsTimeoutSeconds) { + checkPositive( + healthMetricsTimeoutSeconds, + ServerConfigOptions.MasterServerConfigOptions.HEALTH_METRICS_TIMEOUT_SECONDS + + " must be > 0"); + this.healthMetricsTimeoutSeconds = healthMetricsTimeoutSeconds; + } + public void setJobMetricsBackupInterval(int jobMetricsBackupInterval) { checkPositive( jobMetricsBackupInterval, diff --git a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelDomConfigProcessor.java b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelDomConfigProcessor.java index f2826a8e0a..1d3218bb0d 100644 --- a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelDomConfigProcessor.java +++ b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelDomConfigProcessor.java @@ -235,6 +235,15 @@ public class YamlSeaTunnelDomConfigProcessor extends AbstractDomConfigProcessor .key() .equals(name)) { engineConfig.setStateCleanupDelayMillis(Long.parseLong(getTextContent(node))); + } else if (ServerConfigOptions.MasterServerConfigOptions.HEALTH_METRICS_TIMEOUT_SECONDS + .key() + .equals(name)) { + engineConfig.setHealthMetricsTimeoutSeconds( + getIntegerValue( + ServerConfigOptions.MasterServerConfigOptions + .HEALTH_METRICS_TIMEOUT_SECONDS + .key(), + getTextContent(node))); } else if (ServerConfigOptions.MasterServerConfigOptions.CONNECTOR_JAR_STORAGE_CONFIG .key() .equals(name)) { diff --git a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/server/ServerConfigOptions.java b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/server/ServerConfigOptions.java index 903e1d9e1a..d3a23c5c22 100644 --- a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/server/ServerConfigOptions.java +++ b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/server/ServerConfigOptions.java @@ -191,6 +191,15 @@ public class ServerConfigOptions { // The options about Hazelcast IMAP store end ///////////////////////////////////////////////// + public static final Option<Integer> HEALTH_METRICS_TIMEOUT_SECONDS = + Options.key("health-metrics-timeout-seconds") + .intType() + .defaultValue(3) + .withDescription( + "The shared deadline (in seconds) for collecting health metrics from all cluster members " + + "for the system-monitoring-information REST API. Members that do not answer in time " + + "are reported with a timeout marker instead of blocking the response."); + ///////////////////////////////////////////////// // The options for checkpoint start public static final Option<Integer> CHECKPOINT_INTERVAL = diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/SeaTunnelHealthMonitor.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/SeaTunnelHealthMonitor.java index 988cbe3cd0..5b5be64806 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/SeaTunnelHealthMonitor.java +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/SeaTunnelHealthMonitor.java @@ -27,6 +27,8 @@ import com.hazelcast.logging.ILogger; import com.hazelcast.spi.properties.ClusterProperty; import lombok.Getter; +import java.util.Locale; + import static com.hazelcast.internal.diagnostics.HealthMonitorLevel.valueOf; import static com.hazelcast.spi.properties.ClusterProperty.HEALTH_MONITORING_THRESHOLD_CPU_PERCENTAGE; import static com.hazelcast.spi.properties.ClusterProperty.HEALTH_MONITORING_THRESHOLD_MEMORY_PERCENTAGE; @@ -72,7 +74,7 @@ public class SeaTunnelHealthMonitor { * @return a string of the given number as a format float with two decimal places and a period */ private static String percentageString(double p) { - return format("%.2f%%", p); + return format(Locale.ROOT, "%.2f%%", p); } private static String numberToUnit(long number) { @@ -80,7 +82,7 @@ public class SeaTunnelHealthMonitor { // 1024 is for 1024 kb is 1 MB etc double step = Math.pow(1024, i); if (number > step) { - return format("%3.1f%s", number / step, UNITS[i]); + return format(Locale.ROOT, "%3.1f%s", number / step, UNITS[i]); } } return Long.toString(number); @@ -252,11 +254,11 @@ public class SeaTunnelHealthMonitor { private void renderLoad() { sb.append("load.process") .append('=') - .append(format("%.2f", osProcessCpuLoad.read())) + .append(format(Locale.ROOT, "%.2f", osProcessCpuLoad.read())) .append("%, "); sb.append("load.system") .append('=') - .append(format("%.2f", osSystemCpuLoad.read())) + .append(format(Locale.ROOT, "%.2f", osSystemCpuLoad.read())) .append("%, "); double value = osSystemLoadAverage.read(); @@ -265,7 +267,7 @@ public class SeaTunnelHealthMonitor { } else { sb.append("load.systemAverage") .append('=') - .append(format("%.2f", osSystemLoadAverage.read())) + .append(format(Locale.ROOT, "%.2f", osSystemLoadAverage.read())) .append(", "); } } @@ -412,7 +414,11 @@ public class SeaTunnelHealthMonitor { .append(operationServiceRunningOperationsCount.read()) .append(", "); sb.append("operations.pending.invocations.percentage=") - .append(format("%.2f", operationServicePendingInvocationsPercentage.read())) + .append( + format( + Locale.ROOT, + "%.2f", + operationServicePendingInvocationsPercentage.read())) .append("%, "); sb.append("operations.pending.invocations.count=") .append(operationServicePendingInvocationsCount.read()) diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/BaseService.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/BaseService.java index ba7344de80..95c05e60cd 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/BaseService.java +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/BaseService.java @@ -72,6 +72,7 @@ import com.hazelcast.internal.json.JsonValue; import com.hazelcast.internal.serialization.Data; import com.hazelcast.internal.util.JsonUtil; import com.hazelcast.map.IMap; +import com.hazelcast.spi.impl.InternalCompletableFuture; import com.hazelcast.spi.impl.NodeEngineImpl; import lombok.extern.slf4j.Slf4j; @@ -82,6 +83,7 @@ import java.util.Arrays; import java.util.Comparator; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; @@ -91,6 +93,7 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.regex.Matcher; @@ -1398,31 +1401,91 @@ public abstract class BaseService { voidPassiveCompletableFuture.join(); } + /** + * Collects one health-metrics entry per cluster member for the system-monitoring-information + * REST API. + * + * <p>Requests are dispatched to every member before any of them is awaited, and all responses + * are awaited against a single shared deadline, so the total latency of this call is bounded by + * {@code seatunnel.engine.health-metrics-timeout-seconds} instead of growing with the member + * count. A member that misses the deadline (or whose request/response fails) is still reported + * with its address and an {@code error} marker, so consumers can tell which member is missing + * and why instead of receiving an anonymous empty object. + */ protected JsonArray getSystemMonitoringInformationJsonValues() { Cluster cluster = nodeEngine.getHazelcastInstance().getCluster(); Set<Member> members = cluster.getMembers(); - JsonArray jsonValues = - members.stream() - .map( - member -> { - Address address = member.getAddress(); - String input = null; - try { - input = - (String) - NodeEngineUtil.sendOperationToMemberNode( - nodeEngine, - new GetClusterHealthMetricsOperation(), - address) - .get(); - } catch (InterruptedException | ExecutionException e) { - - log.error("Failed to get cluster health metrics", e); - } - return parseSystemMonitoringMetrics(input, address); - }) - .collect(JsonArray::new, JsonArray::add, JsonArray::add); + Map<Address, InternalCompletableFuture<Object>> futures = new LinkedHashMap<>(); + for (Member member : members) { + Address address = member.getAddress(); + try { + futures.put( + address, + NodeEngineUtil.sendOperationToMemberNode( + nodeEngine, new GetClusterHealthMetricsOperation(), address)); + } catch (Exception e) { + log.error("Failed to send cluster health metrics request to {}", address, e); + futures.put(address, null); + } + } + return collectHealthMetrics(futures); + } + + int getHealthMetricsTimeoutSeconds() { + return getSeaTunnelServer(false) + .getSeaTunnelConfig() + .getEngineConfig() + .getHealthMetricsTimeoutSeconds(); + } + + private JsonObject unfinishedMember(Address address, String reason) { + JsonObject memberInfo = new JsonObject(); + if (address != null) { + memberInfo.add("host", address.getHost()); + memberInfo.add("port", address.getPort()); + } + memberInfo.add("error", reason); + return memberInfo; + } + + JsonArray collectHealthMetrics(Map<Address, InternalCompletableFuture<Object>> futures) { + int timeoutSeconds = getHealthMetricsTimeoutSeconds(); + long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds); + JsonArray jsonValues = new JsonArray(); + for (Map.Entry<Address, InternalCompletableFuture<Object>> entry : futures.entrySet()) { + Address address = entry.getKey(); + InternalCompletableFuture<Object> future = entry.getValue(); + if (future == null) { + jsonValues.add(unfinishedMember(address, "dispatch-failure")); + continue; + } + // clamp to a positive remaining time so already-completed futures are still collected + // once the shared deadline has passed + long remainingNanos = Math.max(1L, deadlineNanos - System.nanoTime()); + try { + String input = (String) future.get(remainingNanos, TimeUnit.NANOSECONDS); + jsonValues.add(parseSystemMonitoringMetrics(input, address)); + } catch (TimeoutException e) { + log.warn( + "Timeout after {}s waiting for health metrics from {}", + timeoutSeconds, + address); + // cancel() only releases the local future; it cannot stop the remote operation + future.cancel(false); + jsonValues.add(unfinishedMember(address, "timeout")); + } catch (InterruptedException e) { + future.cancel(false); + Thread.currentThread().interrupt(); + log.warn( + "Interrupted while waiting for health metrics from {}, stop waiting for the remaining members", + address); + break; + } catch (ExecutionException e) { + log.error("Failed to get cluster health metrics from {}", address, e); + jsonValues.add(unfinishedMember(address, "execution-failure")); + } + } return jsonValues; } diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/SeaTunnelHealthMonitorTest.java b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/SeaTunnelHealthMonitorTest.java new file mode 100644 index 0000000000..5e847fb949 --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/SeaTunnelHealthMonitorTest.java @@ -0,0 +1,64 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.engine.server; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Method; +import java.util.Locale; + +public class SeaTunnelHealthMonitorTest { + + /** + * Verifies that the percentage formatter always uses {@link Locale#ROOT} so the rendered output + * is stable regardless of the JVM default locale. Some locales (e.g. {@code de_DE}) would + * otherwise produce {@code 12,34} instead of {@code 12.34}. + */ + @Test + public void testPercentageStringIsLocaleStable() throws Exception { + Method method = + SeaTunnelHealthMonitor.class.getDeclaredMethod("percentageString", double.class); + method.setAccessible(true); + + Locale previous = Locale.getDefault(); + try { + Locale.setDefault(Locale.GERMANY); + String result = (String) method.invoke(null, 12.345); + Assertions.assertEquals("12.35%", result); + } finally { + Locale.setDefault(previous); + } + } + + /** Verifies that the number-to-unit formatter always uses {@link Locale#ROOT}. */ + @Test + public void testNumberToUnitIsLocaleStable() throws Exception { + Method method = SeaTunnelHealthMonitor.class.getDeclaredMethod("numberToUnit", long.class); + method.setAccessible(true); + + Locale previous = Locale.getDefault(); + try { + Locale.setDefault(Locale.GERMANY); + String result = (String) method.invoke(null, 2L * 1024 * 1024); + Assertions.assertEquals("2.0M", result); + } finally { + Locale.setDefault(previous); + } + } +} diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/BaseServiceHealthMetricsTest.java b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/BaseServiceHealthMetricsTest.java new file mode 100644 index 0000000000..c1fb08ccc5 --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/BaseServiceHealthMetricsTest.java @@ -0,0 +1,158 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.engine.server.rest.service; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import com.hazelcast.cluster.Address; +import com.hazelcast.internal.json.JsonArray; +import com.hazelcast.internal.json.JsonObject; +import com.hazelcast.spi.impl.InternalCompletableFuture; + +import java.net.UnknownHostException; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +/** + * Verifies the member-collection contract of {@link + * BaseService#getSystemMonitoringInformationJsonValues()}: all members share one deadline, and + * members whose metrics are not collected are still reported with their address and an error marker + * instead of an anonymous empty object. + */ +public class BaseServiceHealthMetricsTest { + + private static Address address(int port) throws UnknownHostException { + return new Address("127.0.0.1", port); + } + + private static InternalCompletableFuture<Object> future(Object value) { + InternalCompletableFuture<Object> future = new InternalCompletableFuture<>(); + if (value != null) { + future.complete(value); + } + return future; + } + + /** Avoids the live SeaTunnelServer config lookup by pinning a short timeout. */ + private static final class TestService extends BaseService { + + private TestService() { + super(null); + } + + @Override + int getHealthMetricsTimeoutSeconds() { + return 1; + } + } + + private static void assertTimeoutEntry( + JsonObject entry, String expectedHost, int expectedPort) { + Assertions.assertEquals(expectedHost, entry.get("host").asString()); + Assertions.assertEquals(expectedPort, entry.get("port").asInt()); + Assertions.assertEquals("timeout", entry.get("error").asString()); + } + + @Test + public void testTimedOutMemberIsReportedWithAddressAndErrorMarker() throws Exception { + Map<Address, InternalCompletableFuture<Object>> futures = new LinkedHashMap<>(); + futures.put(address(5801), future(null)); + + JsonArray values = new TestService().collectHealthMetrics(futures); + + Assertions.assertEquals(1, values.size()); + assertTimeoutEntry(values.get(0).asObject(), "127.0.0.1", 5801); + } + + @Test + public void testSharedDeadlineBoundsTotalWaitForAllMembers() throws Exception { + Map<Address, InternalCompletableFuture<Object>> futures = new LinkedHashMap<>(); + futures.put(address(5801), future(null)); + futures.put(address(5802), future(null)); + futures.put(address(5803), future(null)); + futures.put( + address(5804), + future("isMaster=true, host=127.0.0.1, port=5804, heap.memory.used=135.7M")); + + long startNanos = System.nanoTime(); + JsonArray values = new TestService().collectHealthMetrics(futures); + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); + + Assertions.assertEquals(4, values.size()); + assertTimeoutEntry(values.get(0).asObject(), "127.0.0.1", 5801); + assertTimeoutEntry(values.get(1).asObject(), "127.0.0.1", 5802); + assertTimeoutEntry(values.get(2).asObject(), "127.0.0.1", 5803); + JsonObject collected = values.get(3).asObject(); + Assertions.assertEquals("true", collected.get("isMaster").asString()); + Assertions.assertEquals("135.7M", collected.get("heap.memory.used").asString()); + // a per-member deadline would wait 3 x timeout for the three hanging members + Assertions.assertTrue( + elapsedMillis < 2400, + "shared deadline should bound the total wait, took " + elapsedMillis + "ms"); + } + + @Test + public void testInterruptStopsWaitingForRemainingMembers() throws Exception { + Map<Address, InternalCompletableFuture<Object>> futures = new LinkedHashMap<>(); + InternalCompletableFuture<Object> hanging = future(null); + futures.put(address(5801), hanging); + futures.put(address(5802), future("isMaster=true, host=127.0.0.1, port=5802")); + + Thread.currentThread().interrupt(); + try { + JsonArray values = new TestService().collectHealthMetrics(futures); + Assertions.assertEquals(0, values.size()); + Assertions.assertTrue(hanging.isDone() && hanging.isCancelled()); + } finally { + // the collector must have restored the interrupt flag; clear it for other tests + Assertions.assertTrue(Thread.interrupted()); + } + } + + @Test + public void testFailedResponseIsReportedWithErrorMarker() throws Exception { + Map<Address, InternalCompletableFuture<Object>> futures = new LinkedHashMap<>(); + InternalCompletableFuture<Object> failed = future(null); + failed.completeExceptionally(new RuntimeException("boom")); + futures.put(address(5801), failed); + + JsonArray values = new TestService().collectHealthMetrics(futures); + + Assertions.assertEquals(1, values.size()); + JsonObject entry = values.get(0).asObject(); + Assertions.assertEquals("127.0.0.1", entry.get("host").asString()); + Assertions.assertEquals(5801, entry.get("port").asInt()); + Assertions.assertEquals("execution-failure", entry.get("error").asString()); + } + + @Test + public void testDispatchFailureIsReportedWithErrorMarker() throws Exception { + Map<Address, InternalCompletableFuture<Object>> futures = new LinkedHashMap<>(); + futures.put(address(5801), null); + + JsonArray values = new TestService().collectHealthMetrics(futures); + + Assertions.assertEquals(1, values.size()); + JsonObject entry = values.get(0).asObject(); + Assertions.assertEquals("127.0.0.1", entry.get("host").asString()); + Assertions.assertEquals(5801, entry.get("port").asInt()); + Assertions.assertEquals("dispatch-failure", entry.get("error").asString()); + } +}
