[To dev/1.3] Restore IT observation checks (#18899, #19095) [do not squash]

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

@@ -0,0 +1,122 @@
+/*
+ * 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.it.utils;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+import org.apache.iotdb.itbase.env.BaseEnv;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({LocalStandaloneIT.class, ClusterIT.class})
+public class IoTDBTestUtilsIT {
+
+  @BeforeClass
+  public static void setUp() throws Exception {
+    EnvFactory.getEnv().initClusterEnvironment();
+  }
+
+  @AfterClass
+  public static void tearDown() {
+    EnvFactory.getEnv().cleanClusterEnvironment();
+  }
+
+  @Test
+  public void testAssertDataAlwaysFailsOnMismatch() {
+    assertThrows(
+        AssertionError.class,
+        () ->
+            TestUtils.assertDataAlwaysOnEnv(
+                EnvFactory.getEnv(), "count databases", "unexpected,", Collections.emptySet(), 1));
+  }
+
+  @Test
+  public void testAssertDataAlwaysObservationPeriod() {
+    assertObservationPeriod(
+        () ->
+            TestUtils.assertDataAlwaysOnEnv(
+                EnvFactory.getEnv(), "count databases", "count,", Collections.singleton("0,"), 1));
+  }
+
+  @Test
+  public void testAssertDataAlwaysIncludesQueryTime() {
+    final AtomicInteger queries = new AtomicInteger();
+    final BaseEnv slowEnv = delayQueries(BaseEnv.class, EnvFactory.getEnv(), queries);
+    assertObservationPeriod(
+        () ->
+            TestUtils.assertDataAlwaysOnEnv(
+                slowEnv, "count databases", "count,", Collections.singleton("0,"), 1));
+    // A successful query taking longer than the observation period must not cause a timeout.
+    assertEquals(1, queries.get());
+  }
+
+  private <T> T delayQueries(final Class<T> type, final T delegate, final AtomicInteger queries) {
+    return type.cast(
+        Proxy.newProxyInstance(
+            type.getClassLoader(),
+            new Class<?>[] {type},
+            (proxy, method, args) -> {
+              if ("executeQuery".equals(method.getName())) {
+                queries.incrementAndGet();
+                TimeUnit.SECONDS.sleep(7);
+              }
+              final Object result;
+              try {
+                result = method.invoke(delegate, args);
+              } catch (InvocationTargetException e) {
+                throw e.getCause();
+              }
+              if (result instanceof Connection) {
+                return delayQueries(Connection.class, (Connection) result, queries);
+              }
+              if (result instanceof Statement) {
+                return delayQueries(Statement.class, (Statement) result, queries);
+              }
+              return result;
+            }));
+  }
+
+  private void assertObservationPeriod(final Runnable assertion) {
+    final long start = System.nanoTime();
+    assertion.run();
+    assertTrue(
+        "The assertion must observe queries for the requested period",
+        System.nanoTime() - start >= TimeUnit.SECONDS.toNanos(1));
+  }
+}

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

@@ -983,27 +983,37 @@ public static void assertDataAlwaysOnEnv(
       long consistentSeconds) {
     try (Connection connection = env.getConnection();
         Statement statement = connection.createStatement()) {
-      // Keep retrying if there are execution failures
-      await()
-          .pollInSameThread()
-          .pollDelay(1L, TimeUnit.SECONDS)
-          .pollInterval(1L, TimeUnit.SECONDS)
-          .atMost(consistentSeconds, TimeUnit.SECONDS)
-          .failFast(
-              () -> {
-                try {
-                  TestUtils.assertResultSetEqual(
-                      executeQueryWithRetry(statement, sql), expectedHeader, expectedResSet);
-                } catch (Exception e) {
-                  Assert.fail();
-                }
-              });
+      assertAlways(
+          consistentSeconds,
+          () -> {
+            try {
+              TestUtils.assertResultSetEqual(
+                  executeQueryWithRetry(statement, sql), expectedHeader, expectedResSet);
+            } catch (Exception e) {
+              Assert.fail();
+            }
+          });
     } catch (Exception e) {
       e.printStackTrace();
       fail();
     }
   }
 
+  /** Observe for a bounded period, including time spent executing and checking each query. */
+  private static void assertAlways(final long consistentSeconds, final Runnable assertion)
+      throws InterruptedException {
+    final long start = System.nanoTime();
+    final long duration = TimeUnit.SECONDS.toNanos(consistentSeconds);
+    while (true) {
+      assertion.run();
+      final long remaining = duration - (System.nanoTime() - start);
+      if (remaining <= 0) {
+        return;
+      }
+      TimeUnit.NANOSECONDS.sleep(Math.min(remaining, TimeUnit.SECONDS.toNanos(1)));
+    }
+  }
+
   public static void stopForciblyAndRestartDataNodes() {
     EnvFactory.getEnv().shutdownForciblyAllDataNodes();
     EnvFactory.getEnv().startAllDataNodes();

integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeInclusionIT.java

@@ -96,7 +96,10 @@ public void testPureSchemaInclusion() throws Exception {
           null);
 
       TestUtils.assertDataAlwaysOnEnv(
-          receiverEnv, "select * from root.**", "Time,", Collections.emptySet());
+          receiverEnv,
+          "select * from root.**",
+          "Time,root.ln.wf01.wt01.status,",
+          Collections.emptySet());
     }
   }
 
@@ -166,8 +169,12 @@ public void testPureSchemaInclusionWithMultiplePattern() throws Exception {
           Arrays.asList("insert into root.ln.wf01.wt01(time,status) values(now(),false)", "flush"),
           null);
 
+      // Keep the header independent of wildcard expansion order.
       TestUtils.assertDataAlwaysOnEnv(
-          receiverEnv, "select * from root.ln.**", "Time,", Collections.emptySet());
+          receiverEnv,
+          "select wf01.wt01.status, wf02.wt01.status from root.ln",
+          "Time,root.ln.wf01.wt01.status,root.ln.wf02.wt01.status,",
+          Collections.emptySet());
     }
   }
 
@@ -239,7 +246,10 @@ public void testPureSchemaInclusionWithExclusionPattern() throws Exception {
           null);
 
       TestUtils.assertDataAlwaysOnEnv(
-          receiverEnv, "select * from root.ln.**", "Time,", Collections.emptySet());
+          receiverEnv,
+          "select * from root.ln.**",
+          "Time,root.ln.wf01.wt01.status,",
+          Collections.emptySet());
     }
   }
 
@@ -277,7 +287,7 @@ public void testAuthExclusion() throws Exception {
       TestUtils.executeNonQuery(senderEnv, "create user `ln_write_user` 'write_pwd'", null);
 
       TestUtils.assertDataAlwaysOnEnv(
-          receiverEnv, "list user", "user,", Collections.singleton("root,"));
+          receiverEnv, "list user", "User,", Collections.singleton("root,"));
     }
   }
 
@@ -320,10 +330,10 @@ public void testAuthInclusionWithPattern() throws Exception {
               "GRANT READ_DATA, WRITE_DATA ON root.** TO USER ln_write_user;"),
           null);
 
-      TestUtils.assertDataAlwaysOnEnv(
+      TestUtils.assertDataEventuallyOnEnv(
           receiverEnv,
           "LIST PRIVILEGES OF USER ln_write_user",
-          "ROLE,PATH,PRIVILEGES,GRANT OPTION,",
+          "Role,Path,Privileges,GrantOption,",
           new HashSet<>(
               Arrays.asList(
                   ",root.**,MANAGE_USER,true,",

integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeMetaHistoricalIT.java

@@ -259,7 +259,7 @@ public void testAuthInclusion() throws Exception {
           "Database,SchemaReplicationFactor,DataReplicationFactor,TimePartitionOrigin,TimePartitionInterval,",
           Collections.emptySet());
       TestUtils.assertDataAlwaysOnEnv(
-          receiverEnv, "select * from root.**", "Time", Collections.emptySet());
+          receiverEnv, "select * from root.**", "Time,", Collections.emptySet());
 
       TestUtils.executeNonQuery(senderEnv, "CREATE ROLE test", null);