[To dev/1.3] [Pipe] Fix pipe sink metrics lifecycle race (#18722)

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java

@@ -58,7 +58,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(connectorMap.keySet());
     for (String taskID : taskIDs) {
@@ -248,7 +248,7 @@ private void createHistogram(final String taskID) {
   }
 
   @Override
-  public void unbindFrom(final AbstractMetricService metricService) {
+  public synchronized void unbindFrom(final AbstractMetricService metricService) {
     final ImmutableSet<String> taskIDs = ImmutableSet.copyOf(connectorMap.keySet());
     for (final String taskID : taskIDs) {
       deregister(taskID);
@@ -353,15 +353,15 @@ private void removeHistogram(final String taskID) {
 
   //////////////////////////// register & deregister (pipe integration) ////////////////////////////
 
-  public void register(final PipeSinkSubtask pipeSinkSubtask) {
+  public synchronized void register(final PipeSinkSubtask pipeSinkSubtask) {
     final String taskID = pipeSinkSubtask.getTaskID();
     connectorMap.putIfAbsent(taskID, pipeSinkSubtask);
     if (Objects.nonNull(metricService)) {
       createMetrics(taskID);
     }
   }
 
-  public void deregister(final String taskID) {
+  public synchronized void deregister(final String taskID) {
     if (!connectorMap.containsKey(taskID)) {
       LOGGER.warn(
           "Failed to deregister pipe data region sink metrics, PipeSinkSubtask({}) does not exist",

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.getConnectorIndex()).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));
+    }
+  }
+}