This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new ce6c1f5f477 Fix pipe sink metrics lifecycle race (#18722)
ce6c1f5f477 is described below
commit ce6c1f5f477b2da5b5071890c715c6b6a9652e45
Author: Caideyipi <[email protected]>
AuthorDate: Tue Sep 29 12:25:53 2026 +0800
Fix pipe sink metrics lifecycle race (#18722)
---
.../metric/sink/PipeDataRegionSinkMetrics.java | 8 +-
.../metric/sink/PipeDataRegionSinkMetricsTest.java | 98 ++++++++++++++++++++++
2 files changed, 102 insertions(+), 4 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java
index a52d32ed0d7..9990625dd79 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java
@@ -59,7 +59,7 @@ public class PipeDataRegionSinkMetrics implements IMetricSet {
//////////////////////////// bindTo & unbindFrom (metric framework)
////////////////////////////
@Override
- public void bindTo(final AbstractMetricService metricService) {
+ public synchronized void bindTo(final AbstractMetricService metricService) {
this.metricService = metricService;
final ImmutableSet<String> taskIDs = ImmutableSet.copyOf(sinkMap.keySet());
for (String taskID : taskIDs) {
@@ -247,7 +247,7 @@ public class PipeDataRegionSinkMetrics implements
IMetricSet {
}
@Override
- public void unbindFrom(final AbstractMetricService metricService) {
+ public synchronized void unbindFrom(final AbstractMetricService
metricService) {
final ImmutableSet<String> taskIDs = ImmutableSet.copyOf(sinkMap.keySet());
for (final String taskID : taskIDs) {
deregister(taskID);
@@ -335,7 +335,7 @@ public class PipeDataRegionSinkMetrics implements
IMetricSet {
//////////////////////////// register & deregister (pipe integration)
////////////////////////////
- public void register(final PipeSinkSubtask pipeSinkSubtask) {
+ public synchronized void register(final PipeSinkSubtask pipeSinkSubtask) {
final String taskID = pipeSinkSubtask.getTaskID();
sinkMap.putIfAbsent(taskID, pipeSinkSubtask);
if (Objects.nonNull(metricService)) {
@@ -343,7 +343,7 @@ public class PipeDataRegionSinkMetrics implements
IMetricSet {
}
}
- public void deregister(final String taskID) {
+ public synchronized void deregister(final String taskID) {
if (!sinkMap.containsKey(taskID)) {
LOGGER.warn(
DataNodePipeMessages.FAILED_TO_DEREGISTER_PIPE_DATA_REGION_SINK,
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetricsTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetricsTest.java
new file mode 100644
index 00000000000..04d71410962
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetricsTest.java
@@ -0,0 +1,98 @@
+/*
+ * 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.iotdb.db.pipe.metric.sink;
+
+import org.apache.iotdb.db.pipe.agent.task.subtask.sink.PipeSinkSubtask;
+import org.apache.iotdb.metrics.AbstractMetricService;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.lang.reflect.Constructor;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.mockito.Mockito.when;
+
+public class PipeDataRegionSinkMetricsTest {
+
+ @Test
+ public void testDeregisterWaitsForBind() throws Exception {
+ final String taskId = "data-region-sink-task";
+ final CountDownLatch bindEntered = new CountDownLatch(1);
+ final CountDownLatch continueBind = new CountDownLatch(1);
+ final AtomicBoolean blockBindOnce = new AtomicBoolean(true);
+ final PipeSinkSubtask subtask = Mockito.mock(PipeSinkSubtask.class);
+ when(subtask.getTaskID()).thenReturn(taskId);
+ when(subtask.getPipeName())
+ .thenAnswer(
+ invocation -> {
+ if (blockBindOnce.compareAndSet(true, false)) {
+ bindEntered.countDown();
+ Assert.assertTrue(continueBind.await(5, TimeUnit.SECONDS));
+ }
+ return "pipe";
+ });
+ when(subtask.getAttributeSortedString()).thenReturn("sink");
+ when(subtask.getCreationTime()).thenReturn(1L);
+ when(subtask.getSinkIndex()).thenReturn(0);
+
+ final AbstractMetricService metricService =
+ Mockito.mock(AbstractMetricService.class, Mockito.RETURNS_MOCKS);
+
+ final Constructor<PipeDataRegionSinkMetrics> constructor =
+ PipeDataRegionSinkMetrics.class.getDeclaredConstructor();
+ constructor.setAccessible(true);
+ final PipeDataRegionSinkMetrics metrics = constructor.newInstance();
+ metrics.register(subtask);
+
+ final ExecutorService executor = Executors.newFixedThreadPool(2);
+ final Future<?> bindFuture = executor.submit(() ->
metrics.bindTo(metricService));
+ Future<?> deregisterFuture = null;
+ try {
+ Assert.assertTrue(bindEntered.await(5, TimeUnit.SECONDS));
+ deregisterFuture = executor.submit(() -> metrics.deregister(taskId));
+ try {
+ try {
+ deregisterFuture.get(200, TimeUnit.MILLISECONDS);
+ Assert.fail("Deregistration must wait until metric binding
completes");
+ } catch (final TimeoutException expected) {
+ // Expected: bindTo holds the lifecycle lock until all metrics are
created.
+ }
+ } finally {
+ continueBind.countDown();
+ }
+
+ bindFuture.get(5, TimeUnit.SECONDS);
+ Assert.assertNotNull(deregisterFuture);
+ deregisterFuture.get(5, TimeUnit.SECONDS);
+ } finally {
+ continueBind.countDown();
+ executor.shutdownNow();
+ Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+}
