diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java new file mode 100644 index 0000000000000..68ab36b346edb --- /dev/null +++ b/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 delayQueries(final Class 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)); + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java index b2a7ba4d22e8c..7ce8f63c9a1e3 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java +++ b/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(); diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeInclusionIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeInclusionIT.java index 1385829578fdf..d00f5edbfd6b5 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeInclusionIT.java +++ b/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,", diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeMetaHistoricalIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeMetaHistoricalIT.java index 0827d36217901..baa295c3578f2 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeMetaHistoricalIT.java +++ b/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);