[DO NOT MERGE]

.github/workflows/daily-it.yml

@@ -1,6 +1,9 @@
 name: Daily IT
 
 on:
+  pull_request:
+    branches:
+      - master
   schedule:
     # Run at UTC 19:00 every day (CST 03:00 AM)
     - cron: "0 19 * * *"

integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java

@@ -37,6 +37,7 @@
 import java.util.function.Consumer;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -66,6 +67,12 @@ public void testAssertAlwaysFail() {
         () -> TestUtils.assertAlwaysFail(EnvFactory.getEnv(), "show databases", 1));
   }
 
+  @Test
+  public void testHalfTimeoutElapsedUsesSeconds() {
+    assertFalse(TestUtils.isHalfTimeoutElapsed(299_999, 600));
+    assertTrue(TestUtils.isHalfTimeoutElapsed(300_001, 600));
+  }
+
   @Test
   public void testAssertDataAlwaysFailsOnMismatch() {
     final BaseEnv env = EnvFactory.getEnv();

integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java

@@ -85,6 +85,10 @@ public class TestUtils {
 
   public static final String TIME_PRECISION_IN_NS = "ns";
 
+  static boolean isHalfTimeoutElapsed(final long elapsedTimeInMillis, final long timeoutSeconds) {
+    return elapsedTimeInMillis > TimeUnit.SECONDS.toMillis(timeoutSeconds) / 2;
+  }
+
   /**
    * Creates a statement on the write node without comparing results across DataNodes.
    *
@@ -1702,7 +1706,9 @@ public static void assertDataEventuallyOnEnv(
                   // For IoTV2 batch mode, the pipe receiver may need to flush because the replica
                   // sync requires tsFile to process. We flush in the middle of assertion because we
                   // don't know when the data reaches the receiver in general cases
-                  if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) {
+                  if (!flushed[0]
+                      && isHalfTimeoutElapsed(
+                          System.currentTimeMillis() - startTime, timeoutSeconds)) {
                     flushed[0] = true;
                     statement.execute("flush");
                   }
@@ -1783,7 +1789,9 @@ public static void assertDataSizeEventuallyOnEnv(
                   // For IoTV2 batch mode, the pipe receiver may need to flush because the replica
                   // sync requires tsFile to process. We flush in the middle of assertion because we
                   // don't know when the data reaches the receiver in general cases
-                  if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) {
+                  if (!flushed[0]
+                      && isHalfTimeoutElapsed(
+                          System.currentTimeMillis() - startTime, timeoutSeconds)) {
                     flushed[0] = true;
                     statement.execute("flush");
                   }
@@ -1822,7 +1830,9 @@ public static void assertDataEventuallyOnEnv(
                   // For IoTV2 batch mode, the pipe receiver may need to flush because the replica
                   // sync requires tsFile to process. We flush in the middle of assertion because we
                   // don't know when the data reaches the receiver in general cases
-                  if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) {
+                  if (!flushed[0]
+                      && isHalfTimeoutElapsed(
+                          System.currentTimeMillis() - startTime, timeoutSeconds)) {
                     flushed[0] = true;
                     statement.execute("flush");
                   }
@@ -1904,7 +1914,9 @@ public static void assertDataEventuallyOnEnv(
                   // For IoTV2 batch mode, the pipe receiver may need to flush because the replica
                   // sync requires tsFile to process. We flush in the middle of assertion because we
                   // don't know when the data reaches the receiver in general cases
-                  if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) {
+                  if (!flushed[0]
+                      && isHalfTimeoutElapsed(
+                          System.currentTimeMillis() - startTime, timeoutSeconds)) {
                     flushed[0] = true;
                     statement.execute("flush");
                   }
@@ -2024,7 +2036,9 @@ public static void assertDataEventuallyOnEnv(
                   // For IoTV2 batch mode, the pipe receiver may need to flush because the replica
                   // sync requires tsFile to process. We flush in the middle of assertion because we
                   // don't know when the data reaches the receiver in general cases
-                  if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) {
+                  if (!flushed[0]
+                      && isHalfTimeoutElapsed(
+                          System.currentTimeMillis() - startTime, timeoutSeconds)) {
                     flushed[0] = true;
                     statement.execute("flush");
                   }
@@ -2095,8 +2109,8 @@ public static void assertDataAlwaysOnEnv(
                   // sync requires tsFile to process. We flush in the middle of assertion because we
                   // don't know when the data reaches the receiver in general cases
                   if (!flushed[0]
-                      && System.currentTimeMillis() - startTime
-                          > TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) {
+                      && isHalfTimeoutElapsed(
+                          System.currentTimeMillis() - startTime, consistentSeconds)) {
                     flushed[0] = true;
                     statement.execute("flush");
                   }
@@ -2137,8 +2151,8 @@ public static void assertDataAlwaysOnEnv(
                   // sync requires tsFile to process. We flush in the middle of assertion because we
                   // don't know when the data reaches the receiver in general cases
                   if (!flushed[0]
-                      && System.currentTimeMillis() - startTime
-                          > TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) {
+                      && isHalfTimeoutElapsed(
+                          System.currentTimeMillis() - startTime, consistentSeconds)) {
                     flushed[0] = true;
                     statement.execute("flush");
                   }
@@ -2192,8 +2206,8 @@ public static void assertDataAlwaysOnEnv(
                   // sync requires tsFile to process. We flush in the middle of assertion because we
                   // don't know when the data reaches the receiver in general cases
                   if (!flushed[0]
-                      && System.currentTimeMillis() - startTime
-                          > TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) {
+                      && isHalfTimeoutElapsed(
+                          System.currentTimeMillis() - startTime, consistentSeconds)) {
                     flushed[0] = true;
                     statement.execute("flush");
                   }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java

@@ -176,16 +176,22 @@ private void initSink(final Map<String, String> sinkAttributes) throws Exception
    */
   @Override
   protected boolean executeOnce() throws Exception {
-    if (isClosed.get()) {
+    if (isClosed.get() || shouldStopSubmittingSelf.get()) {
       return false;
     }
 
     final Event event = lastEvent != null ? lastEvent : source.supply();
-    // Record the last event for retry when exception occurs
+    // Keep the event for retry or restart if supply() returns after the task is stopped.
     setLastEvent(event);
 
+    // close() may have released the last event before supply() returned.
+    if (isClosed.get()) {
+      clearReferenceCountAndReleaseLastEvent(event);
+      return false;
+    }
+
     try {
-      if (event == null) {
+      if (event == null || shouldStopSubmittingSelf.get()) {
         return false;
       }
 

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtaskTest.java

@@ -0,0 +1,150 @@
+/*
+ * 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.confignode.manager.pipe.agent.task;
+
+import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
+import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.commons.pipe.agent.task.subtask.PipeAbstractSinkSubtask;
+import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
+import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.pipe.api.PipeConnector;
+import org.apache.iotdb.pipe.api.PipeExtractor;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.lang.reflect.Field;
+import java.util.Collections;
+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 static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
+
+public class PipeConfigNodeSubtaskTest {
+
+  private PipeConfigNodeSubtask subtask;
+  private PipeExtractor source;
+  private PipeConnector sink;
+  private boolean subtaskClosed;
+
+  @Before
+  public void setUp() throws Exception {
+    source = mock(PipeExtractor.class);
+    sink = mock(PipeConnector.class);
+    subtask =
+        new PipeConfigNodeSubtask(
+            "PipeConfigNodeSubtaskTest",
+            System.currentTimeMillis(),
+            Collections.emptyMap(),
+            Collections.emptyMap(),
+            Collections.singletonMap(
+                PipeSinkConstant.CONNECTOR_KEY,
+                BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName()),
+            new PipeTaskMeta(MinimumProgressIndex.INSTANCE, Integer.MIN_VALUE));
+    final Field sourceField = PipeConfigNodeSubtask.class.getDeclaredField("source");
+    sourceField.setAccessible(true);
+    sourceField.set(subtask, source);
+    final Field sinkField = PipeAbstractSinkSubtask.class.getDeclaredField("outputPipeSink");
+    sinkField.setAccessible(true);
+    sinkField.set(subtask, sink);
+  }
+
+  @After
+  public void tearDown() {
+    if (!subtaskClosed) {
+      subtask.close();
+    }
+  }
+
+  @Test
+  public void testStoppedSubtaskDoesNotPollSource() throws Exception {
+    assertFalse(subtask.executeOnce());
+    verifyZeroInteractions(source, sink);
+  }
+
+  @Test
+  public void testStopDuringSupplyRetainsEventUntilRestart() throws Exception {
+    assertStopDuringSupply(false);
+  }
+
+  @Test
+  public void testDropDuringSupplyReleasesEvent() throws Exception {
+    assertStopDuringSupply(true);
+  }
+
+  private void assertStopDuringSupply(final boolean drop) throws Exception {
+    final EnrichedEvent event = mock(EnrichedEvent.class);
+    final CountDownLatch supplyStarted = new CountDownLatch(1);
+    final CountDownLatch supplyReleased = new CountDownLatch(1);
+    final ExecutorService worker = Executors.newSingleThreadExecutor();
+    when(source.supply())
+        .thenAnswer(
+            invocation -> {
+              supplyStarted.countDown();
+              assertTrue(supplyReleased.await(10, TimeUnit.SECONDS));
+              return event;
+            });
+    subtask.allowSubmittingSelf();
+    try {
+      final Future<Boolean> execution = worker.submit(() -> subtask.executeOnce());
+      assertTrue(supplyStarted.await(10, TimeUnit.SECONDS));
+      assertTrue(subtask.disallowSubmittingSelf());
+      if (drop) {
+        subtask.close();
+        subtaskClosed = true;
+      }
+      supplyReleased.countDown();
+
+      assertFalse(execution.get(10, TimeUnit.SECONDS));
+      assertFalse(subtask.executeOnce());
+      verify(sink, never()).transfer(event);
+      verify(event, never()).decreaseReferenceCount(anyString(), anyBoolean());
+
+      if (drop) {
+        verify(event).clearReferenceCount(anyString());
+      } else {
+        verifyZeroInteractions(sink);
+        subtask.allowSubmittingSelf();
+        assertTrue(subtask.executeOnce());
+        verify(source).supply();
+        verify(sink).transfer(event);
+        verify(event).decreaseReferenceCount(anyString(), eq(true));
+      }
+    } finally {
+      supplyReleased.countDown();
+      worker.shutdownNow();
+      assertTrue(worker.awaitTermination(10, TimeUnit.SECONDS));
+    }
+  }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java

@@ -556,6 +556,9 @@ public boolean internallyDecreaseResourceReferenceCount(final String holderMessa
               holderMessage),
           e);
     } finally {
+      // Retries must repeat the privilege check, so release its metadata only with the event.
+      tableNames = null;
+      treeSchemaMap = null;
       if (Objects.nonNull(pipeName)) {
         try {
           PipeDataNodeSinglePipeMetrics.getInstance()
@@ -932,10 +935,6 @@ else if (Objects.nonNull(treeSchemaMap)) {
                   e.getMessage());
       LOGGER.warn(errorMsg, e);
       throw new PipeException(errorMsg, e);
-    } finally {
-      // GC useless
-      tableNames = null;
-      treeSchemaMap = null;
     }
   }