diff --git a/example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java b/example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java index 62c869e02bf0a..3aea4e0a76245 100644 --- a/example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java +++ b/example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java @@ -30,9 +30,13 @@ import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet; import org.apache.tsfile.read.TsFileReader; +import org.apache.tsfile.read.common.Path; +import org.apache.tsfile.read.expression.QueryExpression; +import org.apache.tsfile.read.query.dataset.QueryDataSet; import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Properties; @@ -46,11 +50,12 @@ public class SubscriptionSessionExample { private static final int PORT = 6667; private static final String TOPIC_1 = "topic1"; - private static final String TOPIC_2 = "`topic2`"; + private static final String TOPIC_2 = "`'topic2'`"; - public static final long SLEEP_NS = 1_000_000_000L; - public static final long POLL_TIMEOUT_MS = 10_000L; + private static final long SLEEP_NS = 1_000_000_000L; + private static final long POLL_TIMEOUT_MS = 10_000L; private static final int MAX_RETRY_TIMES = 3; + private static final int PARALLELISM = 8; private static void prepareData() throws Exception { // Open session @@ -66,13 +71,13 @@ private static void prepareData() throws Exception { // Insert some historical data final long currentTime = System.currentTimeMillis(); - for (int i = 0; i < 10000; ++i) { + for (int i = 0; i < 100; ++i) { session.executeNonQueryStatement( String.format("insert into root.db.d1(time, s1, s2) values (%s, 1, 2)", i)); session.executeNonQueryStatement( - String.format("insert into root.db.d2(time, s3, s4) values (%s, 3, 4)", currentTime + i)); + String.format("insert into root.db.d2(time, s1, s2) values (%s, 3, 4)", currentTime + i)); session.executeNonQueryStatement( - String.format("insert into root.sg.d3(time, s5) values (%s, 5)", currentTime + 2 * i)); + String.format("insert into root.sg.d3(time, s1) values (%s, 5)", currentTime + 2 * i)); } session.executeNonQueryStatement("flush"); @@ -104,11 +109,16 @@ private static void dataQuery() throws Exception { session = null; } - private static void subscriptionExample1() throws Exception { + /** single pull consumer subscribe topic with path and time range */ + private static void dataSubscription1() throws Exception { // Create topics try (final SubscriptionSession subscriptionSession = new SubscriptionSession(HOST, PORT)) { subscriptionSession.open(); - subscriptionSession.createTopic(TOPIC_1); + final Properties config = new Properties(); + config.put(TopicConstant.PATH_KEY, "root.db.d1.s1"); + config.put(TopicConstant.START_TIME_KEY, 25); + config.put(TopicConstant.END_TIME_KEY, 75); + subscriptionSession.createTopic(TOPIC_1, config); } int retryCount = 0; @@ -151,7 +161,8 @@ private static void subscriptionExample1() throws Exception { consumer1.close(); } - private static void subscriptionExample2() throws Exception { + /** multi pull consumer subscribe topic with tsfile format */ + private static void dataSubscription2() throws Exception { try (final SubscriptionSession subscriptionSession = new SubscriptionSession(HOST, PORT)) { subscriptionSession.open(); final Properties config = new Properties(); @@ -160,7 +171,7 @@ private static void subscriptionExample2() throws Exception { } final List threads = new ArrayList<>(); - for (int i = 0; i < 8; ++i) { + for (int i = 0; i < PARALLELISM; ++i) { final int idx = i; final Thread thread = new Thread( @@ -187,7 +198,16 @@ private static void subscriptionExample2() throws Exception { } for (final SubscriptionMessage message : messages) { try (final TsFileReader reader = message.getTsFileHandler().openReader()) { - // do something... + final QueryDataSet dataSet = + reader.query( + QueryExpression.create( + Arrays.asList( + new Path("root.db.d2", "s2", true), + new Path("root.db.d3", "s1", true)), + null)); + while (dataSet.hasNext()) { + System.out.println(dataSet.next()); + } } } consumer2.commitSync(messages); @@ -208,8 +228,8 @@ private static void subscriptionExample2() throws Exception { public static void main(final String[] args) throws Exception { prepareData(); - dataQuery(); - subscriptionExample1(); - subscriptionExample2(); + // dataQuery(); + // dataSubscription1(); + dataSubscription2(); } } diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/AbstractSubscriptionIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/AbstractSubscriptionIT.java new file mode 100644 index 0000000000000..a168dbc78ba72 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/AbstractSubscriptionIT.java @@ -0,0 +1,49 @@ +/* + * 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.subscription.it; + +import org.apache.iotdb.session.subscription.consumer.SubscriptionExecutorServiceManager; + +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.rules.TestName; +import org.junit.rules.TestRule; + +public abstract class AbstractSubscriptionIT { + + @Rule public TestName testName = new TestName(); + + @Rule public final TestRule skipOnSetUpFailure = new SkipOnSetUpFailure("setUp"); + + @Before + public void setUp() { + // set thread name + Thread.currentThread().setName(String.format("%s - main", testName.getMethodName())); + + // set thread pools core size + SubscriptionExecutorServiceManager.setControlFlowExecutorCorePoolSize(1); + SubscriptionExecutorServiceManager.setUpstreamDataFlowExecutorCorePoolSize(1); + SubscriptionExecutorServiceManager.setDownstreamDataFlowExecutorCorePoolSize(1); + } + + @After + public void tearDown() {} +} diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/IoTDBSubscriptionITConstant.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/IoTDBSubscriptionITConstant.java index 7d16110cd1f00..840abb4ccb5b5 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/IoTDBSubscriptionITConstant.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/IoTDBSubscriptionITConstant.java @@ -19,11 +19,24 @@ package org.apache.iotdb.subscription.it; +import org.awaitility.Awaitility; +import org.awaitility.core.ConditionFactory; + +import java.util.concurrent.TimeUnit; + public class IoTDBSubscriptionITConstant { - public static final long AWAITILITY_POLL_DELAY_SECOND = 1L; - public static final long AWAITILITY_POLL_INTERVAL_SECOND = 2L; - public static final long AWAITILITY_AT_MOST_SECOND = 600L; + private static final long AWAITILITY_POLL_DELAY_SECOND = 1L; + private static final long AWAITILITY_POLL_INTERVAL_SECOND = 2L; + private static final long AWAITILITY_AT_MOST_SECOND = 600L; + + public static final ConditionFactory AWAIT = + Awaitility.await() + .pollInSameThread() + .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) + .pollInterval( + IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) + .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS); public static final long SLEEP_NS = 1_000_000_000L; public static final long POLL_TIMEOUT_MS = 10_000L; diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionRestartIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/cluster/IoTDBSubscriptionRestartIT.java similarity index 90% rename from integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionRestartIT.java rename to integration-test/src/test/java/org/apache/iotdb/subscription/it/cluster/IoTDBSubscriptionRestartIT.java index d3680219578ab..41dd654383d66 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionRestartIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/cluster/IoTDBSubscriptionRestartIT.java @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.iotdb.subscription.it.local; +package org.apache.iotdb.subscription.it.cluster; import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionReq; @@ -36,17 +36,14 @@ import org.apache.iotdb.session.subscription.consumer.SubscriptionPullConsumer; import org.apache.iotdb.session.subscription.payload.SubscriptionMessage; import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet; +import org.apache.iotdb.subscription.it.AbstractSubscriptionIT; import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant; -import org.apache.iotdb.subscription.it.SkipOnSetUpFailure; -import org.awaitility.Awaitility; import org.junit.After; import org.junit.Assert; import org.junit.Before; -import org.junit.Rule; import org.junit.Test; import org.junit.experimental.categories.Category; -import org.junit.rules.TestRule; import org.junit.runner.RunWith; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -54,22 +51,23 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.LockSupport; +import static org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant.AWAIT; import static org.junit.Assert.fail; @RunWith(IoTDBTestRunner.class) @Category({ClusterIT.class}) -public class IoTDBSubscriptionRestartIT { +public class IoTDBSubscriptionRestartIT extends AbstractSubscriptionIT { private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBSubscriptionRestartIT.class); - @Rule public final TestRule skipOnSetUpFailure = new SkipOnSetUpFailure("setUp"); - + @Override @Before - public void setUp() throws Exception { + public void setUp() { + super.setUp(); + EnvFactory.getEnv() .getConfig() .getCommonConfig() @@ -82,8 +80,11 @@ public void setUp() throws Exception { EnvFactory.getEnv().initClusterEnvironment(3, 3); } + @Override @After - public void tearDown() throws Exception { + public void tearDown() { + super.tearDown(); + EnvFactory.getEnv().cleanClusterEnvironment(); } @@ -195,18 +196,14 @@ public void testSubscriptionAfterRestartCluster() throws Exception { } finally { LOGGER.info("consumer exiting..."); } - }); + }, + String.format("%s - %s", testName.getMethodName(), consumer)); thread.start(); // Check timestamps size try { // Keep retrying if there are execution failures - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted(() -> Assert.assertEquals(100, timestamps.size())); + AWAIT.untilAsserted(() -> Assert.assertEquals(100, timestamps.size())); } catch (final Exception e) { e.printStackTrace(); fail(e.getMessage()); @@ -312,7 +309,8 @@ public void testSubscriptionAfterRestartDataNode() throws Exception { } finally { LOGGER.info("consumer exiting..."); } - }); + }, + String.format("%s - %s", testName.getMethodName(), consumer)); thread.start(); // Start DN 1 & DN 2 @@ -343,12 +341,7 @@ public void testSubscriptionAfterRestartDataNode() throws Exception { // Check timestamps size try { // Keep retrying if there are execution failures - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted(() -> Assert.assertEquals(200, timestamps.size())); + AWAIT.untilAsserted(() -> Assert.assertEquals(200, timestamps.size())); } catch (final Exception e) { e.printStackTrace(); fail(e.getMessage()); @@ -443,7 +436,8 @@ public void testSubscriptionWhenConfigNodeLeaderChange() throws Exception { } finally { LOGGER.info("consumer exiting..."); } - }); + }, + String.format("%s - %s", testName.getMethodName(), consumer)); thread.start(); // Shutdown leader CN @@ -486,12 +480,7 @@ public void testSubscriptionWhenConfigNodeLeaderChange() throws Exception { // Check timestamps size try { // Keep retrying if there are execution failures - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted(() -> Assert.assertEquals(200, timestamps.size())); + AWAIT.untilAsserted(() -> Assert.assertEquals(200, timestamps.size())); } catch (final Exception e) { e.printStackTrace(); fail(e.getMessage()); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/AbstractSubscriptionDualIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/AbstractSubscriptionDualIT.java index 3d8eb45a7fdfe..fb25cdc3a6c09 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/AbstractSubscriptionDualIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/AbstractSubscriptionDualIT.java @@ -21,29 +21,20 @@ import org.apache.iotdb.it.env.MultiEnvFactory; import org.apache.iotdb.itbase.env.BaseEnv; -import org.apache.iotdb.session.subscription.consumer.SubscriptionExecutorServiceManager; +import org.apache.iotdb.subscription.it.AbstractSubscriptionIT; import org.junit.After; import org.junit.Before; -import org.junit.Rule; -import org.junit.rules.TestName; -abstract class AbstractSubscriptionDualIT { +abstract class AbstractSubscriptionDualIT extends AbstractSubscriptionIT { protected BaseEnv senderEnv; protected BaseEnv receiverEnv; - @Rule public TestName testName = new TestName(); - + @Override @Before public void setUp() { - // set thread name - Thread.currentThread().setName(String.format("%s - main", testName.getMethodName())); - - // set thread pools core size - SubscriptionExecutorServiceManager.setControlFlowExecutorCorePoolSize(1); - SubscriptionExecutorServiceManager.setUpstreamDataFlowExecutorCorePoolSize(1); - SubscriptionExecutorServiceManager.setDownstreamDataFlowExecutorCorePoolSize(1); + super.setUp(); MultiEnvFactory.createEnv(2); senderEnv = MultiEnvFactory.getEnv(0); @@ -55,7 +46,7 @@ public void setUp() { receiverEnv.initClusterEnvironment(); } - void setUpConfig() { + protected void setUpConfig() { // enable auto create schema senderEnv.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true); receiverEnv.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(true); @@ -65,8 +56,11 @@ void setUpConfig() { receiverEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000); } + @Override @After - public final void tearDown() { + public void tearDown() { + super.tearDown(); + senderEnv.cleanClusterEnvironment(); receiverEnv.cleanClusterEnvironment(); } diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionConsumerGroupIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionConsumerGroupIT.java index db7b9e7ca3a51..650f861f498d0 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionConsumerGroupIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionConsumerGroupIT.java @@ -42,7 +42,6 @@ import org.apache.tsfile.read.expression.QueryExpression; import org.apache.tsfile.read.query.dataset.QueryDataSet; import org.apache.tsfile.utils.Pair; -import org.awaitility.Awaitility; import org.junit.Assert; import org.junit.Before; import org.junit.Test; @@ -62,11 +61,11 @@ import java.util.Map; import java.util.Properties; import java.util.Set; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.LockSupport; import java.util.stream.Collectors; +import static org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant.AWAIT; import static org.junit.Assert.fail; @RunWith(IoTDBTestRunner.class) @@ -109,7 +108,7 @@ static final class SubscriptionInfo { } @Override - void setUpConfig() { + protected void setUpConfig() { super.setUpConfig(); // Enable air gap receiver @@ -1015,22 +1014,16 @@ private void pollMessagesAndCheck( try (final Connection connection = receiverEnv.getConnection(); final Statement statement = connection.createStatement()) { // Keep retrying if there are execution failures - Awaitility.await() - .pollInSameThread() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted( - () -> { - if (receiverCrashed.get()) { - LOGGER.info("detect receiver crashed, skipping this test..."); - return; - } - TestUtils.assertSingleResultSetEqual( - TestUtils.executeQueryWithRetry(statement, "select count(*) from root.**"), - expectedHeaderWithResult); - }); + AWAIT.untilAsserted( + () -> { + if (receiverCrashed.get()) { + LOGGER.info("detect receiver crashed, skipping this test..."); + return; + } + TestUtils.assertSingleResultSetEqual( + TestUtils.executeQueryWithRetry(statement, "select count(*) from root.**"), + expectedHeaderWithResult); + }); } } catch (final Exception e) { e.printStackTrace(); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTimePrecisionIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTimePrecisionIT.java index 2b1cc407b7f79..39a9f2225f6bb 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTimePrecisionIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTimePrecisionIT.java @@ -30,7 +30,6 @@ import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant; import org.apache.tsfile.write.record.Tablet; -import org.awaitility.Awaitility; import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; @@ -43,10 +42,10 @@ import java.util.Iterator; import java.util.List; import java.util.Properties; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.LockSupport; +import static org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant.AWAIT; import static org.junit.Assert.fail; @RunWith(IoTDBTestRunner.class) @@ -57,7 +56,7 @@ public class IoTDBSubscriptionTimePrecisionIT extends AbstractSubscriptionDualIT LoggerFactory.getLogger(IoTDBSubscriptionTimePrecisionIT.class); @Override - void setUpConfig() { + protected void setUpConfig() { super.setUpConfig(); // Set timestamp precision to nanosecond @@ -160,7 +159,8 @@ public void testTopicTimePrecision() throws Exception { } finally { LOGGER.info("consumer exiting..."); } - }); + }, + String.format("%s - consumer", testName.getMethodName())); thread.start(); // Check data on receiver @@ -168,21 +168,16 @@ public void testTopicTimePrecision() throws Exception { try (final Connection connection = receiverEnv.getConnection(); final Statement statement = connection.createStatement()) { // Keep retrying if there are execution failures - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted( - () -> - TestUtils.assertSingleResultSetEqual( - TestUtils.executeQueryWithRetry(statement, "select count(*) from root.**"), - new HashMap() { - { - put("count(root.db.d1.s2)", "100"); - put("count(root.db.d2.s1)", "100"); - } - })); + AWAIT.untilAsserted( + () -> + TestUtils.assertSingleResultSetEqual( + TestUtils.executeQueryWithRetry(statement, "select count(*) from root.**"), + new HashMap() { + { + put("count(root.db.d1.s2)", "100"); + put("count(root.db.d2.s1)", "100"); + } + })); } } catch (final Exception e) { e.printStackTrace(); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java index 0cd6bc0d0a106..ac7e83339844b 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java @@ -34,7 +34,6 @@ import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant; import org.apache.tsfile.write.record.Tablet; -import org.awaitility.Awaitility; import org.junit.Assert; import org.junit.Test; import org.junit.experimental.categories.Category; @@ -51,11 +50,11 @@ import java.util.List; import java.util.Properties; import java.util.Set; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.LockSupport; +import static org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant.AWAIT; import static org.junit.Assert.fail; @RunWith(IoTDBTestRunner.class) @@ -136,7 +135,8 @@ public void testTopicPathSubscription() throws Exception { } finally { LOGGER.info("consumer exiting..."); } - }); + }, + String.format("%s - consumer", testName.getMethodName())); thread.start(); // Check data on receiver @@ -144,21 +144,16 @@ public void testTopicPathSubscription() throws Exception { try (final Connection connection = receiverEnv.getConnection(); final Statement statement = connection.createStatement()) { // Keep retrying if there are execution failures - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted( - () -> - TestUtils.assertSingleResultSetEqual( - TestUtils.executeQueryWithRetry(statement, "select count(*) from root.**"), - new HashMap() { - { - put("count(root.db.d1.s)", "100"); - put("count(root.db.d2.s)", "100"); - } - })); + AWAIT.untilAsserted( + () -> + TestUtils.assertSingleResultSetEqual( + TestUtils.executeQueryWithRetry(statement, "select count(*) from root.**"), + new HashMap() { + { + put("count(root.db.d1.s)", "100"); + put("count(root.db.d2.s)", "100"); + } + })); } } catch (final Exception e) { e.printStackTrace(); @@ -238,7 +233,8 @@ public void testTopicTimeSubscription() throws Exception { } finally { LOGGER.info("consumer exiting..."); } - }); + }, + String.format("%s - consumer", testName.getMethodName())); thread.start(); // Check data on receiver @@ -246,20 +242,15 @@ public void testTopicTimeSubscription() throws Exception { try (final Connection connection = receiverEnv.getConnection(); final Statement statement = connection.createStatement()) { // Keep retrying if there are execution failures - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted( - () -> - TestUtils.assertSingleResultSetEqual( - TestUtils.executeQueryWithRetry(statement, "select count(*) from root.**"), - new HashMap() { - { - put("count(root.db.d2.s)", "100"); - } - })); + AWAIT.untilAsserted( + () -> + TestUtils.assertSingleResultSetEqual( + TestUtils.executeQueryWithRetry(statement, "select count(*) from root.**"), + new HashMap() { + { + put("count(root.db.d2.s)", "100"); + } + })); } } catch (final Exception e) { e.printStackTrace(); @@ -336,7 +327,8 @@ public void testTopicProcessorSubscription() throws Exception { } finally { LOGGER.info("consumer exiting..."); } - }); + }, + String.format("%s - consumer", testName.getMethodName())); thread.start(); // Check data on receiver @@ -348,17 +340,12 @@ public void testTopicProcessorSubscription() throws Exception { try (final Connection connection = receiverEnv.getConnection(); final Statement statement = connection.createStatement()) { // Keep retrying if there are execution failures - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted( - () -> - TestUtils.assertResultSetEqual( - TestUtils.executeQueryWithRetry(statement, "select * from root.**"), - "Time,root.db.d1.at1,", - expectedResSet)); + AWAIT.untilAsserted( + () -> + TestUtils.assertResultSetEqual( + TestUtils.executeQueryWithRetry(statement, "select * from root.**"), + "Time,root.db.d1.at1,", + expectedResSet)); } } catch (final Exception e) { e.printStackTrace(); @@ -392,9 +379,9 @@ public void testTopicNameWithBackQuote() throws Exception { } // Create topic on sender - final String topic1 = "`topic1`"; - final String topic2 = "`'topic2'`"; - final String topic3 = "`\"topic3\"`"; + final String topic1 = "`topic4`"; + final String topic2 = "`'topic5'`"; + final String topic3 = "`\"topic6\"`"; final String host = senderEnv.getIP(); final int port = Integer.parseInt(senderEnv.getPort()); try (final SubscriptionSession session = new SubscriptionSession(host, port)) { @@ -464,7 +451,8 @@ public void testTopicNameWithBackQuote() throws Exception { } finally { LOGGER.info("consumer exiting..."); } - }); + }, + String.format("%s - consumer", testName.getMethodName())); thread.start(); // Check data on receiver @@ -472,20 +460,15 @@ public void testTopicNameWithBackQuote() throws Exception { try (final Connection connection = receiverEnv.getConnection(); final Statement statement = connection.createStatement()) { // Keep retrying if there are execution failures - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted( - () -> - TestUtils.assertSingleResultSetEqual( - TestUtils.executeQueryWithRetry(statement, "select count(*) from root.**"), - new HashMap() { - { - put("count(root.db.d1.s)", "300"); - } - })); + AWAIT.untilAsserted( + () -> + TestUtils.assertSingleResultSetEqual( + TestUtils.executeQueryWithRetry(statement, "select count(*) from root.**"), + new HashMap() { + { + put("count(root.db.d1.s)", "300"); + } + })); } } catch (final Exception e) { e.printStackTrace(); @@ -507,7 +490,7 @@ public void testTopicInvalidTimeRangeConfig() throws Exception { final Properties properties = new Properties(); properties.put(TopicConstant.START_TIME_KEY, "2024-01-32"); properties.put(TopicConstant.END_TIME_KEY, TopicConstant.NOW_TIME_VALUE); - session.createTopic("topic1", properties); + session.createTopic("topic7", properties); fail(); } catch (final Exception ignored) { } @@ -519,7 +502,7 @@ public void testTopicInvalidTimeRangeConfig() throws Exception { final Properties properties = new Properties(); properties.put(TopicConstant.START_TIME_KEY, "2001.01.01T08:00:00"); properties.put(TopicConstant.END_TIME_KEY, "2000.01.01T08:00:00"); - session.createTopic("topic2", properties); + session.createTopic("topic8", properties); fail(); } catch (final Exception ignored) { } @@ -532,7 +515,7 @@ public void testTopicInvalidPathConfig() throws Exception { final Properties config = new Properties(); config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE); config.put(TopicConstant.PATH_KEY, "root.db.*.s"); - testTopicInvalidRuntimeConfigTemplate("topic3", config); + testTopicInvalidRuntimeConfigTemplate("topic9", config); } @Test @@ -543,7 +526,7 @@ public void testTopicInvalidProcessorConfig() throws Exception { config.put("processor", "tumbling-time-sampling-processor"); config.put("processor.tumbling-time.interval-seconds", "1"); config.put("processor.down-sampling.split-file", "true"); - testTopicInvalidRuntimeConfigTemplate("topic4", config); + testTopicInvalidRuntimeConfigTemplate("topic10", config); } private void testTopicInvalidRuntimeConfigTemplate( @@ -600,7 +583,8 @@ private void testTopicInvalidRuntimeConfigTemplate( } finally { LOGGER.info("consumer exiting..."); } - })); + }, + String.format("%s - consumer", testName.getMethodName()))); // Insert some realtime data on sender threads.add( @@ -628,18 +612,15 @@ private void testTopicInvalidRuntimeConfigTemplate( fail(e.getMessage()); } dataPrepared.set(true); - })); + }, + String.format("%s - data inserter", testName.getMethodName()))); for (final Thread thread : threads) { thread.start(); } - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval(IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - // The expected SubscriptionRuntimeCriticalException was not thrown if result is false - .untilTrue(result); + // The expected SubscriptionRuntimeCriticalException was not thrown if result is false + AWAIT.untilTrue(result); for (final Thread thread : threads) { thread.join(); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/AbstractSubscriptionLocalIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/AbstractSubscriptionLocalIT.java new file mode 100644 index 0000000000000..bb248f745d4ed --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/AbstractSubscriptionLocalIT.java @@ -0,0 +1,45 @@ +/* + * 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.subscription.it.local; + +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.subscription.it.AbstractSubscriptionIT; + +import org.junit.After; +import org.junit.Before; + +abstract class AbstractSubscriptionLocalIT extends AbstractSubscriptionIT { + + @Override + @Before + public void setUp() { + super.setUp(); + + EnvFactory.getEnv().initClusterEnvironment(); + } + + @Override + @After + public void tearDown() { + super.tearDown(); + + EnvFactory.getEnv().cleanClusterEnvironment(); + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionBasicIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionBasicIT.java index 13a5e8abc2617..fccbee6d95da7 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionBasicIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionBasicIT.java @@ -40,10 +40,7 @@ import org.apache.tsfile.read.common.Path; import org.apache.tsfile.read.expression.QueryExpression; import org.apache.tsfile.read.query.dataset.QueryDataSet; -import org.awaitility.Awaitility; -import org.junit.After; import org.junit.Assert; -import org.junit.Before; import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; @@ -57,31 +54,21 @@ import java.util.Collections; import java.util.List; import java.util.Properties; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.LockSupport; import java.util.stream.Collectors; +import static org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant.AWAIT; import static org.junit.Assert.fail; @RunWith(IoTDBTestRunner.class) @Category({LocalStandaloneIT.class}) -public class IoTDBSubscriptionBasicIT { +public class IoTDBSubscriptionBasicIT extends AbstractSubscriptionLocalIT { private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBSubscriptionBasicIT.class); - @Before - public void setUp() throws Exception { - EnvFactory.getEnv().initClusterEnvironment(); - } - - @After - public void tearDown() throws Exception { - EnvFactory.getEnv().cleanClusterEnvironment(); - } - @Test public void testBasicSubscribeTablets() throws Exception { // Insert some historical data @@ -146,18 +133,14 @@ public void testBasicSubscribeTablets() throws Exception { } finally { LOGGER.info("consumer exiting..."); } - }); + }, + String.format("%s - consumer", testName.getMethodName())); thread.start(); // Check row count try { // Keep retrying if there are execution failures - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted(() -> Assert.assertEquals(100, rowCount.get())); + AWAIT.untilAsserted(() -> Assert.assertEquals(100, rowCount.get())); } catch (final Exception e) { e.printStackTrace(); fail(e.getMessage()); @@ -242,18 +225,14 @@ public void testBasicSubscribeTsFile() throws Exception { } finally { LOGGER.info("consumer exiting..."); } - }); + }, + String.format("%s - consumer", testName.getMethodName())); thread.start(); // Check row count try { // Keep retrying if there are execution failures - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted(() -> Assert.assertEquals(100, rowCount.get())); + AWAIT.untilAsserted(() -> Assert.assertEquals(100, rowCount.get())); } catch (final Exception e) { e.printStackTrace(); fail(e.getMessage()); @@ -379,18 +358,14 @@ public void onFailure(final Throwable e) { } finally { LOGGER.info("consumer exiting..."); } - }); + }, + String.format("%s - consumer", testName.getMethodName())); thread.start(); // Check row count try { // Keep retrying if there are execution failures - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted(() -> Assert.assertEquals(100, rowCount.get())); + AWAIT.untilAsserted(() -> Assert.assertEquals(100, rowCount.get())); Assert.assertTrue(commitSuccessCount.get() > lastCommitSuccessCount.get()); Assert.assertEquals(0, commitFailureCount.get()); } catch (final Exception e) { @@ -416,12 +391,7 @@ public void onFailure(final Throwable e) { // Check row count try { // Keep retrying if there are execution failures - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted(() -> Assert.assertEquals(200, rowCount.get())); + AWAIT.untilAsserted(() -> Assert.assertEquals(200, rowCount.get())); Assert.assertTrue(commitSuccessCount.get() > lastCommitSuccessCount.get()); Assert.assertEquals(0, commitFailureCount.get()); } catch (final Exception e) { @@ -486,16 +456,11 @@ public void testBasicPushConsumer() { consumer.subscribe(topicName); // The push consumer should automatically poll 10 rows of data by 1 onReceive() - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted( - () -> { - Assert.assertEquals(10, rowCount.get()); - Assert.assertTrue(onReceiveCount.get() > lastOnReceiveCount.get()); - }); + AWAIT.untilAsserted( + () -> { + Assert.assertEquals(10, rowCount.get()); + Assert.assertTrue(onReceiveCount.get() > lastOnReceiveCount.get()); + }); lastOnReceiveCount.set(onReceiveCount.get()); @@ -508,16 +473,11 @@ public void testBasicPushConsumer() { session.executeNonQueryStatement("flush"); } - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted( - () -> { - Assert.assertEquals(20, rowCount.get()); - Assert.assertTrue(onReceiveCount.get() > lastOnReceiveCount.get()); - }); + AWAIT.untilAsserted( + () -> { + Assert.assertEquals(20, rowCount.get()); + Assert.assertTrue(onReceiveCount.get() > lastOnReceiveCount.get()); + }); lastOnReceiveCount.set(onReceiveCount.get()); @@ -529,16 +489,11 @@ public void testBasicPushConsumer() { session.executeNonQueryStatement("flush"); } - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted( - () -> { - Assert.assertEquals(30, rowCount.get()); - Assert.assertTrue(onReceiveCount.get() > lastOnReceiveCount.get()); - }); + AWAIT.untilAsserted( + () -> { + Assert.assertEquals(30, rowCount.get()); + Assert.assertTrue(onReceiveCount.get() > lastOnReceiveCount.get()); + }); consumer.unsubscribe(topicName); } catch (final Exception e) { @@ -624,22 +579,18 @@ public void testPollUnsubscribedTopics() throws Exception { } finally { LOGGER.info("consumer exiting..."); } - }); + }, + String.format("%s - consumer", testName.getMethodName())); thread.start(); // Check row count try { // Keep retrying if there are execution failures - Awaitility.await() - .pollDelay(IoTDBSubscriptionITConstant.AWAITILITY_POLL_DELAY_SECOND, TimeUnit.SECONDS) - .pollInterval( - IoTDBSubscriptionITConstant.AWAITILITY_POLL_INTERVAL_SECOND, TimeUnit.SECONDS) - .atMost(IoTDBSubscriptionITConstant.AWAITILITY_AT_MOST_SECOND, TimeUnit.SECONDS) - .untilAsserted( - () -> { - Assert.assertEquals(100, rowCount.get()); - Assert.assertEquals((100 + 199) * 100 / 2, timestampSum.get()); - }); + AWAIT.untilAsserted( + () -> { + Assert.assertEquals(100, rowCount.get()); + Assert.assertEquals((100 + 199) * 100 / 2, timestampSum.get()); + }); } catch (final Exception e) { e.printStackTrace(); fail(e.getMessage()); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionIdempotentIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionIdempotentIT.java index b671acd6709f4..85fff437f18e6 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionIdempotentIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionIdempotentIT.java @@ -25,9 +25,7 @@ import org.apache.iotdb.session.subscription.SubscriptionSession; import org.apache.iotdb.session.subscription.consumer.SubscriptionPullConsumer; -import org.junit.After; import org.junit.Assert; -import org.junit.Before; import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; @@ -38,20 +36,10 @@ @RunWith(IoTDBTestRunner.class) @Category({LocalStandaloneIT.class}) -public class IoTDBSubscriptionIdempotentIT { +public class IoTDBSubscriptionIdempotentIT extends AbstractSubscriptionLocalIT { private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBSubscriptionIdempotentIT.class); - @Before - public void setUp() throws Exception { - EnvFactory.getEnv().initClusterEnvironment(); - } - - @After - public void tearDown() throws Exception { - EnvFactory.getEnv().cleanClusterEnvironment(); - } - @Test public void testSubscribeOrUnsubscribeNonExistedTopicTest() { final String host = EnvFactory.getEnv().getIP(); diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionIdentifierSemanticException.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionIdentifierSemanticException.java new file mode 100644 index 0000000000000..6eb9ba6d73484 --- /dev/null +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionIdentifierSemanticException.java @@ -0,0 +1,42 @@ +/* + * 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.rpc.subscription.exception; + +import java.util.Objects; + +public class SubscriptionIdentifierSemanticException extends SubscriptionException { + + public SubscriptionIdentifierSemanticException(final String message) { + super(message); + } + + public SubscriptionIdentifierSemanticException(final String message, final Throwable cause) { + super(message, cause); + } + + @Override + public boolean equals(final Object obj) { + return obj instanceof SubscriptionIdentifierSemanticException + && Objects.equals( + getMessage(), ((SubscriptionIdentifierSemanticException) obj).getMessage()) + && Objects.equals( + getTimeStamp(), ((SubscriptionIdentifierSemanticException) obj).getTimeStamp()); + } +} diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/response/PipeSubscribeSubscribeResp.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/response/PipeSubscribeSubscribeResp.java index a5540150898d2..1d1571b0122ee 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/response/PipeSubscribeSubscribeResp.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/response/PipeSubscribeSubscribeResp.java @@ -22,10 +22,25 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.service.rpc.thrift.TPipeSubscribeResp; +import org.apache.tsfile.utils.PublicBAOS; +import org.apache.tsfile.utils.ReadWriteIOUtils; + +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.HashSet; import java.util.Objects; +import java.util.Set; public class PipeSubscribeSubscribeResp extends TPipeSubscribeResp { + private transient Set topicNames = new HashSet<>(); // subscribed topic names + + public Set getTopicNames() { + return topicNames; + } + /////////////////////////////// Thrift /////////////////////////////// /** @@ -34,11 +49,32 @@ public class PipeSubscribeSubscribeResp extends TPipeSubscribeResp { */ public static PipeSubscribeSubscribeResp toTPipeSubscribeResp(final TSStatus status) { final PipeSubscribeSubscribeResp resp = new PipeSubscribeSubscribeResp(); + resp.status = status; + resp.version = PipeSubscribeResponseVersion.VERSION_1.getVersion(); + resp.type = PipeSubscribeResponseType.ACK.getType(); + return resp; + } + + /** + * Serialize the incoming parameters into `PipeSubscribeSubscribeResp`, called by the subscription + * server. + */ + public static PipeSubscribeSubscribeResp toTPipeSubscribeResp( + final TSStatus status, final Set topicNames) throws IOException { + final PipeSubscribeSubscribeResp resp = new PipeSubscribeSubscribeResp(); resp.status = status; resp.version = PipeSubscribeResponseVersion.VERSION_1.getVersion(); resp.type = PipeSubscribeResponseType.ACK.getType(); + try (final PublicBAOS byteArrayOutputStream = new PublicBAOS(); + final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) { + ReadWriteIOUtils.writeObjectSet(topicNames, outputStream); + resp.body = + Collections.singletonList( + ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size())); + } + return resp; } @@ -47,11 +83,19 @@ public static PipeSubscribeSubscribeResp fromTPipeSubscribeResp( final TPipeSubscribeResp subscribeResp) { final PipeSubscribeSubscribeResp resp = new PipeSubscribeSubscribeResp(); + if (Objects.nonNull(subscribeResp.body)) { + for (final ByteBuffer byteBuffer : subscribeResp.body) { + if (Objects.nonNull(byteBuffer) && byteBuffer.hasRemaining()) { + resp.topicNames = ReadWriteIOUtils.readObjectSet(byteBuffer); + break; + } + } + } + resp.status = subscribeResp.status; resp.version = subscribeResp.version; resp.type = subscribeResp.type; resp.body = subscribeResp.body; - return resp; } @@ -66,7 +110,8 @@ public boolean equals(final Object obj) { return false; } final PipeSubscribeSubscribeResp that = (PipeSubscribeSubscribeResp) obj; - return Objects.equals(this.status, that.status) + return Objects.equals(this.topicNames, that.topicNames) + && Objects.equals(this.status, that.status) && this.version == that.version && this.type == that.type && Objects.equals(this.body, that.body); @@ -74,6 +119,6 @@ public boolean equals(final Object obj) { @Override public int hashCode() { - return Objects.hash(status, version, type, body); + return Objects.hash(topicNames, status, version, type, body); } } diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/response/PipeSubscribeUnsubscribeResp.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/response/PipeSubscribeUnsubscribeResp.java index a93e38013acc6..e5d8ccf69cf4b 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/response/PipeSubscribeUnsubscribeResp.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/response/PipeSubscribeUnsubscribeResp.java @@ -22,10 +22,25 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.service.rpc.thrift.TPipeSubscribeResp; +import org.apache.tsfile.utils.PublicBAOS; +import org.apache.tsfile.utils.ReadWriteIOUtils; + +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.HashSet; import java.util.Objects; +import java.util.Set; public class PipeSubscribeUnsubscribeResp extends TPipeSubscribeResp { + private transient Set topicNames = new HashSet<>(); // subscribed topic names + + public Set getTopicNames() { + return topicNames; + } + /////////////////////////////// Thrift /////////////////////////////// /** @@ -34,11 +49,32 @@ public class PipeSubscribeUnsubscribeResp extends TPipeSubscribeResp { */ public static PipeSubscribeUnsubscribeResp toTPipeSubscribeResp(final TSStatus status) { final PipeSubscribeUnsubscribeResp resp = new PipeSubscribeUnsubscribeResp(); + resp.status = status; + resp.version = PipeSubscribeResponseVersion.VERSION_1.getVersion(); + resp.type = PipeSubscribeResponseType.ACK.getType(); + return resp; + } + + /** + * Serialize the incoming parameters into `PipeSubscribeUnsubscribeResp`, called by the + * subscription server. + */ + public static PipeSubscribeUnsubscribeResp toTPipeSubscribeResp( + final TSStatus status, final Set topicNames) throws IOException { + final PipeSubscribeUnsubscribeResp resp = new PipeSubscribeUnsubscribeResp(); resp.status = status; resp.version = PipeSubscribeResponseVersion.VERSION_1.getVersion(); resp.type = PipeSubscribeResponseType.ACK.getType(); + try (final PublicBAOS byteArrayOutputStream = new PublicBAOS(); + final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) { + ReadWriteIOUtils.writeObjectSet(topicNames, outputStream); + resp.body = + Collections.singletonList( + ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size())); + } + return resp; } @@ -47,11 +83,19 @@ public static PipeSubscribeUnsubscribeResp fromTPipeSubscribeResp( final TPipeSubscribeResp unsubscribeResp) { final PipeSubscribeUnsubscribeResp resp = new PipeSubscribeUnsubscribeResp(); + if (Objects.nonNull(unsubscribeResp.body)) { + for (final ByteBuffer byteBuffer : unsubscribeResp.body) { + if (Objects.nonNull(byteBuffer) && byteBuffer.hasRemaining()) { + resp.topicNames = ReadWriteIOUtils.readObjectSet(byteBuffer); + break; + } + } + } + resp.status = unsubscribeResp.status; resp.version = unsubscribeResp.version; resp.type = unsubscribeResp.type; resp.body = unsubscribeResp.body; - return resp; } @@ -66,7 +110,8 @@ public boolean equals(final Object obj) { return false; } final PipeSubscribeUnsubscribeResp that = (PipeSubscribeUnsubscribeResp) obj; - return Objects.equals(this.status, that.status) + return Objects.equals(this.topicNames, that.topicNames) + && Objects.equals(this.status, that.status) && this.version == that.version && this.type == that.type && Objects.equals(this.body, that.body); @@ -74,6 +119,6 @@ public boolean equals(final Object obj) { @Override public int hashCode() { - return Objects.hash(status, version, type, body); + return Objects.hash(topicNames, status, version, type, body); } } diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java index 2bbd4a129a91b..a6b418f022010 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java @@ -40,6 +40,7 @@ import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponseType; import org.apache.iotdb.rpc.subscription.payload.poll.TabletsPayload; import org.apache.iotdb.session.subscription.payload.SubscriptionMessage; +import org.apache.iotdb.session.subscription.util.IdentifierUtils; import org.apache.iotdb.session.subscription.util.RandomStringGenerator; import org.apache.iotdb.session.subscription.util.SubscriptionPollTimer; import org.apache.iotdb.session.util.SessionUtils; @@ -50,8 +51,10 @@ import java.io.File; import java.io.IOException; import java.io.RandomAccessFile; +import java.net.URLEncoder; import java.nio.file.FileAlreadyExistsException; import java.nio.file.Files; +import java.nio.file.InvalidPathException; import java.nio.file.Path; import java.nio.file.Paths; import java.util.ArrayList; @@ -69,6 +72,7 @@ import java.util.concurrent.ScheduledFuture; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.LockSupport; +import java.util.stream.Collectors; abstract class SubscriptionConsumer implements AutoCloseable { @@ -92,6 +96,10 @@ abstract class SubscriptionConsumer implements AutoCloseable { private final String fileSaveDir; private final boolean fileSaveFsync; + protected volatile Set subscribedTopicNames = new HashSet<>(); + + /////////////////////////////// getter /////////////////////////////// + public String getConsumerId() { return consumerId; } @@ -100,6 +108,10 @@ public String getConsumerGroupId() { return consumerGroupId; } + public Set getSubscribedTopicNames() { + return subscribedTopicNames; + } + /////////////////////////////// ctor /////////////////////////////// protected SubscriptionConsumer(final Builder builder) { @@ -221,6 +233,17 @@ public void subscribe(final String... topicNames) throws SubscriptionException { } public void subscribe(final Set topicNames) throws SubscriptionException { + // parse topic names from external source + subscribe(topicNames, true); + } + + private void subscribe(Set topicNames, final boolean needParse) + throws SubscriptionException { + if (needParse) { + topicNames = + topicNames.stream().map(IdentifierUtils::parseIdentifier).collect(Collectors.toSet()); + } + providers.acquireReadLock(); try { subscribeWithRedirection(topicNames); @@ -238,6 +261,17 @@ public void unsubscribe(final String... topicNames) throws SubscriptionException } public void unsubscribe(final Set topicNames) throws SubscriptionException { + // parse topic names from external source + unsubscribe(topicNames, true); + } + + private void unsubscribe(Set topicNames, final boolean needParse) + throws SubscriptionException { + if (needParse) { + topicNames = + topicNames.stream().map(IdentifierUtils::parseIdentifier).collect(Collectors.toSet()); + } + providers.acquireReadLock(); try { unsubscribeWithRedirection(topicNames); @@ -284,33 +318,51 @@ private Path getFileDir(final String topicName) throws IOException { return dirPath; } - private Path getFilePath(final String topicName, String fileName) throws SubscriptionException { - Path filePath; + private Path getFilePath( + final String topicName, + final String fileName, + final boolean allowFileAlreadyExistsException, + final boolean allowInvalidPathException) + throws SubscriptionException { try { - filePath = getFileDir(topicName).resolve(fileName); + final Path filePath = getFileDir(topicName).resolve(fileName); Files.createFile(filePath); + return filePath; } catch (final FileAlreadyExistsException fileAlreadyExistsException) { - fileName += "." + RandomStringGenerator.generate(16); - try { - filePath = getFileDir(topicName).resolve(fileName); - Files.createFile(filePath); - } catch (final IOException e) { - throw new SubscriptionRuntimeNonCriticalException(e.getMessage(), e); + if (allowFileAlreadyExistsException) { + return getFilePath( + topicName, fileName + "." + RandomStringGenerator.generate(16), false, true); } + throw new SubscriptionRuntimeNonCriticalException( + fileAlreadyExistsException.getMessage(), fileAlreadyExistsException); + } catch (final InvalidPathException invalidPathException) { + if (allowInvalidPathException) { + return getFilePath(URLEncoder.encode(topicName), fileName, true, false); + } + throw new SubscriptionRuntimeNonCriticalException( + invalidPathException.getMessage(), invalidPathException); } catch (final IOException e) { throw new SubscriptionRuntimeNonCriticalException(e.getMessage(), e); } - return filePath; } /////////////////////////////// poll /////////////////////////////// - protected List poll(final Set topicNames, final long timeoutMs) + protected List poll( + /* @NotNull */ final Set topicNames, final long timeoutMs) throws SubscriptionException { final List messages = new ArrayList<>(); final SubscriptionPollTimer timer = new SubscriptionPollTimer(System.currentTimeMillis(), timeoutMs); + // check topic names + topicNames.stream() + .filter(topicName -> !subscribedTopicNames.contains(topicName)) + .forEach( + topicName -> + LOGGER.warn( + "SubscriptionConsumer {} does not subscribe to topic {}", this, topicName)); + do { try { // poll tablets or file @@ -336,14 +388,7 @@ protected List poll(final Set topicNames, final lon case ERROR: final ErrorPayload payload = (ErrorPayload) pollResponse.getPayload(); final String errorMessage = payload.getErrorMessage(); - final boolean critical = payload.isCritical(); - LOGGER.warn( - "Error occurred when SubscriptionConsumer {} polling topics {}: {}, critical: {}", - this, - topicNames, - errorMessage, - critical); - if (critical) { + if (payload.isCritical()) { throw new SubscriptionRuntimeCriticalException(errorMessage); } else { throw new SubscriptionRuntimeNonCriticalException(errorMessage); @@ -398,7 +443,7 @@ private Optional pollFile( final SubscriptionCommitContext commitContext, final String fileName) throws SubscriptionException { final String topicName = commitContext.getTopicName(); - final Path filePath = getFilePath(topicName, fileName); + final Path filePath = getFilePath(topicName, fileName, true, true); final File file = filePath.toFile(); try (final RandomAccessFile fileWriter = new RandomAccessFile(file, "rw")) { return Optional.of(pollFileInternal(commitContext, file, fileWriter)); @@ -576,6 +621,9 @@ private List pollInternal(final Set topicNames try { final SubscriptionProvider provider = providers.getNextAvailableProvider(); if (Objects.isNull(provider) || !provider.isAvailable()) { + if (isClosed()) { + return Collections.emptyList(); + } throw new SubscriptionConnectionException( String.format( "Cluster has no available subscription providers when %s poll topic %s", @@ -601,6 +649,9 @@ private List pollFileInternal( try { final SubscriptionProvider provider = providers.getProvider(dataNodeId); if (Objects.isNull(provider) || !provider.isAvailable()) { + if (isClosed()) { + return Collections.emptyList(); + } throw new SubscriptionConnectionException( String.format( "something unexpected happened when %s poll file from subscription provider with data node id %s, the subscription provider may be unavailable or not existed", @@ -660,6 +711,9 @@ private void commitInternal( try { final SubscriptionProvider provider = providers.getProvider(dataNodeId); if (Objects.isNull(provider) || !provider.isAvailable()) { + if (isClosed()) { + return; + } throw new SubscriptionConnectionException( String.format( "something unexpected happened when %s commit (nack: %s) messages to subscription provider with data node id %s, the subscription provider may be unavailable or not existed", @@ -775,7 +829,7 @@ private void subscribeWithRedirection(final Set topicNames) throws Subsc } for (final SubscriptionProvider provider : providers) { try { - provider.subscribe(topicNames); + subscribedTopicNames = provider.subscribe(topicNames); return; } catch (final Exception e) { LOGGER.warn( @@ -805,7 +859,7 @@ private void unsubscribeWithRedirection(final Set topicNames) } for (final SubscriptionProvider provider : providers) { try { - provider.unsubscribe(topicNames); + subscribedTopicNames = provider.unsubscribe(topicNames); return; } catch (final Exception e) { LOGGER.warn( @@ -897,12 +951,12 @@ public Builder password(final String password) { } public Builder consumerId(final String consumerId) { - this.consumerId = consumerId; + this.consumerId = IdentifierUtils.parseIdentifier(consumerId); return this; } public Builder consumerGroupId(final String consumerGroupId) { - this.consumerGroupId = consumerGroupId; + this.consumerGroupId = IdentifierUtils.parseIdentifier(consumerGroupId); return this; } diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionProvider.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionProvider.java index 70589472f8287..9bce0d6774faf 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionProvider.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionProvider.java @@ -40,6 +40,8 @@ import org.apache.iotdb.rpc.subscription.payload.request.PipeSubscribeUnsubscribeReq; import org.apache.iotdb.rpc.subscription.payload.response.PipeSubscribeHandshakeResp; import org.apache.iotdb.rpc.subscription.payload.response.PipeSubscribePollResp; +import org.apache.iotdb.rpc.subscription.payload.response.PipeSubscribeSubscribeResp; +import org.apache.iotdb.rpc.subscription.payload.response.PipeSubscribeUnsubscribeResp; import org.apache.iotdb.service.rpc.thrift.TPipeSubscribeResp; import org.apache.iotdb.session.subscription.SubscriptionSession; import org.apache.iotdb.session.subscription.SubscriptionSessionConnection; @@ -215,7 +217,7 @@ void heartbeat() throws SubscriptionException { verifyPipeSubscribeSuccess(resp.status); } - void subscribe(final Set topicNames) throws SubscriptionException { + Set subscribe(final Set topicNames) throws SubscriptionException { final PipeSubscribeSubscribeReq req; try { req = PipeSubscribeSubscribeReq.toTPipeSubscribeReq(topicNames); @@ -241,9 +243,12 @@ void subscribe(final Set topicNames) throws SubscriptionException { throw new SubscriptionConnectionException(e.getMessage(), e); } verifyPipeSubscribeSuccess(resp.status); + final PipeSubscribeSubscribeResp subscribeResp = + PipeSubscribeSubscribeResp.fromTPipeSubscribeResp(resp); + return subscribeResp.getTopicNames(); } - void unsubscribe(final Set topicNames) throws SubscriptionException { + Set unsubscribe(final Set topicNames) throws SubscriptionException { final PipeSubscribeUnsubscribeReq req; try { req = PipeSubscribeUnsubscribeReq.toTPipeSubscribeReq(topicNames); @@ -269,6 +274,9 @@ void unsubscribe(final Set topicNames) throws SubscriptionException { throw new SubscriptionConnectionException(e.getMessage(), e); } verifyPipeSubscribeSuccess(resp.status); + final PipeSubscribeUnsubscribeResp unsubscribeResp = + PipeSubscribeUnsubscribeResp.fromTPipeSubscribeResp(resp); + return unsubscribeResp.getTopicNames(); } List poll(final SubscriptionPollRequest pollMessage) diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionPullConsumer.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionPullConsumer.java index 2306e4d05c1e2..38dfc530ee352 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionPullConsumer.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionPullConsumer.java @@ -22,6 +22,7 @@ import org.apache.iotdb.rpc.subscription.config.ConsumerConstant; import org.apache.iotdb.rpc.subscription.exception.SubscriptionException; import org.apache.iotdb.session.subscription.payload.SubscriptionMessage; +import org.apache.iotdb.session.subscription.util.IdentifierUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -39,6 +40,7 @@ import java.util.concurrent.ConcurrentSkipListSet; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; /** * The {@link SubscriptionPullConsumer} corresponds to the pull consumption mode in the message @@ -150,7 +152,12 @@ public List poll(final Set topicNames, final Durati @Override public List poll(final Set topicNames, final long timeoutMs) throws SubscriptionException { - final List messages = super.poll(topicNames, timeoutMs); + // parse topic names from external source + final Set parsedTopicNames = + topicNames.stream().map(IdentifierUtils::parseIdentifier).collect(Collectors.toSet()); + + // poll messages + final List messages = super.poll(parsedTopicNames, timeoutMs); // add to uncommitted messages if (autoCommit) { diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionPushConsumer.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionPushConsumer.java index 7ff1dd39b15bd..7323473552caf 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionPushConsumer.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionPushConsumer.java @@ -27,7 +27,6 @@ import org.slf4j.LoggerFactory; import java.util.ArrayList; -import java.util.Collections; import java.util.List; import java.util.Objects; import java.util.Properties; @@ -162,9 +161,12 @@ public void run() { return; } + if (subscribedTopicNames.isEmpty()) { + return; + } + try { - // Poll all subscribed topics by passing an empty set - final List messages = poll(Collections.emptySet(), autoPollTimeoutMs); + final List messages = poll(subscribedTopicNames, autoPollTimeoutMs); if (ackStrategy.equals(AckStrategy.BEFORE_CONSUME)) { ack(messages); diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionFileHandler.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionFileHandler.java index 60c269b86370c..0ec121f96992a 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionFileHandler.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionFileHandler.java @@ -39,14 +39,14 @@ public SubscriptionFileHandler(final String absolutePath) { /** * @return a new File instance of the corresponding file */ - public File getFile() { + public synchronized File getFile() { return new File(absolutePath); } /** * @return a new Path instance of the corresponding file */ - public Path getPath() { + public synchronized Path getPath() { return Paths.get(absolutePath); } @@ -54,7 +54,7 @@ public Path getPath() { * @return the path to the source file * @throws IOException if an I/O error occurs */ - public Path deleteFile() throws IOException { + public synchronized Path deleteFile() throws IOException { final Path sourcePath = getPath(); Files.delete(sourcePath); return sourcePath; @@ -65,7 +65,7 @@ public Path deleteFile() throws IOException { * @return the path to the target file * @throws IOException if an I/O error occurs */ - public Path moveFile(final String target) throws IOException { + public synchronized Path moveFile(final String target) throws IOException { return this.moveFile(Paths.get(target)); } @@ -74,7 +74,10 @@ public Path moveFile(final String target) throws IOException { * @return the path to the target file * @throws IOException if an I/O error occurs */ - public Path moveFile(final Path target) throws IOException { + public synchronized Path moveFile(final Path target) throws IOException { + if (!Files.exists(target.getParent())) { + Files.createDirectories(target.getParent()); + } return Files.move(getPath(), target, StandardCopyOption.REPLACE_EXISTING); } @@ -83,7 +86,7 @@ public Path moveFile(final Path target) throws IOException { * @return the path to the target file * @throws IOException if an I/O error occurs */ - public Path copyFile(final String target) throws IOException { + public synchronized Path copyFile(final String target) throws IOException { return this.copyFile(Paths.get(target)); } @@ -92,7 +95,10 @@ public Path copyFile(final String target) throws IOException { * @return the path to the target file * @throws IOException if an I/O error occurs */ - public Path copyFile(final Path target) throws IOException { + public synchronized Path copyFile(final Path target) throws IOException { + if (!Files.exists(target.getParent())) { + Files.createDirectories(target.getParent()); + } return Files.copy( getPath(), target, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.COPY_ATTRIBUTES); } diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/util/IdentifierUtils.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/util/IdentifierUtils.java new file mode 100644 index 0000000000000..9f6d09ef44e0f --- /dev/null +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/util/IdentifierUtils.java @@ -0,0 +1,51 @@ +/* + * 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.session.subscription.util; + +import org.apache.iotdb.rpc.subscription.exception.SubscriptionIdentifierSemanticException; + +import org.apache.tsfile.common.constant.TsFileConstant; +import org.apache.tsfile.read.common.parser.PathVisitor; + +public class IdentifierUtils { + + /** + * refer org.apache.iotdb.db.queryengine.plan.parser.ASTVisitor#parseIdentifier(java.lang.String) + */ + public static String parseIdentifier(final String src) { + if (src.startsWith(TsFileConstant.BACK_QUOTE_STRING) + && src.endsWith(TsFileConstant.BACK_QUOTE_STRING)) { + return src.substring(1, src.length() - 1) + .replace(TsFileConstant.DOUBLE_BACK_QUOTE_STRING, TsFileConstant.BACK_QUOTE_STRING); + } + checkIdentifier(src); + return src; + } + + private static void checkIdentifier(final String src) { + if (!TsFileConstant.IDENTIFIER_PATTERN.matcher(src).matches() + || PathVisitor.isRealNumber(src)) { + throw new SubscriptionIdentifierSemanticException( + String.format( + "%s is illegal, identifier not enclosed with backticks can only consist of digits, characters and underscore.", + src)); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionAgentLauncher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionAgentLauncher.java index 32dd078dd7805..d83996cda470b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionAgentLauncher.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionAgentLauncher.java @@ -87,7 +87,7 @@ public static synchronized void launchSubscriptionTopicAgent() throws StartupExc } return; - } catch (SubscriptionException | ClientManagerException | TException e) { + } catch (final SubscriptionException | ClientManagerException | TException e) { retry++; LOGGER.warn( "Failed to get topic meta from config node for {} times, will retry at most {} times.", @@ -97,7 +97,7 @@ public static synchronized void launchSubscriptionTopicAgent() throws StartupExc try { Thread.sleep( retry * SubscriptionConfig.getInstance().getSubscriptionLaunchRetryIntervalMs()); - } catch (InterruptedException interruptedException) { + } catch (final InterruptedException interruptedException) { LOGGER.info( "Interrupted while sleeping, will retry to get topic meta from config node.", interruptedException); @@ -146,7 +146,7 @@ public static synchronized void launchSubscriptionConsumerAgent() throws Startup } return; - } catch (SubscriptionException | ClientManagerException | TException e) { + } catch (final SubscriptionException | ClientManagerException | TException e) { retry++; LOGGER.warn( "Failed to get consumer group meta from config node for {} times, will retry at most {} times.", @@ -156,7 +156,7 @@ public static synchronized void launchSubscriptionConsumerAgent() throws Startup try { Thread.sleep( retry * SubscriptionConfig.getInstance().getSubscriptionLaunchRetryIntervalMs()); - } catch (InterruptedException interruptedException) { + } catch (final InterruptedException interruptedException) { LOGGER.info( "Interrupted while sleeping, will retry to get consumer group meta from config node.", interruptedException); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java index aff4893c61a03..21064ac34c89a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java @@ -141,23 +141,22 @@ public void bindPrefetchingQueue(final SubscriptionConnectorSubtask subtask) { broker.bindPrefetchingQueue(subtask.getTopicName(), subtask.getInputPendingQueue()); } - public void unbindPrefetchingQueue(final SubscriptionConnectorSubtask subtask) { - final String consumerGroupId = subtask.getConsumerGroupId(); + public void unbindPrefetchingQueue( + final String consumerGroupId, final String topicName, final boolean doRemove) { final SubscriptionBroker broker = consumerGroupIdToSubscriptionBroker.get(consumerGroupId); if (Objects.isNull(broker)) { LOGGER.warn("Subscription: consumer group [{}] does not exist", consumerGroupId); return; } - broker.unbindPrefetchingQueue(subtask.getTopicName()); + broker.unbindPrefetchingQueue(topicName, doRemove); } - public void executePrefetch(final SubscriptionConnectorSubtask subtask) { - final String consumerGroupId = subtask.getConsumerGroupId(); + public void executePrefetch(final String consumerGroupId, final String topicName) { final SubscriptionBroker broker = consumerGroupIdToSubscriptionBroker.get(consumerGroupId); if (Objects.isNull(broker)) { LOGGER.warn("Subscription: consumer group [{}] does not exist", consumerGroupId); return; } - broker.executePrefetch(subtask.getTopicName()); + broker.executePrefetch(topicName); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java index d6572923635c4..5284d295863d6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java @@ -62,7 +62,7 @@ protected void releaseWriteLock() { ////////////////////////// ConsumerGroupMeta Management Entry ////////////////////////// public TPushConsumerGroupMetaRespExceptionMessage handleSingleConsumerGroupMetaChanges( - ConsumerGroupMeta consumerGroupMetaFromCoordinator) { + final ConsumerGroupMeta consumerGroupMetaFromCoordinator) { acquireWriteLock(); try { if (consumerGroupMetaFromCoordinator.isEmpty()) { @@ -71,7 +71,7 @@ public TPushConsumerGroupMetaRespExceptionMessage handleSingleConsumerGroupMetaC handleSingleConsumerGroupMetaChangesInternal(consumerGroupMetaFromCoordinator); } return null; - } catch (Exception e) { + } catch (final Exception e) { final String consumerGroupId = consumerGroupMetaFromCoordinator.getConsumerGroupId(); final String exceptionMessage = String.format( @@ -121,14 +121,15 @@ private void handleSingleConsumerGroupMetaChangesInternal( } public TPushConsumerGroupMetaRespExceptionMessage handleConsumerGroupMetaChanges( - List consumerGroupMetasFromCoordinator) { + final List consumerGroupMetasFromCoordinator) { acquireWriteLock(); try { - for (ConsumerGroupMeta consumerGroupMetaFromCoordinator : consumerGroupMetasFromCoordinator) { + for (final ConsumerGroupMeta consumerGroupMetaFromCoordinator : + consumerGroupMetasFromCoordinator) { try { handleSingleConsumerGroupMetaChangesInternal(consumerGroupMetaFromCoordinator); return null; - } catch (Exception e) { + } catch (final Exception e) { final String consumerGroupId = consumerGroupMetaFromCoordinator.getConsumerGroupId(); final String exceptionMessage = String.format( @@ -146,12 +147,12 @@ public TPushConsumerGroupMetaRespExceptionMessage handleConsumerGroupMetaChanges } public TPushConsumerGroupMetaRespExceptionMessage handleDropConsumerGroup( - String consumerGroupId) { + final String consumerGroupId) { acquireWriteLock(); try { handleDropConsumerGroupInternal(consumerGroupId); return null; - } catch (Exception e) { + } catch (final Exception e) { final String exceptionMessage = String.format( "Subscription: Failed to drop consumer group %s, because %s", consumerGroupId, e); @@ -163,7 +164,7 @@ public TPushConsumerGroupMetaRespExceptionMessage handleDropConsumerGroup( } } - private void handleDropConsumerGroupInternal(String consumerGroupId) { + private void handleDropConsumerGroupInternal(final String consumerGroupId) { if (SubscriptionAgent.broker().isBrokerExist(consumerGroupId)) { if (!SubscriptionAgent.broker().dropBroker(consumerGroupId)) { final String exceptionMessage = @@ -180,7 +181,7 @@ private void handleDropConsumerGroupInternal(String consumerGroupId) { consumerGroupMetaKeeper.removeConsumerGroupMeta(consumerGroupId); } - public boolean isConsumerExisted(String consumerGroupId, String consumerId) { + public boolean isConsumerExisted(final String consumerGroupId, final String consumerId) { acquireReadLock(); try { final ConsumerGroupMeta consumerGroupMeta = @@ -191,7 +192,8 @@ public boolean isConsumerExisted(String consumerGroupId, String consumerId) { } } - public Set getTopicsSubscribedByConsumer(String consumerGroupId, String consumerId) { + public Set getTopicsSubscribedByConsumer( + final String consumerGroupId, final String consumerId) { acquireReadLock(); try { return consumerGroupMetaKeeper.getTopicsSubscribedByConsumer(consumerGroupId, consumerId); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java index 1bb0735b019a3..71f319374cd53 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java @@ -51,7 +51,7 @@ public class SubscriptionReceiverAgent { PipeSubscribeRequestVersion.VERSION_1.getVersion(), SubscriptionReceiverV1::new); } - public TPipeSubscribeResp handle(TPipeSubscribeReq req) { + public TPipeSubscribeResp handle(final TPipeSubscribeReq req) { final byte reqVersion = req.getVersion(); if (RECEIVER_CONSTRUCTORS.containsKey(reqVersion)) { return getReceiver(reqVersion).handle(req); @@ -69,7 +69,7 @@ public TPipeSubscribeResp handle(TPipeSubscribeReq req) { } } - private SubscriptionReceiver getReceiver(byte reqVersion) { + private SubscriptionReceiver getReceiver(final byte reqVersion) { if (receiverThreadLocal.get() == null) { return setAndGetReceiver(reqVersion); } @@ -88,7 +88,7 @@ private SubscriptionReceiver getReceiver(byte reqVersion) { return receiverThreadLocal.get(); } - private SubscriptionReceiver setAndGetReceiver(byte reqVersion) { + private SubscriptionReceiver setAndGetReceiver(final byte reqVersion) { if (RECEIVER_CONSTRUCTORS.containsKey(reqVersion)) { receiverThreadLocal.set(RECEIVER_CONSTRUCTORS.get(reqVersion).get()); } else { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java index fb27e8825b8e8..45145bd2b67e0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java @@ -60,12 +60,12 @@ protected void releaseWriteLock() { ////////////////////////// Topic Management Entry ////////////////////////// public TPushTopicMetaRespExceptionMessage handleSingleTopicMetaChanges( - TopicMeta topicMetaFromCoordinator) { + final TopicMeta topicMetaFromCoordinator) { acquireWriteLock(); try { handleSingleTopicMetaChangesInternal(topicMetaFromCoordinator); return null; - } catch (Exception e) { + } catch (final Exception e) { final String topicName = topicMetaFromCoordinator.getTopicName(); final String exceptionMessage = String.format( @@ -86,13 +86,13 @@ private void handleSingleTopicMetaChangesInternal(final TopicMeta metaFromCoordi } public TPushTopicMetaRespExceptionMessage handleTopicMetaChanges( - List topicMetasFromCoordinator) { + final List topicMetasFromCoordinator) { acquireWriteLock(); try { - for (TopicMeta topicMetaFromCoordinator : topicMetasFromCoordinator) { + for (final TopicMeta topicMetaFromCoordinator : topicMetasFromCoordinator) { try { handleSingleTopicMetaChangesInternal(topicMetaFromCoordinator); - } catch (Exception e) { + } catch (final Exception e) { final String topicName = topicMetaFromCoordinator.getTopicName(); final String exceptionMessage = String.format( @@ -109,12 +109,12 @@ public TPushTopicMetaRespExceptionMessage handleTopicMetaChanges( } } - public TPushTopicMetaRespExceptionMessage handleDropTopic(String topicName) { + public TPushTopicMetaRespExceptionMessage handleDropTopic(final String topicName) { acquireWriteLock(); try { handleDropTopicInternal(topicName); return null; - } catch (Exception e) { + } catch (final Exception e) { final String exceptionMessage = String.format("Subscription: Failed to drop topic %s, because %s", topicName, e); LOGGER.warn(exceptionMessage); @@ -125,11 +125,11 @@ public TPushTopicMetaRespExceptionMessage handleDropTopic(String topicName) { } } - private void handleDropTopicInternal(String topicName) { + private void handleDropTopicInternal(final String topicName) { topicMetaKeeper.removeTopicMeta(topicName); } - public boolean isTopicExisted(String topicName) { + public boolean isTopicExisted(final String topicName) { acquireReadLock(); try { return topicMetaKeeper.containsTopicMeta(topicName); @@ -138,7 +138,7 @@ public boolean isTopicExisted(String topicName) { } } - public String getTopicFormat(String topicName) { + public String getTopicFormat(final String topicName) { acquireReadLock(); try { return topicMetaKeeper diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java index 90785c1602e62..d26ea8b445817 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java @@ -65,6 +65,10 @@ public List poll(final String consumerId, final Set t final String topicName = entry.getKey(); final SubscriptionPrefetchingQueue prefetchingQueue = entry.getValue(); if (topicNames.contains(topicName)) { + if (prefetchingQueue.isClosed()) { + LOGGER.warn("Subscription: prefetching queue bound to topic [{}] is closed", topicName); + continue; + } final SubscriptionEvent event = prefetchingQueue.poll(consumerId); if (Objects.nonNull(event)) { events.add(event); @@ -95,6 +99,10 @@ public List pollTsFile( LOGGER.warn(errorMessage); throw new SubscriptionException(errorMessage); } + if (prefetchingQueue.isClosed()) { + LOGGER.warn("Subscription: prefetching queue bound to topic [{}] is closed", topicName); + return Collections.emptyList(); + } final SubscriptionEvent event = ((SubscriptionPrefetchingTsFileQueue) prefetchingQueue) .pollTsFile(consumerId, fileName, writingOffset); @@ -117,6 +125,10 @@ public List commit( "Subscription: prefetching queue bound to topic [{}] does not exist", topicName); continue; } + if (prefetchingQueue.isClosed()) { + LOGGER.warn("Subscription: prefetching queue bound to topic [{}] is closed", topicName); + continue; + } if (!nack) { if (prefetchingQueue.ack(commitContext)) { successfulCommitContexts.add(commitContext); @@ -155,18 +167,25 @@ public void bindPrefetchingQueue( } } - public void unbindPrefetchingQueue(final String topicName) { + public void unbindPrefetchingQueue(final String topicName, final boolean doRemove) { final SubscriptionPrefetchingQueue prefetchingQueue = topicNameToPrefetchingQueue.get(topicName); if (Objects.isNull(prefetchingQueue)) { LOGGER.warn("Subscription: prefetching queue bound to topic [{}] does not exist", topicName); return; } - // clean up uncommitted events + + // mark prefetching queue closed first + prefetchingQueue.markClosed(); + + // clean up events in prefetching queue, this operation is idempotent prefetchingQueue.cleanup(); - topicNameToPrefetchingQueue.remove(topicName); - SubscriptionPrefetchingQueueMetrics.getInstance() - .deregister(prefetchingQueue.getPrefetchingQueueId()); + + if (doRemove) { + topicNameToPrefetchingQueue.remove(topicName); + SubscriptionPrefetchingQueueMetrics.getInstance() + .deregister(prefetchingQueue.getPrefetchingQueueId()); + } } public void executePrefetch(final String topicName) { @@ -176,6 +195,10 @@ public void executePrefetch(final String topicName) { LOGGER.warn("Subscription: prefetching queue bound to topic [{}] does not exist", topicName); return; } + if (prefetchingQueue.isClosed()) { + LOGGER.warn("Subscription: prefetching queue bound to topic [{}] is closed", topicName); + return; + } prefetchingQueue.executePrefetch(); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java index 43aef19568148..9dec709230f5d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java @@ -46,6 +46,9 @@ public abstract class SubscriptionPrefetchingQueue { protected final Map uncommittedEvents; private final AtomicLong subscriptionCommitIdGenerator = new AtomicLong(0); + private volatile boolean isCompleted = false; + private volatile boolean isClosed = false; + public SubscriptionPrefetchingQueue( final String brokerId, final String topicName, @@ -61,11 +64,16 @@ public SubscriptionPrefetchingQueue( public abstract void executePrefetch(); - /** clean up uncommitted events */ public void cleanup() { + // clean up uncommitted events for (final SubscriptionEvent event : uncommittedEvents.values()) { + event.clearReferenceCount(); SubscriptionEventBinaryCache.getInstance().resetByteBuffer(event, true); } + uncommittedEvents.clear(); + + // no need to clean up events in inputPendingQueue, see + // org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask.close } /////////////////////////////// commit /////////////////////////////// @@ -150,4 +158,22 @@ public long getUncommittedEventCount() { public long getCurrentCommitId() { return subscriptionCommitIdGenerator.get(); } + + /////////////////////////////// termination /////////////////////////////// + + public boolean isClosed() { + return isClosed; + } + + public void markClosed() { + isClosed = true; + } + + public boolean isCompleted() { + return isCompleted; + } + + public void markCompleted() { + isCompleted = true; + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingTabletsQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingTabletsQueue.java index 195194ef3c899..14d3315e1b79d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingTabletsQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingTabletsQueue.java @@ -63,6 +63,15 @@ public SubscriptionPrefetchingTabletsQueue( this.prefetchingQueue = new LinkedBlockingQueue<>(); } + @Override + public void cleanup() { + super.cleanup(); + + // no need to clean up events in prefetchingQueue, since all events in prefetchingQueue are also + // in uncommittedEvents + prefetchingQueue.clear(); + } + @Override public SubscriptionEvent poll(final String consumerId) { if (prefetchingQueue.isEmpty()) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingTsFileQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingTsFileQueue.java index fc5a1a8a81eeb..3f2dac42db6c0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingTsFileQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingTsFileQueue.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.pipe.task.connection.UnboundedBlockingPendingQueue; import org.apache.iotdb.db.pipe.event.UserDefinedEnrichedEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; +import org.apache.iotdb.db.subscription.event.SubscriptionEventBinaryCache; import org.apache.iotdb.db.subscription.event.SubscriptionTsFileEvent; import org.apache.iotdb.pipe.api.event.Event; import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent; @@ -63,6 +64,21 @@ public SubscriptionPrefetchingTsFileQueue( this.consumerIdToCurrentEventMap = new ConcurrentHashMap<>(); } + @Override + public void cleanup() { + super.cleanup(); + + // clean up events in consumerIdToCurrentEventMap + consumerIdToCurrentEventMap + .values() + .forEach( + (event) -> { + event.clearReferenceCount(); + SubscriptionEventBinaryCache.getInstance().resetByteBuffer(event, true); + }); + consumerIdToCurrentEventMap.clear(); + } + @Override public SubscriptionTsFileEvent poll(final String consumerId) { if (hasUnPollableOnTheFlySubscriptionTsFileEvent(consumerId)) { @@ -78,6 +94,14 @@ public SubscriptionTsFileEvent poll(final String consumerId) { Event event; while (Objects.nonNull( event = UserDefinedEnrichedEvent.maybeOf(inputPendingQueue.waitedPoll()))) { + if (!(event instanceof EnrichedEvent)) { + LOGGER.warn( + "Subscription: SubscriptionPrefetchingTsFileQueue {} only support poll EnrichedEvent. Ignore {}.", + this, + event); + continue; + } + if (event instanceof TabletInsertionEvent) { final String errorMessage = String.format( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java index c327b7e058d73..0f7feb08e95e4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java @@ -76,6 +76,12 @@ public void decreaseReferenceCount() { } } + public void clearReferenceCount() { + for (final EnrichedEvent enrichedEvent : enrichedEvents) { + enrichedEvent.clearReferenceCount(this.getClass().getName()); + } + } + //////////////////////////// pollable //////////////////////////// public void recordLastPolledTimestamp() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java index 8339c3d3c8812..9876339101aef 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java @@ -31,7 +31,6 @@ import org.apache.iotdb.db.protocol.client.ConfigNodeClient; import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager; import org.apache.iotdb.db.protocol.client.ConfigNodeInfo; -import org.apache.iotdb.db.queryengine.plan.parser.ASTVisitor; import org.apache.iotdb.db.subscription.agent.SubscriptionAgent; import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingQueue; import org.apache.iotdb.db.subscription.event.SubscriptionEvent; @@ -72,6 +71,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.IOException; import java.nio.ByteBuffer; import java.util.Collections; import java.util.List; @@ -247,7 +247,7 @@ private TPipeSubscribeResp handlePipeSubscribeHeartbeatInternal( private TPipeSubscribeResp handlePipeSubscribeSubscribe(final PipeSubscribeSubscribeReq req) { try { return handlePipeSubscribeSubscribeInternal(req); - } catch (final SubscriptionException e) { + } catch (final SubscriptionException | IOException e) { final String exceptionMessage = String.format( "Subscription: something unexpected happened when subscribing: %s, req: %s", e, req); @@ -258,7 +258,7 @@ private TPipeSubscribeResp handlePipeSubscribeSubscribe(final PipeSubscribeSubsc } private TPipeSubscribeResp handlePipeSubscribeSubscribeInternal( - final PipeSubscribeSubscribeReq req) { + final PipeSubscribeSubscribeReq req) throws IOException { // check consumer config thread local final ConsumerConfig consumerConfig = consumerConfigThreadLocal.get(); if (Objects.isNull(consumerConfig)) { @@ -268,18 +268,21 @@ private TPipeSubscribeResp handlePipeSubscribeSubscribeInternal( } // subscribe topics - Set topicNames = req.getTopicNames(); - topicNames = topicNames.stream().map(ASTVisitor::parseIdentifier).collect(Collectors.toSet()); + final Set topicNames = req.getTopicNames(); subscribe(consumerConfig, topicNames); LOGGER.info("Subscription: consumer {} subscribe {} successfully", consumerConfig, topicNames); - return PipeSubscribeSubscribeResp.toTPipeSubscribeResp(RpcUtils.SUCCESS_STATUS); + return PipeSubscribeSubscribeResp.toTPipeSubscribeResp( + RpcUtils.SUCCESS_STATUS, + SubscriptionAgent.consumer() + .getTopicsSubscribedByConsumer( + consumerConfig.getConsumerGroupId(), consumerConfig.getConsumerId())); } private TPipeSubscribeResp handlePipeSubscribeUnsubscribe(final PipeSubscribeUnsubscribeReq req) { try { return handlePipeSubscribeUnsubscribeInternal(req); - } catch (final SubscriptionException e) { + } catch (final SubscriptionException | IOException e) { final String exceptionMessage = String.format( "Subscription: something unexpected happened when unsubscribing: %s, req: %s", @@ -291,7 +294,7 @@ private TPipeSubscribeResp handlePipeSubscribeUnsubscribe(final PipeSubscribeUns } private TPipeSubscribeResp handlePipeSubscribeUnsubscribeInternal( - final PipeSubscribeUnsubscribeReq req) { + final PipeSubscribeUnsubscribeReq req) throws IOException { // check consumer config thread local final ConsumerConfig consumerConfig = consumerConfigThreadLocal.get(); if (Objects.isNull(consumerConfig)) { @@ -302,13 +305,16 @@ private TPipeSubscribeResp handlePipeSubscribeUnsubscribeInternal( } // unsubscribe topics - Set topicNames = req.getTopicNames(); - topicNames = topicNames.stream().map(ASTVisitor::parseIdentifier).collect(Collectors.toSet()); + final Set topicNames = req.getTopicNames(); unsubscribe(consumerConfig, topicNames); LOGGER.info( "Subscription: consumer {} unsubscribe {} successfully", consumerConfig, topicNames); - return PipeSubscribeUnsubscribeResp.toTPipeSubscribeResp(RpcUtils.SUCCESS_STATUS); + return PipeSubscribeUnsubscribeResp.toTPipeSubscribeResp( + RpcUtils.SUCCESS_STATUS, + SubscriptionAgent.consumer() + .getTopicsSubscribedByConsumer( + consumerConfig.getConsumerGroupId(), consumerConfig.getConsumerId())); } private TPipeSubscribeResp handlePipeSubscribePoll(final PipeSubscribePollReq req) { @@ -403,15 +409,11 @@ private List handlePipeSubscribePollInternal( SubscriptionAgent.consumer() .getTopicsSubscribedByConsumer( consumerConfig.getConsumerGroupId(), consumerConfig.getConsumerId()); - final Set topicNames; - if (messagePayload.getTopicNames().isEmpty()) { + Set topicNames = messagePayload.getTopicNames(); + if (topicNames.isEmpty()) { // poll all subscribed topics topicNames = subscribedTopicNames; } else { - topicNames = - messagePayload.getTopicNames().stream() - .map(ASTVisitor::parseIdentifier) - .collect(Collectors.toSet()); // filter unsubscribed topics topicNames.removeIf((topicName) -> !subscribedTopicNames.contains(topicName)); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionConnectorSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionConnectorSubtask.java index bb4838b903e7e..b429b8399420e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionConnectorSubtask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionConnectorSubtask.java @@ -31,14 +31,14 @@ public class SubscriptionConnectorSubtask extends PipeConnectorSubtask { private final String consumerGroupId; public SubscriptionConnectorSubtask( - String taskID, - long creationTime, - String attributeSortedString, - int connectorIndex, - UnboundedBlockingPendingQueue inputPendingQueue, - PipeConnector outputPipeConnector, - String topicName, - String consumerGroupId) { + final String taskID, + final long creationTime, + final String attributeSortedString, + final int connectorIndex, + final UnboundedBlockingPendingQueue inputPendingQueue, + final PipeConnector outputPipeConnector, + final String topicName, + final String consumerGroupId) { super( taskID, creationTime, @@ -56,7 +56,7 @@ protected boolean executeOnce() { return false; } - SubscriptionAgent.broker().executePrefetch(this); + SubscriptionAgent.broker().executePrefetch(consumerGroupId, topicName); // always return true return true; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionConnectorSubtaskLifeCycle.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionConnectorSubtaskLifeCycle.java index 14633dfafefe9..43827ab85c42e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionConnectorSubtaskLifeCycle.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionConnectorSubtaskLifeCycle.java @@ -38,9 +38,9 @@ public class SubscriptionConnectorSubtaskLifeCycle extends PipeConnectorSubtaskL private int registeredTaskCount; public SubscriptionConnectorSubtaskLifeCycle( - PipeConnectorSubtaskExecutor executor, // SubscriptionSubtaskExecutor - PipeConnectorSubtask subtask, // SubscriptionConnectorSubtask - UnboundedBlockingPendingQueue pendingQueue) { + final PipeConnectorSubtaskExecutor executor, // SubscriptionSubtaskExecutor + final PipeConnectorSubtask subtask, // SubscriptionConnectorSubtask + final UnboundedBlockingPendingQueue pendingQueue) { super(executor, subtask, pendingQueue); runningTaskCount = 0; @@ -54,6 +54,7 @@ public synchronized void register() { } if (registeredTaskCount == 0) { + // bind prefetching queue SubscriptionAgent.broker().bindPrefetchingQueue((SubscriptionConnectorSubtask) subtask); executor.register(subtask); runningTaskCount = 0; @@ -68,11 +69,13 @@ public synchronized void register() { } @Override - public synchronized boolean deregister(String ignored) { + public synchronized boolean deregister(final String ignored) { if (registeredTaskCount <= 0) { throw new IllegalStateException("registeredTaskCount <= 0"); } + // no need to discard events of pipe + try { if (registeredTaskCount > 1) { return false; @@ -91,45 +94,12 @@ public synchronized boolean deregister(String ignored) { } } - @Override - public synchronized void start() { - if (runningTaskCount < 0) { - throw new IllegalStateException("runningTaskCount < 0"); - } - - if (runningTaskCount == 0) { - executor.start(subtask.getTaskID()); - } - - runningTaskCount++; - LOGGER.info( - "Start subtask {}. runningTaskCount: {}, registeredTaskCount: {}", - subtask, - runningTaskCount, - registeredTaskCount); - } - - @Override - public synchronized void stop() { - if (runningTaskCount <= 0) { - throw new IllegalStateException("runningTaskCount <= 0"); - } - - if (runningTaskCount == 1) { - executor.stop(subtask.getTaskID()); - } - - runningTaskCount--; - LOGGER.info( - "Stop subtask {}. runningTaskCount: {}, registeredTaskCount: {}", - subtask, - runningTaskCount, - registeredTaskCount); - } - @Override public synchronized void close() { - executor.deregister(subtask.getTaskID()); - SubscriptionAgent.broker().unbindPrefetchingQueue((SubscriptionConnectorSubtask) subtask); + super.close(); + + final String consumerGroupId = ((SubscriptionConnectorSubtask) subtask).getConsumerGroupId(); + final String topicName = ((SubscriptionConnectorSubtask) subtask).getTopicName(); + SubscriptionAgent.broker().unbindPrefetchingQueue(consumerGroupId, topicName, true); } }