This is an automated email from the ASF dual-hosted git repository.
dybyte pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 59dcd935b6 [Fix][Zeta] Avoid telemetry collector blocking before
CoordinatorService is ready (#10841)
59dcd935b6 is described below
commit 59dcd935b60a326558cae9c7de84e267edcfb56f
Author: zhiwei.niu <[email protected]>
AuthorDate: Sun May 10 21:42:48 2026 +0800
[Fix][Zeta] Avoid telemetry collector blocking before CoordinatorService is
ready (#10841)
---
.../seatunnel/engine/server/SeaTunnelServer.java | 4 +
.../telemetry/metrics/AbstractCollector.java | 5 +
.../metrics/exports/ClusterMetricExports.java | 41 ++--
.../metrics/exports/JobMetricExports.java | 4 +-
.../exports/JobThreadPoolStatusExports.java | 3 +-
.../TelemetryCollectorCoordinatorGuardTest.java | 240 +++++++++++++++++++++
6 files changed, 275 insertions(+), 22 deletions(-)
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/SeaTunnelServer.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/SeaTunnelServer.java
index ee3be21afe..b0abdb664a 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/SeaTunnelServer.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/SeaTunnelServer.java
@@ -424,6 +424,10 @@ public class SeaTunnelServer
return (key.hashCode() & 0x7FFFFFFF) % partitionCount;
}
+ public boolean isCoordinatorActive() {
+ return coordinatorService.isCoordinatorActive();
+ }
+
public SeaTunnelConfig getSeaTunnelConfig() {
return seaTunnelConfig;
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/AbstractCollector.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/AbstractCollector.java
index 1165e14d8d..3ef0203385 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/AbstractCollector.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/AbstractCollector.java
@@ -69,6 +69,11 @@ public abstract class AbstractCollector extends Collector {
return getServer().getCoordinatorService();
}
+ // Non-blocking coordinator readiness check; call before
getCoordinatorService() in collect().
+ protected boolean isCoordinatorReady() {
+ return getServer().isCoordinatorActive();
+ }
+
protected ManagementService getManagementService() {
return getNode().hazelcastInstance.getManagementService();
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/ClusterMetricExports.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/ClusterMetricExports.java
index 19910e6a75..7cbc2a6431 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/ClusterMetricExports.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/ClusterMetricExports.java
@@ -19,13 +19,12 @@ package
org.apache.seatunnel.engine.server.telemetry.metrics.exports;
import org.apache.seatunnel.engine.server.telemetry.metrics.AbstractCollector;
-import com.hazelcast.cluster.impl.MemberImpl;
+import com.hazelcast.cluster.Address;
import com.hazelcast.instance.impl.Node;
import io.prometheus.client.GaugeMetricFamily;
import java.net.UnknownHostException;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.List;
public class ClusterMetricExports extends AbstractCollector {
@@ -54,40 +53,44 @@ public class ClusterMetricExports extends AbstractCollector
{
"cluster_time",
"Cluster start time",
clusterLabelNames("hazelcastVersion"));
- List<String> labelValues =
labelValues(getClusterService().getClusterVersion().toString());
-
- metricFamily.addMetric(labelValues,
getClusterService().getClusterTime());
+ metricFamily.addMetric(
+
labelValues(getClusterService().getClusterVersion().toString()),
+ getClusterService().getClusterTime());
mfs.add(metricFamily);
}
private void clusterInfo(final List<MetricFamilySamples> mfs) {
+ // Snapshot once to avoid TOCTOU race during master election.
+ Address masterAddr = getClusterService().getMasterAddress();
+ if (masterAddr == null) {
+ return;
+ }
+ // Keep the historical label format compatible with previous IP:port
output.
+ String masterIpPort;
+ try {
+ masterIpPort =
+ masterAddr.getInetAddress().getHostAddress() + ":" +
masterAddr.getPort();
+ } catch (UnknownHostException e) {
+ getLogger(ClusterMetricExports.class)
+ .warning("Skip cluster_info metric: unable to resolve
master address", e);
+ return;
+ }
GaugeMetricFamily metricFamily =
new GaugeMetricFamily(
"cluster_info",
"Cluster info",
clusterLabelNames("hazelcastVersion", "master"));
- List<String> labelValues = null;
- try {
- labelValues =
- labelValues(
-
getClusterService().getClusterVersion().toString(), masterAddress());
- } catch (UnknownHostException e) {
- e.printStackTrace();
- }
-
+ List<String> labelValues =
+
labelValues(getClusterService().getClusterVersion().toString(), masterIpPort);
metricFamily.addMetric(labelValues, 1.0);
mfs.add(metricFamily);
}
private void nodeCount(final List<MetricFamilySamples> mfs) {
- Collection<MemberImpl> memberImpls =
getClusterService().getMemberImpls();
-
GaugeMetricFamily metricFamily =
new GaugeMetricFamily(
"node_count", "Cluster node total count ",
clusterLabelNames());
- List<String> labelValues = labelValues();
-
- metricFamily.addMetric(labelValues, memberImpls.size());
+ metricFamily.addMetric(labelValues(),
getClusterService().getMemberImpls().size());
mfs.add(metricFamily);
}
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/JobMetricExports.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/JobMetricExports.java
index 53ff9e05a6..b197e29bb1 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/JobMetricExports.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/JobMetricExports.java
@@ -36,8 +36,8 @@ public class JobMetricExports extends AbstractCollector {
@Override
public List<MetricFamilySamples> collect() {
List<MetricFamilySamples> mfs = new ArrayList();
- // Only the master can get job metrics
- if (isMaster()) {
+ // Report metrics only when the local node is ACTIVE and the master is
available.
+ if (isMaster() && isCoordinatorReady()) {
CoordinatorService coordinatorService = getCoordinatorService();
JobCounter jobCountMetrics =
coordinatorService.getJobCountMetrics();
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/JobThreadPoolStatusExports.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/JobThreadPoolStatusExports.java
index c75a3324c8..ed9b6adc68 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/JobThreadPoolStatusExports.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/JobThreadPoolStatusExports.java
@@ -39,7 +39,8 @@ public class JobThreadPoolStatusExports extends
AbstractCollector {
@Override
public List<MetricFamilySamples> collect() {
List<MetricFamilySamples> mfs = new ArrayList();
- if (isMaster()) {
+ // Only report metrics when the local node is master and in READY
state.
+ if (isMaster() && isCoordinatorReady()) {
ThreadPoolStatus threadPoolStatusMetrics =
getServer().getThreadPoolStatusMetrics();
List<String> labelNames = clusterLabelNames(ADDRESS, "type");
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/TelemetryCollectorCoordinatorGuardTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/TelemetryCollectorCoordinatorGuardTest.java
new file mode 100644
index 0000000000..c41a923580
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/TelemetryCollectorCoordinatorGuardTest.java
@@ -0,0 +1,240 @@
+/*
+ * 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.telemetry.metrics;
+
+import org.apache.seatunnel.engine.server.CoordinatorService;
+import org.apache.seatunnel.engine.server.SeaTunnelServer;
+import org.apache.seatunnel.engine.server.telemetry.metrics.entity.JobCounter;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.entity.ThreadPoolStatus;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.ClusterMetricExports;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.JobMetricExports;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.JobThreadPoolStatusExports;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import com.hazelcast.cluster.Address;
+import com.hazelcast.cluster.impl.MemberImpl;
+import com.hazelcast.config.Config;
+import com.hazelcast.instance.impl.Node;
+import com.hazelcast.internal.cluster.Versions;
+import com.hazelcast.internal.cluster.impl.ClusterServiceImpl;
+import com.hazelcast.logging.ILogger;
+import com.hazelcast.spi.impl.NodeEngineImpl;
+import com.hazelcast.version.Version;
+import io.prometheus.client.Collector;
+
+import java.lang.reflect.Field;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Collections;
+import java.util.List;
+
+public class TelemetryCollectorCoordinatorGuardTest {
+
+ private Node mockNode;
+ private SeaTunnelServer mockServer;
+ private CoordinatorService mockCoordinatorService;
+ private ClusterServiceImpl mockClusterService;
+ private ILogger mockLogger;
+
+ @BeforeEach
+ void setUp() throws UnknownHostException, NoSuchFieldException,
IllegalAccessException {
+ mockNode = Mockito.mock(Node.class);
+ mockServer = Mockito.mock(SeaTunnelServer.class);
+ mockCoordinatorService = Mockito.mock(CoordinatorService.class);
+ mockClusterService = Mockito.mock(ClusterServiceImpl.class);
+ mockLogger = Mockito.mock(ILogger.class);
+
+ NodeEngineImpl mockNodeEngine = Mockito.mock(NodeEngineImpl.class);
+ MemberImpl mockMember = Mockito.mock(MemberImpl.class);
+ Config mockConfig = Mockito.mock(Config.class);
+
+ Mockito.when(mockNode.getNodeEngine()).thenReturn(mockNodeEngine);
+ Mockito.when(mockNodeEngine.getService(SeaTunnelServer.SERVICE_NAME))
+ .thenReturn(mockServer);
+ Mockito.when(mockNodeEngine.getLocalMember()).thenReturn(mockMember);
+
Mockito.when(mockNode.getClusterService()).thenReturn(mockClusterService);
+
Mockito.when(mockNode.getLogger(Mockito.any(Class.class))).thenReturn(mockLogger);
+ Mockito.when(mockNode.getConfig()).thenReturn(mockConfig);
+ Mockito.when(mockConfig.getClusterName()).thenReturn("test-cluster");
+
+ // AbstractCollector.getLocalMember() reads Node.nodeEngine as a
direct public field,
+ // not via a getter, so we inject it via reflection.
+ Field nodeEngineField = Node.class.getDeclaredField("nodeEngine");
+ nodeEngineField.setAccessible(true);
+ nodeEngineField.set(mockNode, mockNodeEngine);
+
+ InetAddress inetAddress = InetAddress.getByName("127.0.0.1");
+ Mockito.when(mockMember.getInetAddress()).thenReturn(inetAddress);
+ Mockito.when(mockMember.getPort()).thenReturn(5801);
+
+
Mockito.when(mockServer.getCoordinatorService()).thenReturn(mockCoordinatorService);
+
+ // Default stubs for ClusterService — always needed because
clusterTime() and nodeCount()
+ // run unconditionally on every collect() call.
+ Version clusterVersion =
+ Version.of(
+ Versions.CURRENT_CLUSTER_VERSION.getMajor(),
+ Versions.CURRENT_CLUSTER_VERSION.getMinor());
+
Mockito.when(mockClusterService.getClusterVersion()).thenReturn(clusterVersion);
+
Mockito.when(mockClusterService.getMemberImpls()).thenReturn(Collections.emptyList());
+
Mockito.when(mockClusterService.getClusterTime()).thenReturn(System.currentTimeMillis());
+ }
+
+ @Test
+ void testJobMetricExportsReturnsEmptyWhenCoordinatorNotReady() {
+ Mockito.when(mockNode.isMaster()).thenReturn(true);
+ Mockito.when(mockServer.isCoordinatorActive()).thenReturn(false);
+
+ JobMetricExports exports = new JobMetricExports(mockNode);
+ List<Collector.MetricFamilySamples> result = exports.collect();
+
+ Assertions.assertTrue(
+ result.isEmpty(),
+ "collect() must return empty when coordinator is not ready"
+ + " to avoid blocking Hazelcast operation threads");
+ Mockito.verify(mockServer, Mockito.never()).getCoordinatorService();
+ }
+
+ @Test
+ void testJobMetricExportsReturnsMetricsWhenCoordinatorReady() {
+ Mockito.when(mockNode.isMaster()).thenReturn(true);
+ Mockito.when(mockServer.isCoordinatorActive()).thenReturn(true);
+ Mockito.when(mockCoordinatorService.getJobCountMetrics())
+ .thenReturn(new JobCounter(0, 0, 0, 1, 0, 0, 0, 0, 0));
+
+ JobMetricExports exports = new JobMetricExports(mockNode);
+ List<Collector.MetricFamilySamples> result = exports.collect();
+
+ Assertions.assertFalse(
+ result.isEmpty(), "collect() must return metrics when
coordinator is ready");
+ Mockito.verify(mockCoordinatorService).getJobCountMetrics();
+ }
+
+ @Test
+ void testJobMetricExportsReturnsEmptyWhenNotMaster() {
+ Mockito.when(mockNode.isMaster()).thenReturn(false);
+
+ JobMetricExports exports = new JobMetricExports(mockNode);
+ List<Collector.MetricFamilySamples> result = exports.collect();
+
+ Assertions.assertTrue(result.isEmpty(), "collect() must return empty
on non-master node");
+ Mockito.verify(mockServer, Mockito.never()).isCoordinatorActive();
+ }
+
+ @Test
+ void testJobThreadPoolStatusExportsReturnsEmptyWhenCoordinatorNotReady() {
+ Mockito.when(mockNode.isMaster()).thenReturn(true);
+ Mockito.when(mockServer.isCoordinatorActive()).thenReturn(false);
+
+ JobThreadPoolStatusExports exports = new
JobThreadPoolStatusExports(mockNode);
+ List<Collector.MetricFamilySamples> result = exports.collect();
+
+ Assertions.assertTrue(
+ result.isEmpty(),
+ "collect() must return empty when coordinator is not ready"
+ + " to avoid blocking Hazelcast operation threads");
+ Mockito.verify(mockServer,
Mockito.never()).getThreadPoolStatusMetrics();
+ }
+
+ @Test
+ void testJobThreadPoolStatusExportsReturnsMetricsWhenCoordinatorReady() {
+ Mockito.when(mockNode.isMaster()).thenReturn(true);
+ Mockito.when(mockServer.isCoordinatorActive()).thenReturn(true);
+ ThreadPoolStatus status = new ThreadPoolStatus(1, 2, 10, 3, 100L,
110L, 0L, 0L);
+
Mockito.when(mockServer.getThreadPoolStatusMetrics()).thenReturn(status);
+
+ JobThreadPoolStatusExports exports = new
JobThreadPoolStatusExports(mockNode);
+ List<Collector.MetricFamilySamples> result = exports.collect();
+
+ Assertions.assertFalse(
+ result.isEmpty(), "collect() must return metrics when
coordinator is ready");
+ Mockito.verify(mockServer).getThreadPoolStatusMetrics();
+ }
+
+ @Test
+ void testJobThreadPoolStatusExportsReturnsEmptyWhenNotMaster() {
+ Mockito.when(mockNode.isMaster()).thenReturn(false);
+
+ JobThreadPoolStatusExports exports = new
JobThreadPoolStatusExports(mockNode);
+ List<Collector.MetricFamilySamples> result = exports.collect();
+
+ Assertions.assertTrue(result.isEmpty(), "collect() must return empty
on non-master node");
+ Mockito.verify(mockServer, Mockito.never()).isCoordinatorActive();
+ }
+
+ //
-------------------------------------------------------------------------
+ // ClusterMetricExports
+ //
-------------------------------------------------------------------------
+
+ @Test
+ void testClusterMetricExportsSkipsClusterInfoWhenMasterAddressNull() {
+ Mockito.when(mockClusterService.getMasterAddress()).thenReturn(null);
+
+ List<Collector.MetricFamilySamples> result = new
ClusterMetricExports(mockNode).collect();
+
+ boolean hasClusterInfo = result.stream().anyMatch(s ->
"cluster_info".equals(s.name));
+ Assertions.assertFalse(
+ hasClusterInfo, "cluster_info must be skipped when master
address is null");
+ }
+
+ @Test
+ void
testClusterMetricExportsIncludesClusterInfoWhenMasterAddressAvailable()
+ throws UnknownHostException {
+ Mockito.when(mockClusterService.getMasterAddress())
+ .thenReturn(new Address("127.0.0.1", 5801));
+
+ List<Collector.MetricFamilySamples> result = new
ClusterMetricExports(mockNode).collect();
+
+ Collector.MetricFamilySamples clusterInfoMetric =
+ result.stream().filter(s ->
"cluster_info".equals(s.name)).findFirst().orElse(null);
+ boolean hasClusterInfo = clusterInfoMetric != null;
+ Assertions.assertTrue(
+ hasClusterInfo, "cluster_info must be present when master
address is available");
+ Assertions.assertNotNull(clusterInfoMetric);
+ Assertions.assertFalse(clusterInfoMetric.samples.isEmpty());
+
+ Collector.MetricFamilySamples.Sample sample =
clusterInfoMetric.samples.get(0);
+ int masterLabelIndex = sample.labelNames.indexOf("master");
+ Assertions.assertTrue(masterLabelIndex >= 0, "cluster_info must
contain 'master' label");
+ Assertions.assertEquals("127.0.0.1:5801",
sample.labelValues.get(masterLabelIndex));
+ }
+
+ @Test
+ void
testClusterMetricExportsSkipsClusterInfoAndLogsWarningWhenMasterAddressUnresolvable()
+ throws UnknownHostException {
+ Address mockAddress = Mockito.mock(Address.class);
+ Mockito.when(mockAddress.getInetAddress()).thenThrow(new
UnknownHostException("mock"));
+ Mockito.when(mockAddress.getPort()).thenReturn(5801);
+
Mockito.when(mockClusterService.getMasterAddress()).thenReturn(mockAddress);
+
+ List<Collector.MetricFamilySamples> result = new
ClusterMetricExports(mockNode).collect();
+
+ boolean hasClusterInfo = result.stream().anyMatch(s ->
"cluster_info".equals(s.name));
+ Assertions.assertFalse(
+ hasClusterInfo,
+ "cluster_info must be skipped when master address can't be
resolved");
+ Mockito.verify(mockLogger)
+ .warning(
+ Mockito.eq("Skip cluster_info metric: unable to
resolve master address"),
+ Mockito.any(UnknownHostException.class));
+ }
+}
