Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,129 @@
/*
* 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.TableClusterIT;
import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
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.util.Collections;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

@RunWith(IoTDBTestRunner.class)
@Category({TableLocalStandaloneIT.class, TableClusterIT.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 testAssertAlwaysFail() {
assertObservationPeriod(
() ->
TestUtils.assertAlwaysFail(
EnvFactory.getEnv(), "select * from missing_database.missing_table", 1));
assertThrows(
AssertionError.class,
() -> TestUtils.assertAlwaysFail(EnvFactory.getEnv(), "show databases", 1));
}

@Test
public void testAssertDataAlwaysFailsOnMismatch() {
final BaseEnv env = EnvFactory.getEnv();
final AtomicInteger failures = new AtomicInteger();
final Consumer<String> handleFailure = message -> failures.incrementAndGet();
assertThrows(
AssertionError.class,
() ->
TestUtils.assertDataAlwaysOnEnv(
env, "count databases", "unexpected,", Collections.emptySet(), 1, (String) null));
assertThrows(
AssertionError.class,
() ->
TestUtils.assertDataAlwaysOnEnv(
env, "count databases", "unexpected,", Collections.emptySet(), 1, handleFailure));
assertThrows(
AssertionError.class,
() ->
TestUtils.assertDataAlwaysOnEnv(
env,
"show databases",
"unexpected,",
Collections.emptySet(),
1,
"information_schema",
handleFailure));
assertEquals(2, failures.get());
}

@Test
public void testAssertDataAlwaysObservationPeriod() {
final BaseEnv env = EnvFactory.getEnv();
final Consumer<String> handleFailure = message -> fail(message);
assertObservationPeriod(
() ->
TestUtils.assertDataAlwaysOnEnv(
env, "count databases", "count,", Collections.singleton("0,"), 1, (String) null));
assertObservationPeriod(
() ->
TestUtils.assertDataAlwaysOnEnv(
env, "count databases", "count,", Collections.singleton("0,"), 1, handleFailure));
assertObservationPeriod(
() ->
TestUtils.assertDataAlwaysOnEnv(
env,
"count databases",
"count,",
Collections.singleton("1,"),
1,
"information_schema",
handleFailure));
}

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));
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -26,9 +26,11 @@
import org.apache.iotdb.it.env.cluster.env.AbstractEnv;
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
import org.apache.iotdb.itbase.env.BaseEnv;
import org.apache.iotdb.itbase.runtime.ClusterTestConnection;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.rpc.TSStatusCode;

import org.apache.tsfile.read.common.RowRecord;
import org.junit.Assert;
Expand Down Expand Up @@ -83,6 +85,54 @@ public class TestUtils {

public static final String TIME_PRECISION_IN_NS = "ns";

/**
* Creates a statement on the write node without comparing results across DataNodes.
*
* <p>Use this for node-local diagnostics or changing progress snapshots. Closing the statement
* does not close the supplied connection.
*/
public static Statement createSingleNodeStatement(final Connection connection)
throws SQLException {
final Connection singleNodeConnection =
connection instanceof ClusterTestConnection
? ((ClusterTestConnection) connection).writeConnection.getUnderlyingConnection()
: connection;
return singleNodeConnection.createStatement();
}

/** Drops a table and waits for the asynchronous procedure to finish on a single DataNode. */
public static void dropTable(
final Connection connection, final String database, final String tableName)
throws SQLException {
try (final Statement statement = createSingleNodeStatement(connection)) {
statement.execute("DROP TABLE " + database + "." + tableName);
await()
.atMost(60, TimeUnit.SECONDS)
.pollInSameThread()
.pollInterval(20, TimeUnit.MILLISECONDS)
.until(
() -> {
try (final ResultSet resultSet =
statement.executeQuery(
"SELECT state, error_message FROM information_schema.drop_table_procedures "
+ "WHERE database = '"
+ database.replace("'", "''")
+ "' AND table_name = '"
+ tableName.replace("'", "''")
+ "' ORDER BY procedure_id DESC LIMIT 1")) {
if (!resultSet.next()) {
return false;
}
final String state = resultSet.getString(1);
if ("ROLLEDBACK".equals(state)) {
throw new SQLException(resultSet.getString(2));
}
return "SUCCESS".equals(state);
}
});
}
}

public static String defaultFormatDataTime(long time) {
return RpcUtils.formatDatetime(
RpcUtils.DEFAULT_TIME_FORMAT, TIME_PRECISION_IN_MS, time, DEFAULT_ZONE_ID);
Expand Down Expand Up @@ -969,6 +1019,29 @@ public static void executeTableNonQuery(String sql, String userName, String pass
}
}

public static void dropDatabaseWithRetry(final Statement statement, final String database)
throws SQLException {
// Asynchronous DROP TABLE may still be running when a test starts database cleanup.
final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60);
while (true) {
try {
statement.execute("DROP DATABASE IF EXISTS " + database);
return;
} catch (final SQLException e) {
if (e.getErrorCode() != TSStatusCode.OVERLAP_WITH_EXISTING_TASK.getStatusCode()
|| System.nanoTime() >= deadline) {
throw e;
}
}
try {
Thread.sleep(100);
} catch (final InterruptedException e) {
Thread.currentThread().interrupt();
throw new SQLException(e);
}
}
}

public static void executeNonQueryWithRetry(final BaseEnv env, final String sql) {
executeNonQueryWithRetry(env, sql, SessionConfig.DEFAULT_USER, SessionConfig.DEFAULT_PASSWORD);
}
Expand Down Expand Up @@ -1856,7 +1929,7 @@ public static void assertDataEventuallyOnEnv(
}
}

// Note that this class will accept any exceptions
/** Asserts that a query keeps failing with an SQLException throughout the observation period. */
public static void assertAlwaysFail(final BaseEnv env, final String sql) {
assertAlwaysFail(env, sql, 10);
}
Expand All @@ -1865,13 +1938,15 @@ public static void assertAlwaysFail(
final BaseEnv env, final String sql, final long consistentSeconds) {
try (final Connection connection = env.getConnection(BaseEnv.TABLE_SQL_DIALECT);
final Statement statement = connection.createStatement()) {
// Keep retrying if there are execution failures
await()
.pollInSameThread()
.pollDelay(1L, TimeUnit.SECONDS)
.pollDelay(0L, TimeUnit.SECONDS)
.pollInterval(1L, TimeUnit.SECONDS)
.atMost(consistentSeconds, TimeUnit.SECONDS)
.failFast(() -> Assert.assertThrows(Exception.class, () -> statement.executeQuery(sql)));
.during(consistentSeconds, TimeUnit.SECONDS)
.atMost(consistentSeconds + 5, TimeUnit.SECONDS)
.failFast(
() -> Assert.assertThrows(SQLException.class, () -> statement.executeQuery(sql)))
.until(() -> true);
} catch (final Exception e) {
fail(e.getMessage());
}
Expand Down Expand Up @@ -2004,12 +2079,12 @@ public static void assertDataAlwaysOnEnv(
env.getConnection(
Objects.isNull(database) ? TREE_SQL_DIALECT : BaseEnv.TABLE_SQL_DIALECT);
Statement statement = connection.createStatement()) {
// Keep retrying if there are execution failures
await()
.pollInSameThread()
.pollDelay(1L, TimeUnit.SECONDS)
.pollDelay(0L, TimeUnit.SECONDS)
.pollInterval(1L, TimeUnit.SECONDS)
.atMost(consistentSeconds, TimeUnit.SECONDS)
.during(consistentSeconds, TimeUnit.SECONDS)
.atMost(consistentSeconds + 5, TimeUnit.SECONDS)
.failFast(
() -> {
try {
Expand All @@ -2020,7 +2095,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 > consistentSeconds >> 1) {
&& System.currentTimeMillis() - startTime
> TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) {
flushed[0] = true;
statement.execute("flush");
}
Expand All @@ -2029,7 +2105,8 @@ public static void assertDataAlwaysOnEnv(
} catch (Exception e) {
Assert.fail();
}
});
})
.until(() -> true);
Comment on lines +2108 to +2109

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Fix the remaining lifecycle expectations when enabling continuous assertions

This terminal await also activates the previously dormant checks in IoTDBPipeLifeCycleIT.testLifeCycleWithHistoryDisabled, but that caller has not been updated. After stopPipe("p1"), root.ln0 is created only on the sender, while the receiver retains the two existing series under root.db. The receiver's series count under root.ln0, series count under root.db, and database count for root.ln0 should therefore be 0, 2, and 0; the assertions at lines 227–238 still expect 1 for all three.

I reproduced the failure from a clean reactor build with MultiClusterIT and two 1-ConfigNode/3-DataNode clusters: the first incorrect assertion fails at IoTDBPipeLifeCycleIT.java:227, with actual 0 versus expected 1. Please update all three expectations together with this helper change so the existing tree-model pipe test remains runnable.

@Caideyipi Caideyipi Oct 16, 2026 •

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Applied in f15d245. The lifecycle assertions now expect 0 series under root.ln0, 2 under root.db, and 0 databases under root.ln0. Verified testLifeCycleWithHistoryDisabled with two 1-ConfigNode/3-DataNode clusters and the original observation periods; it passed. The local run used 32 MiB initial node heaps, Serial GC, and ActiveProcessorCount=2 to fit available memory; maximum node heaps and cluster topology were unchanged.

} catch (Exception e) {
e.printStackTrace();
fail();
Expand All @@ -2049,17 +2126,19 @@ public static void assertDataAlwaysOnEnv(
Statement statement = connection.createStatement()) {
await()
.pollInSameThread()
.pollDelay(1L, TimeUnit.SECONDS)
.pollDelay(0L, TimeUnit.SECONDS)
.pollInterval(1L, TimeUnit.SECONDS)
.atMost(consistentSeconds, TimeUnit.SECONDS)
.during(consistentSeconds, TimeUnit.SECONDS)
.atMost(consistentSeconds + 5, TimeUnit.SECONDS)
.failFast(
() -> {
try {
// 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 > consistentSeconds >> 1) {
&& System.currentTimeMillis() - startTime
> TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) {
flushed[0] = true;
statement.execute("flush");
}
Expand All @@ -2076,7 +2155,8 @@ public static void assertDataAlwaysOnEnv(
}
throw e;
}
});
})
.until(() -> true);
} catch (Exception e) {
e.printStackTrace();
fail();
Expand All @@ -2093,14 +2173,15 @@ public static void assertDataAlwaysOnEnv(
Consumer<String> handleFailure) {
final long startTime = System.currentTimeMillis();
final boolean[] flushed = {false};
try (Connection connection = env.getConnection();
try (Connection connection =
env.getConnection(database == null ? TREE_SQL_DIALECT : BaseEnv.TABLE_SQL_DIALECT);
Statement statement = connection.createStatement()) {
// Keep retrying if there are execution failures
await()
.pollInSameThread()
.pollDelay(1L, TimeUnit.SECONDS)
.pollDelay(0L, TimeUnit.SECONDS)
.pollInterval(1L, TimeUnit.SECONDS)
.atMost(consistentSeconds, TimeUnit.SECONDS)
.during(consistentSeconds, TimeUnit.SECONDS)
.atMost(consistentSeconds + 5, TimeUnit.SECONDS)
.failFast(
() -> {
try {
Expand All @@ -2111,7 +2192,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 > consistentSeconds >> 1) {
&& System.currentTimeMillis() - startTime
> TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) {
flushed[0] = true;
statement.execute("flush");
}
Expand All @@ -2128,7 +2210,8 @@ public static void assertDataAlwaysOnEnv(
}
throw e;
}
});
})
.until(() -> true);
} catch (Exception e) {
e.printStackTrace();
fail();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -324,7 +324,7 @@ public void testReceiverPermission() throws Exception {
"show databases",
"Database,TTL(ms),SchemaReplicationFactor,DataReplicationFactor,TimePartitionInterval,",
Collections.singleton("information_schema,INF,null,null,null,"),
(String) null);
"information_schema");

TestUtils.executeNonQuery(
"information_schema",
Expand Down Expand Up @@ -374,7 +374,11 @@ public void testReceiverPermission() throws Exception {

// Shall not be transferred
TestUtils.assertDataAlwaysOnEnv(
receiverEnv, "count databases", "count,", Collections.singleton("2,"), (String) null);
receiverEnv,
"count databases",
"count,",
Collections.singleton("2,"),
"information_schema");

TestUtils.executeNonQuery(
"information_schema",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -414,7 +414,8 @@ public void testLoadAutoCreateWithTableDeletion() throws Exception {
statement.execute("INSERT INTO t1(time,tag1,tag2,s1,s2) values(1, 'd1', 'd2', 'red', 1)");
statement.execute("INSERT INTO t1(time,tag1,tag2,s1,s2) values(2, 'd1', 'd2', 'blue', 2)");
statement.execute("flush");
statement.execute("drop table t1");
// Start historical transfer only after the table and its data have been deleted.
TestUtils.dropTable(connection, "db", "t1");
} catch (Exception e) {
fail(e.getMessage());
}
Expand Down
Loading
Loading