From 3b3d6f03c2fb1827f9b778852ccb1ed4927d18c3 Mon Sep 17 00:00:00 2001 From: VGalaxies Date: Wed, 12 Jun 2024 16:35:44 +0800 Subject: [PATCH 1/3] basic apply patch --- .../iotdb/SubscriptionSessionExample.java | 55 +++- .../it/AbstractSubscriptionIT.java | 49 ++++ .../it/IoTDBSubscriptionITConstant.java | 19 +- .../IoTDBSubscriptionRestartIT.java | 53 ++-- .../it/dual/AbstractSubscriptionDualIT.java | 24 +- .../IoTDBSubscriptionConsumerGroupIT.java | 31 +-- .../IoTDBSubscriptionTimePrecisionIT.java | 33 +-- .../it/dual/IoTDBSubscriptionTopicIT.java | 254 ++++++++++++------ .../it/local/AbstractSubscriptionLocalIT.java | 45 ++++ .../it/local/IoTDBSubscriptionBasicIT.java | 117 +++----- .../local/IoTDBSubscriptionIdempotentIT.java | 14 +- ...bscriptionIdentifierSemanticException.java | 42 +++ .../consumer/SubscriptionConsumer.java | 102 +++++-- .../consumer/SubscriptionProvider.java | 12 +- .../consumer/SubscriptionPullConsumer.java | 9 +- .../consumer/SubscriptionPushConsumer.java | 8 +- .../payload/SubscriptionFileHandler.java | 20 +- .../subscription/util/IdentifierUtils.java | 51 ++++ .../agent/SubscriptionAgentLauncher.java | 8 +- .../agent/SubscriptionBrokerAgent.java | 11 +- .../agent/SubscriptionConsumerAgent.java | 22 +- .../agent/SubscriptionReceiverAgent.java | 6 +- .../agent/SubscriptionTopicAgent.java | 20 +- .../broker/SubscriptionBroker.java | 46 +++- .../broker/SubscriptionPrefetchingQueue.java | 32 ++- .../SubscriptionPrefetchingTabletsQueue.java | 10 + .../SubscriptionPrefetchingTsFileQueue.java | 25 ++ .../subscription/event/SubscriptionEvent.java | 6 + .../receiver/SubscriptionReceiverV1.java | 36 +-- .../subtask/SubscriptionConnectorSubtask.java | 18 +- ...SubscriptionConnectorSubtaskLifeCycle.java | 57 ++-- 31 files changed, 818 insertions(+), 417 deletions(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/subscription/it/AbstractSubscriptionIT.java rename integration-test/src/test/java/org/apache/iotdb/subscription/it/{local => cluster}/IoTDBSubscriptionRestartIT.java (90%) create mode 100644 integration-test/src/test/java/org/apache/iotdb/subscription/it/local/AbstractSubscriptionLocalIT.java create mode 100644 iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionIdentifierSemanticException.java create mode 100644 iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/util/IdentifierUtils.java 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..b633fb9578fec 100644 --- a/example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java +++ b/example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java @@ -25,14 +25,23 @@ import org.apache.iotdb.rpc.subscription.config.TopicConstant; import org.apache.iotdb.session.Session; import org.apache.iotdb.session.subscription.SubscriptionSession; +import org.apache.iotdb.session.subscription.consumer.AckStrategy; +import org.apache.iotdb.session.subscription.consumer.ConsumeResult; import org.apache.iotdb.session.subscription.consumer.SubscriptionPullConsumer; +import org.apache.iotdb.session.subscription.consumer.SubscriptionPushConsumer; import org.apache.iotdb.session.subscription.payload.SubscriptionMessage; import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet; +import org.apache.iotdb.session.subscription.payload.SubscriptionTsFileHandler; 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.nio.file.Paths; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Properties; @@ -46,11 +55,14 @@ 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'`"; + private static final String TOPIC_3 = "`\"topic3\"`"; + private static final String TOPIC_4 = "`\"top \\.i.c4\"`"; - 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 +78,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 +116,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 +168,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 +178,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 +205,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 +235,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..19c033af151ba 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 @@ -20,12 +20,15 @@ package org.apache.iotdb.subscription.it.dual; import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; +import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionResp; import org.apache.iotdb.confignode.rpc.thrift.TShowTopicInfo; import org.apache.iotdb.confignode.rpc.thrift.TShowTopicReq; import org.apache.iotdb.db.it.utils.TestUtils; import org.apache.iotdb.isession.ISession; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.MultiClusterIT2Subscription; +import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.subscription.config.TopicConstant; import org.apache.iotdb.rpc.subscription.exception.SubscriptionRuntimeCriticalException; import org.apache.iotdb.session.subscription.SubscriptionSession; @@ -33,8 +36,11 @@ import org.apache.iotdb.session.subscription.payload.SubscriptionMessage; import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant; +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 org.apache.tsfile.write.record.Tablet; -import org.awaitility.Awaitility; import org.junit.Assert; import org.junit.Test; import org.junit.experimental.categories.Category; @@ -45,17 +51,18 @@ import java.sql.Connection; import java.sql.Statement; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; 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) @@ -64,6 +71,19 @@ public class IoTDBSubscriptionTopicIT extends AbstractSubscriptionDualIT { private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBSubscriptionTopicIT.class); + @Override + protected void setUpConfig() { + super.setUpConfig(); + + // Shorten heartbeat and sync interval to avoid timeout of query mode test + senderEnv + .getConfig() + .getCommonConfig() + .setPipeHeartbeatIntervalSecondsForCollectingPipeMeta(30); + senderEnv.getConfig().getCommonConfig().setPipeMetaSyncerInitialSyncDelayMinutes(1); + senderEnv.getConfig().getCommonConfig().setPipeMetaSyncerSyncIntervalMinutes(1); + } + @Test public void testTopicPathSubscription() throws Exception { // Insert some historical data on sender @@ -136,7 +156,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 +165,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 +254,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 +263,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 +348,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 +361,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 +400,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 +472,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 +481,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 +511,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 +523,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 +536,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 +547,109 @@ 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); + } + + @Test + public void testTopicWithQueryMode() throws Exception { + // Insert some historical data + try (final ISession session = senderEnv.getSessionConnection()) { + for (int i = 0; i < 100; ++i) { + session.executeNonQueryStatement( + String.format("insert into root.db.d1(time, s1) values (%s, 1)", i)); + } + session.executeNonQueryStatement("flush"); + } catch (final Exception e) { + e.printStackTrace(); + fail(e.getMessage()); + } + + // Create topic + final String topicName = "topic11"; + final String host = senderEnv.getIP(); + final int port = Integer.parseInt(senderEnv.getPort()); + try (final SubscriptionSession session = new SubscriptionSession(host, port)) { + session.open(); + final Properties config = new Properties(); + config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE); + config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_QUERY_VALUE); + session.createTopic(topicName, config); + } catch (final Exception e) { + e.printStackTrace(); + fail(e.getMessage()); + } + + // Subscription + final AtomicInteger rowCount = new AtomicInteger(); + final AtomicBoolean isClosed = new AtomicBoolean(false); + final Thread thread = + new Thread( + () -> { + try (final SubscriptionPullConsumer consumer = + new SubscriptionPullConsumer.Builder() + .host(host) + .port(port) + .consumerId("c1") + .consumerGroupId("cg1") + .autoCommit(false) + .buildPullConsumer()) { + consumer.open(); + consumer.subscribe(topicName); + while (!isClosed.get()) { + LockSupport.parkNanos(IoTDBSubscriptionITConstant.SLEEP_NS); // wait some time + final List messages = + consumer.poll(IoTDBSubscriptionITConstant.POLL_TIMEOUT_MS); + for (final SubscriptionMessage message : messages) { + try (final TsFileReader tsFileReader = + message.getTsFileHandler().openReader()) { + final Path path = new Path("root.db.d1", "s1", true); + final QueryDataSet dataSet = + tsFileReader.query( + QueryExpression.create(Collections.singletonList(path), null)); + while (dataSet.hasNext()) { + dataSet.next(); + rowCount.addAndGet(1); + } + } + } + consumer.commitSync(messages); + } + // Exiting the loop represents passing the awaitility test, at this point the result + // of 'show subscription' is empty, so there is no need to explicitly unsubscribe. + } catch (final Exception e) { + e.printStackTrace(); + // Avoid failure + } finally { + LOGGER.info("consumer exiting..."); + } + }, + String.format("%s - consumer", testName.getMethodName())); + thread.start(); + + try { + // Keep retrying if there are execution failures + AWAIT.untilAsserted( + () -> { + // Check row count + Assert.assertEquals(100, rowCount.get()); + // Check empty subscription + try (final SyncConfigNodeIServiceClient client = + (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) { + final TShowSubscriptionResp showSubscriptionResp = + client.showSubscription(new TShowSubscriptionReq()); + Assert.assertEquals( + RpcUtils.SUCCESS_STATUS.getCode(), showSubscriptionResp.status.getCode()); + Assert.assertNotNull(showSubscriptionResp.subscriptionInfoList); + Assert.assertEquals(0, showSubscriptionResp.subscriptionInfoList.size()); + } + }); + } catch (final Exception e) { + e.printStackTrace(); + fail(e.getMessage()); + } finally { + isClosed.set(true); + thread.join(); + } } private void testTopicInvalidRuntimeConfigTemplate( @@ -600,7 +706,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 +735,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/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..f3bd0371fdd2c 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,18 @@ public List poll(final String consumerId, final Set t final String topicName = entry.getKey(); final SubscriptionPrefetchingQueue prefetchingQueue = entry.getValue(); if (topicNames.contains(topicName)) { + // before determining if it is closed + if (prefetchingQueue.isCompleted()) { + LOGGER.info( + "Subscription: prefetching queue bound to topic [{}] is completed, return termination response to client", + topicName); + events.add(prefetchingQueue.generateSubscriptionPollTerminationResponse()); + continue; + } + 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 +107,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 +133,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 +175,30 @@ 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()); + + // mark prefetching queue completed only for topic of query mode + if (SubscriptionAgent.topic().getTopicMode(topicName).equals(TopicConstant.MODE_QUERY_VALUE)) { + prefetchingQueue.markCompleted(); + } + + if (doRemove) { + topicNameToPrefetchingQueue.remove(topicName); + SubscriptionPrefetchingQueueMetrics.getInstance() + .deregister(prefetchingQueue.getPrefetchingQueueId()); + } } public void executePrefetch(final String topicName) { @@ -176,6 +208,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..75f31d48415c8 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 @@ -24,12 +24,16 @@ import org.apache.iotdb.db.pipe.agent.PipeAgent; import org.apache.iotdb.db.subscription.event.SubscriptionEvent; 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.rpc.subscription.payload.poll.SubscriptionCommitContext; +import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponse; +import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponseType; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.Collections; import java.util.Map; import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; @@ -46,6 +50,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 +68,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 +162,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..c61d36263055a 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 @@ -25,6 +25,7 @@ import org.apache.iotdb.db.pipe.event.UserDefinedEnrichedEvent; import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; +import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager; import org.apache.iotdb.db.subscription.event.SubscriptionEvent; @@ -63,6 +64,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..7e0b5a0787e2f 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 @@ -22,7 +22,9 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent; 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.terminate.PipeTerminateEvent; 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 +65,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 +95,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..41e690cbce918 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,15 @@ 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(); + + // Here, the prefetching queue is not actually removed, because it's uncertain whether the + // corresponding underlying pipe is automatically terminated. The actual removal is carried out + // when dropping the subscription. + final String consumerGroupId = ((SubscriptionConnectorSubtask) subtask).getConsumerGroupId(); + final String topicName = ((SubscriptionConnectorSubtask) subtask).getTopicName(); + SubscriptionAgent.broker().unbindPrefetchingQueue(consumerGroupId, topicName, false); } } From 2865746bbd7267778afbcdc633e46acf240c583e Mon Sep 17 00:00:00 2001 From: VGalaxies Date: Wed, 12 Jun 2024 16:43:16 +0800 Subject: [PATCH 2/3] fixup! basic apply patch --- .../iotdb/SubscriptionSessionExample.java | 7 - .../it/dual/IoTDBSubscriptionTopicIT.java | 123 ------------------ .../response/PipeSubscribeSubscribeResp.java | 51 +++++++- .../PipeSubscribeUnsubscribeResp.java | 51 +++++++- .../broker/SubscriptionBroker.java | 13 -- .../broker/SubscriptionPrefetchingQueue.java | 4 - .../SubscriptionPrefetchingTabletsQueue.java | 1 - .../SubscriptionPrefetchingTsFileQueue.java | 1 - 8 files changed, 96 insertions(+), 155 deletions(-) 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 b633fb9578fec..3aea4e0a76245 100644 --- a/example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java +++ b/example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java @@ -25,13 +25,9 @@ import org.apache.iotdb.rpc.subscription.config.TopicConstant; import org.apache.iotdb.session.Session; import org.apache.iotdb.session.subscription.SubscriptionSession; -import org.apache.iotdb.session.subscription.consumer.AckStrategy; -import org.apache.iotdb.session.subscription.consumer.ConsumeResult; import org.apache.iotdb.session.subscription.consumer.SubscriptionPullConsumer; -import org.apache.iotdb.session.subscription.consumer.SubscriptionPushConsumer; import org.apache.iotdb.session.subscription.payload.SubscriptionMessage; import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet; -import org.apache.iotdb.session.subscription.payload.SubscriptionTsFileHandler; import org.apache.tsfile.read.TsFileReader; import org.apache.tsfile.read.common.Path; @@ -39,7 +35,6 @@ import org.apache.tsfile.read.query.dataset.QueryDataSet; import java.io.IOException; -import java.nio.file.Paths; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -56,8 +51,6 @@ public class SubscriptionSessionExample { private static final String TOPIC_1 = "topic1"; private static final String TOPIC_2 = "`'topic2'`"; - private static final String TOPIC_3 = "`\"topic3\"`"; - private static final String TOPIC_4 = "`\"top \\.i.c4\"`"; private static final long SLEEP_NS = 1_000_000_000L; private static final long POLL_TIMEOUT_MS = 10_000L; 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 19c033af151ba..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 @@ -20,15 +20,12 @@ package org.apache.iotdb.subscription.it.dual; import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; -import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionReq; -import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionResp; import org.apache.iotdb.confignode.rpc.thrift.TShowTopicInfo; import org.apache.iotdb.confignode.rpc.thrift.TShowTopicReq; import org.apache.iotdb.db.it.utils.TestUtils; import org.apache.iotdb.isession.ISession; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.MultiClusterIT2Subscription; -import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.subscription.config.TopicConstant; import org.apache.iotdb.rpc.subscription.exception.SubscriptionRuntimeCriticalException; import org.apache.iotdb.session.subscription.SubscriptionSession; @@ -36,10 +33,6 @@ import org.apache.iotdb.session.subscription.payload.SubscriptionMessage; import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant; -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 org.apache.tsfile.write.record.Tablet; import org.junit.Assert; import org.junit.Test; @@ -51,7 +44,6 @@ import java.sql.Connection; import java.sql.Statement; import java.util.ArrayList; -import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; @@ -71,19 +63,6 @@ public class IoTDBSubscriptionTopicIT extends AbstractSubscriptionDualIT { private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBSubscriptionTopicIT.class); - @Override - protected void setUpConfig() { - super.setUpConfig(); - - // Shorten heartbeat and sync interval to avoid timeout of query mode test - senderEnv - .getConfig() - .getCommonConfig() - .setPipeHeartbeatIntervalSecondsForCollectingPipeMeta(30); - senderEnv.getConfig().getCommonConfig().setPipeMetaSyncerInitialSyncDelayMinutes(1); - senderEnv.getConfig().getCommonConfig().setPipeMetaSyncerSyncIntervalMinutes(1); - } - @Test public void testTopicPathSubscription() throws Exception { // Insert some historical data on sender @@ -550,108 +529,6 @@ public void testTopicInvalidProcessorConfig() throws Exception { testTopicInvalidRuntimeConfigTemplate("topic10", config); } - @Test - public void testTopicWithQueryMode() throws Exception { - // Insert some historical data - try (final ISession session = senderEnv.getSessionConnection()) { - for (int i = 0; i < 100; ++i) { - session.executeNonQueryStatement( - String.format("insert into root.db.d1(time, s1) values (%s, 1)", i)); - } - session.executeNonQueryStatement("flush"); - } catch (final Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - - // Create topic - final String topicName = "topic11"; - final String host = senderEnv.getIP(); - final int port = Integer.parseInt(senderEnv.getPort()); - try (final SubscriptionSession session = new SubscriptionSession(host, port)) { - session.open(); - final Properties config = new Properties(); - config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE); - config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_QUERY_VALUE); - session.createTopic(topicName, config); - } catch (final Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - - // Subscription - final AtomicInteger rowCount = new AtomicInteger(); - final AtomicBoolean isClosed = new AtomicBoolean(false); - final Thread thread = - new Thread( - () -> { - try (final SubscriptionPullConsumer consumer = - new SubscriptionPullConsumer.Builder() - .host(host) - .port(port) - .consumerId("c1") - .consumerGroupId("cg1") - .autoCommit(false) - .buildPullConsumer()) { - consumer.open(); - consumer.subscribe(topicName); - while (!isClosed.get()) { - LockSupport.parkNanos(IoTDBSubscriptionITConstant.SLEEP_NS); // wait some time - final List messages = - consumer.poll(IoTDBSubscriptionITConstant.POLL_TIMEOUT_MS); - for (final SubscriptionMessage message : messages) { - try (final TsFileReader tsFileReader = - message.getTsFileHandler().openReader()) { - final Path path = new Path("root.db.d1", "s1", true); - final QueryDataSet dataSet = - tsFileReader.query( - QueryExpression.create(Collections.singletonList(path), null)); - while (dataSet.hasNext()) { - dataSet.next(); - rowCount.addAndGet(1); - } - } - } - consumer.commitSync(messages); - } - // Exiting the loop represents passing the awaitility test, at this point the result - // of 'show subscription' is empty, so there is no need to explicitly unsubscribe. - } catch (final Exception e) { - e.printStackTrace(); - // Avoid failure - } finally { - LOGGER.info("consumer exiting..."); - } - }, - String.format("%s - consumer", testName.getMethodName())); - thread.start(); - - try { - // Keep retrying if there are execution failures - AWAIT.untilAsserted( - () -> { - // Check row count - Assert.assertEquals(100, rowCount.get()); - // Check empty subscription - try (final SyncConfigNodeIServiceClient client = - (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) { - final TShowSubscriptionResp showSubscriptionResp = - client.showSubscription(new TShowSubscriptionReq()); - Assert.assertEquals( - RpcUtils.SUCCESS_STATUS.getCode(), showSubscriptionResp.status.getCode()); - Assert.assertNotNull(showSubscriptionResp.subscriptionInfoList); - Assert.assertEquals(0, showSubscriptionResp.subscriptionInfoList.size()); - } - }); - } catch (final Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } finally { - isClosed.set(true); - thread.join(); - } - } - private void testTopicInvalidRuntimeConfigTemplate( final String topicName, final Properties config) throws Exception { // Create topic 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-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 f3bd0371fdd2c..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,14 +65,6 @@ public List poll(final String consumerId, final Set t final String topicName = entry.getKey(); final SubscriptionPrefetchingQueue prefetchingQueue = entry.getValue(); if (topicNames.contains(topicName)) { - // before determining if it is closed - if (prefetchingQueue.isCompleted()) { - LOGGER.info( - "Subscription: prefetching queue bound to topic [{}] is completed, return termination response to client", - topicName); - events.add(prefetchingQueue.generateSubscriptionPollTerminationResponse()); - continue; - } if (prefetchingQueue.isClosed()) { LOGGER.warn("Subscription: prefetching queue bound to topic [{}] is closed", topicName); continue; @@ -189,11 +181,6 @@ public void unbindPrefetchingQueue(final String topicName, final boolean doRemov // clean up events in prefetching queue, this operation is idempotent prefetchingQueue.cleanup(); - // mark prefetching queue completed only for topic of query mode - if (SubscriptionAgent.topic().getTopicMode(topicName).equals(TopicConstant.MODE_QUERY_VALUE)) { - prefetchingQueue.markCompleted(); - } - if (doRemove) { topicNameToPrefetchingQueue.remove(topicName); SubscriptionPrefetchingQueueMetrics.getInstance() 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 75f31d48415c8..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 @@ -24,16 +24,12 @@ import org.apache.iotdb.db.pipe.agent.PipeAgent; import org.apache.iotdb.db.subscription.event.SubscriptionEvent; 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.rpc.subscription.payload.poll.SubscriptionCommitContext; -import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponse; -import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponseType; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.Collections; import java.util.Map; import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; 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 c61d36263055a..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 @@ -25,7 +25,6 @@ import org.apache.iotdb.db.pipe.event.UserDefinedEnrichedEvent; import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; -import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager; import org.apache.iotdb.db.subscription.event.SubscriptionEvent; 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 7e0b5a0787e2f..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 @@ -22,7 +22,6 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent; 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.terminate.PipeTerminateEvent; 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; From f7b049e77c0250a1ecab792b3846d5083383d821 Mon Sep 17 00:00:00 2001 From: VGalaxies Date: Wed, 12 Jun 2024 16:46:26 +0800 Subject: [PATCH 3/3] fixup --- .../task/subtask/SubscriptionConnectorSubtaskLifeCycle.java | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) 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 41e690cbce918..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 @@ -98,11 +98,8 @@ public synchronized boolean deregister(final String ignored) { public synchronized void close() { super.close(); - // Here, the prefetching queue is not actually removed, because it's uncertain whether the - // corresponding underlying pipe is automatically terminated. The actual removal is carried out - // when dropping the subscription. final String consumerGroupId = ((SubscriptionConnectorSubtask) subtask).getConsumerGroupId(); final String topicName = ((SubscriptionConnectorSubtask) subtask).getTopicName(); - SubscriptionAgent.broker().unbindPrefetchingQueue(consumerGroupId, topicName, false); + SubscriptionAgent.broker().unbindPrefetchingQueue(consumerGroupId, topicName, true); } }