This is an automated email from the ASF dual-hosted git repository. rong pushed a commit to branch iotdb-1528 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit df29b6c0ce30ba6bb46ef3696c3cb57b1997ce7a Author: Steve Yurong Su <[email protected]> AuthorDate: Mon Jul 26 10:27:20 2021 +0800 [IOTDB-1528] CQ IT unstable: sometimes fails on checkCQExecutionResult() --- .../db/integration/IoTDBContinuousQueryIT.java | 103 +++++++-------------- 1 file changed, 36 insertions(+), 67 deletions(-) diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBContinuousQueryIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBContinuousQueryIT.java index b374ed2..f1d2c95 100644 --- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBContinuousQueryIT.java +++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBContinuousQueryIT.java @@ -43,10 +43,6 @@ import java.util.stream.Collectors; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; -/** - * Notice that, all test begins with "IoTDB" is integration test. All test which will start the - * IoTDB server should be defined as integration test. - */ public class IoTDBContinuousQueryIT { private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBContinuousQueryIT.class); @@ -110,7 +106,7 @@ public class IoTDBContinuousQueryIT { } } - String[] timeSeriesArray = { + private final String[] timeSeriesArray = { "root.ln.wf01.wt01.ws01", "root.ln.wf01.wt01.ws02", "root.ln.wf01.wt02.ws01", @@ -146,19 +142,16 @@ public class IoTDBContinuousQueryIT { @Test public void testCreateAndDropContinuousQuery() throws Exception { - createTimeSeries(); statement.execute( "CREATE CONTINUOUS QUERY cq1 " + "BEGIN SELECT max_value(temperature) INTO temperature_max FROM root.ln.*.*.* " + "GROUP BY time(1s) END"); - statement.execute( "CREATE CONTINUOUS QUERY cq2 " + "BEGIN SELECT count(temperature) INTO temperature_cnt FROM root.ln.wf01.*.* " + " GROUP BY time(1s), level=3 END"); - statement.execute( "CREATE CONTINUOUS QUERY cq3 " + "RESAMPLE EVERY 2s FOR 2s " @@ -168,18 +161,15 @@ public class IoTDBContinuousQueryIT { statement.execute("DROP CONTINUOUS QUERY cq1"); statement.execute("DROP CONTINUOUS QUERY cq2"); - checkContinuousQueries(new String[] {"cq3"}); + checkShowContinuousQueriesResult(new String[] {"cq3"}); EnvironmentUtils.shutdownDaemon(); - EnvironmentUtils.stopDaemon(); - setUp(); - checkContinuousQueries(new String[] {"cq3"}); + checkShowContinuousQueriesResult(new String[] {"cq3"}); try { - statement.execute( "CREATE CONTINUOUS QUERY cq3 " + "RESAMPLE EVERY 2s FOR 2s " @@ -190,7 +180,6 @@ public class IoTDBContinuousQueryIT { } try { - statement.execute("DROP CONTINUOUS QUERY cq1"); } catch (Exception e) { assertTrue(e.getMessage().contains("not exist")); @@ -200,13 +189,12 @@ public class IoTDBContinuousQueryIT { "CREATE CONTINUOUS QUERY cq1 " + "BEGIN SELECT max_value(temperature) INTO temperature_max FROM root.ln.*.*.* " + "GROUP BY time(1s) END"); - statement.execute( "CREATE CONTINUOUS QUERY cq2 " + "BEGIN SELECT count(temperature) INTO temperature_cnt FROM root.ln.wf01.*.* " + " GROUP BY time(1s), level=3 END"); - checkContinuousQueries(new String[] {"cq3", "cq1", "cq2"}); + checkShowContinuousQueriesResult(new String[] {"cq3", "cq1", "cq2"}); statement.execute("DROP CONTINUOUS QUERY cq1"); statement.execute("DROP CONTINUOUS QUERY cq2"); @@ -227,7 +215,7 @@ public class IoTDBContinuousQueryIT { Thread.sleep(5500); - checkTimeSeries( + checkShowTimeSeriesResult( new String[] { "root.ln.wf01.wt01.ws01.temperature", "root.ln.wf01.wt01.ws02.temperature", @@ -247,9 +235,8 @@ public class IoTDBContinuousQueryIT { } @Test - public void testContinuousQueryResult() throws Exception { + public void testContinuousQueryResult1() throws Exception { createTimeSeries(); - startDataGenerator(); Thread.sleep(500); @@ -260,24 +247,20 @@ public class IoTDBContinuousQueryIT { + "BEGIN SELECT avg(temperature) INTO temperature_avg FROM root.ln.wf01.*.* " + "GROUP BY time(1s), level=2 END"); - long creationTime = System.currentTimeMillis(); + final long creationTime = System.currentTimeMillis(); Thread.sleep(5500); - boolean hasResult = statement.execute("select temperature_avg from root.ln.wf01"); - Assert.assertTrue(hasResult); - + Assert.assertTrue(statement.execute("select temperature_avg from root.ln.wf01")); checkCQExecutionResult(creationTime, 0, 5000, 1000, 1000, 1000, 2); statement.execute("DROP CQ cq1"); - stopDataGenerator(); } @Test public void testContinuousQueryResult2() throws Exception { createTimeSeries(); - startDataGenerator(); Thread.sleep(500); @@ -288,24 +271,20 @@ public class IoTDBContinuousQueryIT { + "BEGIN SELECT avg(temperature) INTO temperature_avg FROM root.ln.wf01.*.* " + "GROUP BY time(1s), level=2 END"); - long creationTime = System.currentTimeMillis(); + final long creationTime = System.currentTimeMillis(); Thread.sleep(5500); - boolean hasResult = statement.execute("select temperature_avg from root.ln.wf01"); - Assert.assertTrue(hasResult); - + Assert.assertTrue(statement.execute("select temperature_avg from root.ln.wf01")); checkCQExecutionResult(creationTime, 0, 5000, 1000, 2000, 1000, 2); statement.execute("DROP CQ cq1"); - stopDataGenerator(); } @Test public void testContinuousQueryResult3() throws Exception { createTimeSeries(); - startDataGenerator(); Thread.sleep(500); @@ -315,34 +294,29 @@ public class IoTDBContinuousQueryIT { + "BEGIN SELECT avg(temperature) INTO temperature_avg FROM root.ln.wf01.*.* " + "GROUP BY time(1s), level=2 END"); - long creationTime = System.currentTimeMillis(); + final long creationTime = System.currentTimeMillis(); Thread.sleep(5500); - boolean hasResult = statement.execute("select temperature_avg from root.ln.wf01"); - Assert.assertTrue(hasResult); - + Assert.assertTrue(statement.execute("select temperature_avg from root.ln.wf01")); checkCQExecutionResult(creationTime, 0, 5000, 1000, 1000, 1000, 2); statement.execute("DROP CQ cq1"); - stopDataGenerator(); } @Test public void testContinuousQueryResult4() throws Exception { - statement.execute( "CREATE CONTINUOUS QUERY cq1 " + "BEGIN SELECT avg(temperature) INTO temperature_avg FROM root.ln.wf01.*.* " + "GROUP BY time(1s), level=2 END"); - long creationTime = System.currentTimeMillis(); + final long creationTime = System.currentTimeMillis(); Thread.sleep(4500); createTimeSeries(); - startDataGenerator(); Thread.sleep(6000); @@ -350,7 +324,6 @@ public class IoTDBContinuousQueryIT { checkCQExecutionResult(creationTime, 5000, 5500, 1000, 1000, 1000, 2); statement.execute("DROP CQ cq1"); - stopDataGenerator(); } @@ -362,18 +335,19 @@ public class IoTDBContinuousQueryIT { long everyInterval, long groupByInterval, int level) - throws SQLException { - boolean hasResult = statement.execute("select temperature_avg from root.ln.wf01"); - Assert.assertTrue(hasResult); - - List<Pair<Long, String>> result = generateResult(); - - long expectedSize = (duration / everyInterval + 1) * (forInterval / groupByInterval); - Assert.assertEquals(expectedSize, result.size()); - - long leftMost = result.get(0).left + forInterval; - - for (int i = 0; i < result.size(); i++) { + throws SQLException, InterruptedException { + final long expectedSize = (duration / everyInterval + 1) * (forInterval / groupByInterval); + long waitSeconds = 0; + List<Pair<Long, String>> result; + do { + statement.execute("select temperature_avg from root.ln.wf01"); + result = collectQueryResult(); + Thread.sleep(waitSeconds); + waitSeconds += 1000; + } while (result.size() < expectedSize); + + final long leftMost = result.get(0).left + forInterval; + for (int i = 0; i < expectedSize; i++) { long left = result.get(i).left; if (i == 0) { @@ -392,14 +366,13 @@ public class IoTDBContinuousQueryIT { "select avg(temperature) from root.ln.wf01.*.* GROUP BY ([%d, %d), %dms), level=%d", left, left + groupByInterval, groupByInterval, level)); - List<Pair<Long, String>> correctAnswer = generateResult(); + List<Pair<Long, String>> correctAnswer = collectQueryResult(); Assert.assertEquals(1, correctAnswer.size()); - Assert.assertEquals(correctAnswer.get(0).right, result.get(i).right); } } - private List<Pair<Long, String>> generateResult() { + private List<Pair<Long, String>> collectQueryResult() { List<Pair<Long, String>> result = new ArrayList<>(); try (ResultSet resultSet = statement.getResultSet()) { while (resultSet.next()) { @@ -407,21 +380,19 @@ public class IoTDBContinuousQueryIT { String value = resultSet.getString(2); result.add(new Pair<>(Long.parseLong(timestamp), value)); } - } catch (SQLException throwables) { - LOGGER.error(throwables.getMessage()); + } catch (SQLException throwable) { + fail(throwable.getMessage()); } return result; } - private void checkContinuousQueries(String[] continuousQueryArray) throws SQLException { - boolean hasResult = statement.execute("show continuous queries"); - Assert.assertTrue(hasResult); + private void checkShowContinuousQueriesResult(String[] continuousQueryArray) throws SQLException { + Assert.assertTrue(statement.execute("show continuous queries")); List<String> resultList = new ArrayList<>(); try (ResultSet resultSet = statement.getResultSet()) { while (resultSet.next()) { - String cq = resultSet.getString("cq name"); - resultList.add(cq); + resultList.add(resultSet.getString("cq name")); } } Assert.assertEquals(continuousQueryArray.length, resultList.size()); @@ -436,15 +407,13 @@ public class IoTDBContinuousQueryIT { } } - private void checkTimeSeries(String[] timeSeriesArray) throws SQLException { - boolean hasResult = statement.execute("show timeseries"); - Assert.assertTrue(hasResult); + private void checkShowTimeSeriesResult(String[] timeSeriesArray) throws SQLException { + Assert.assertTrue(statement.execute("show timeseries")); List<String> resultList = new ArrayList<>(); try (ResultSet resultSet = statement.getResultSet()) { while (resultSet.next()) { - String timeseries = resultSet.getString("timeseries"); - resultList.add(timeseries); + resultList.add(resultSet.getString("timeseries")); } } Assert.assertEquals(timeSeriesArray.length, resultList.size());
