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 3aea4e0a76245..3c3ae618ce3ec 100644 --- a/example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java +++ b/example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java @@ -25,9 +25,13 @@ 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; @@ -35,6 +39,7 @@ 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; @@ -51,6 +56,8 @@ 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; @@ -226,10 +233,109 @@ private static void dataSubscription2() throws Exception { } } + /** multi push consumer subscribe topic with tsfile format and query mode */ + private static void dataSubscription3() throws Exception { + try (final SubscriptionSession subscriptionSession = new SubscriptionSession(HOST, PORT)) { + subscriptionSession.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); + subscriptionSession.createTopic(TOPIC_3, config); + } + + final List threads = new ArrayList<>(); + for (int i = 0; i < PARALLELISM; ++i) { + final int idx = i; + final Thread thread = + new Thread( + () -> { + // Subscription: builder-style ctor + try (final SubscriptionPushConsumer consumer3 = + new SubscriptionPushConsumer.Builder() + .consumerId("c" + idx) + .consumerGroupId("cg3") + .ackStrategy(AckStrategy.AFTER_CONSUME) + .consumeListener( + message -> { + // do something for SubscriptionTsFileHandler + System.out.println( + message.getTsFileHandler().getFile().getAbsolutePath()); + return ConsumeResult.SUCCESS; + }) + .buildPushConsumer()) { + consumer3.open(); + consumer3.subscribe(TOPIC_3); + while (consumer3.hasMoreData()) { + LockSupport.parkNanos(SLEEP_NS); // wait some time + } + } + }); + thread.start(); + threads.add(thread); + } + + for (final Thread thread : threads) { + thread.join(); + } + } + + /** multi pull consumer subscribe topic with tsfile format and query mode */ + private static void dataSubscription4() throws Exception { + try (final SubscriptionSession subscriptionSession = new SubscriptionSession(HOST, PORT)) { + subscriptionSession.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); + subscriptionSession.createTopic(TOPIC_4, config); + } + + final List threads = new ArrayList<>(); + for (int i = 0; i < PARALLELISM; ++i) { + final int idx = i; + final Thread thread = + new Thread( + () -> { + // Subscription: builder-style ctor + try (final SubscriptionPullConsumer consumer4 = + new SubscriptionPullConsumer.Builder() + .consumerId("c" + idx) + .consumerGroupId("cg4") + .autoCommit(true) + .fileSaveFsync(true) + .buildPullConsumer()) { + consumer4.open(); + consumer4.subscribe(TOPIC_4); + while (true) { + LockSupport.parkNanos(SLEEP_NS); // wait some time + if (!consumer4.hasMoreData()) { + break; + } + for (final SubscriptionMessage message : consumer4.poll(POLL_TIMEOUT_MS)) { + final SubscriptionTsFileHandler handler = message.getTsFileHandler(); + handler.moveFile( + Paths.get(System.getProperty("user.dir"), "new-subscription-dir") + .resolve(handler.getPath().getFileName())); + } + } + } catch (final IOException e) { + throw new RuntimeException(e); + } + }); + thread.start(); + threads.add(thread); + } + + for (final Thread thread : threads) { + thread.join(); + } + } + public static void main(final String[] args) throws Exception { prepareData(); // dataQuery(); // dataSubscription1(); - dataSubscription2(); + // dataSubscription2(); + // dataSubscription3(); + dataSubscription4(); } } diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java index 95fed66d08b21..095562db5a51f 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java @@ -438,6 +438,32 @@ public CommonConfig setCnConnectionTimeoutMs(int connectionTimeoutMs) { return this; } + @Override + public CommonConfig setPipeHeartbeatIntervalSecondsForCollectingPipeMeta( + int pipeHeartbeatIntervalSecondsForCollectingPipeMeta) { + setProperty( + "pipe_heartbeat_interval_seconds_for_collecting_pipe_meta", + String.valueOf(pipeHeartbeatIntervalSecondsForCollectingPipeMeta)); + return this; + } + + @Override + public CommonConfig setPipeMetaSyncerInitialSyncDelayMinutes( + long pipeMetaSyncerInitialSyncDelayMinutes) { + setProperty( + "pipe_meta_syncer_initial_sync_delay_minutes", + String.valueOf(pipeMetaSyncerInitialSyncDelayMinutes)); + return this; + } + + @Override + public CommonConfig setPipeMetaSyncerSyncIntervalMinutes(long pipeMetaSyncerSyncIntervalMinutes) { + setProperty( + "pipe_meta_syncer_sync_interval_minutes", + String.valueOf(pipeMetaSyncerSyncIntervalMinutes)); + return this; + } + // For part of the log directory public String getClusterConfigStr() { return fromConsensusFullNameToAbbr(properties.getProperty(CONFIG_NODE_CONSENSUS_PROTOCOL_CLASS)) diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java index 1851590dd411e..7724bcf09c0af 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java @@ -445,4 +445,29 @@ public CommonConfig setCnConnectionTimeoutMs(int connectionTimeoutMs) { cnConfig.setCnConnectionTimeoutMs(connectionTimeoutMs); return this; } + + @Override + public CommonConfig setPipeHeartbeatIntervalSecondsForCollectingPipeMeta( + int pipeHeartbeatIntervalSecondsForCollectingPipeMeta) { + dnConfig.setPipeHeartbeatIntervalSecondsForCollectingPipeMeta( + pipeHeartbeatIntervalSecondsForCollectingPipeMeta); + cnConfig.setPipeHeartbeatIntervalSecondsForCollectingPipeMeta( + pipeHeartbeatIntervalSecondsForCollectingPipeMeta); + return this; + } + + @Override + public CommonConfig setPipeMetaSyncerInitialSyncDelayMinutes( + long pipeMetaSyncerInitialSyncDelayMinutes) { + dnConfig.setPipeMetaSyncerInitialSyncDelayMinutes(pipeMetaSyncerInitialSyncDelayMinutes); + cnConfig.setPipeMetaSyncerInitialSyncDelayMinutes(pipeMetaSyncerInitialSyncDelayMinutes); + return this; + } + + @Override + public CommonConfig setPipeMetaSyncerSyncIntervalMinutes(long pipeMetaSyncerSyncIntervalMinutes) { + dnConfig.setPipeMetaSyncerSyncIntervalMinutes(pipeMetaSyncerSyncIntervalMinutes); + cnConfig.setPipeMetaSyncerSyncIntervalMinutes(pipeMetaSyncerSyncIntervalMinutes); + return this; + } } diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java index 0ac82fc871373..36675e646865d 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java @@ -313,4 +313,21 @@ public CommonConfig setTagAttributeTotalSize(int tagAttributeTotalSize) { public CommonConfig setCnConnectionTimeoutMs(int connectionTimeoutMs) { return this; } + + @Override + public CommonConfig setPipeHeartbeatIntervalSecondsForCollectingPipeMeta( + int pipeHeartbeatIntervalSecondsForCollectingPipeMeta) { + return this; + } + + @Override + public CommonConfig setPipeMetaSyncerInitialSyncDelayMinutes( + long pipeMetaSyncerInitialSyncDelayMinutes) { + return this; + } + + @Override + public CommonConfig setPipeMetaSyncerSyncIntervalMinutes(long pipeMetaSyncerSyncIntervalMinutes) { + return this; + } } diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java index af47c8254e532..c06c45ddcd42d 100644 --- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java @@ -139,4 +139,11 @@ CommonConfig setEnableAutoLeaderBalanceForIoTConsensus( CommonConfig setTagAttributeTotalSize(int tagAttributeTotalSize); CommonConfig setCnConnectionTimeoutMs(int connectionTimeoutMs); + + CommonConfig setPipeHeartbeatIntervalSecondsForCollectingPipeMeta( + int pipeHeartbeatIntervalSecondsForCollectingPipeMeta); + + CommonConfig setPipeMetaSyncerInitialSyncDelayMinutes(long pipeMetaSyncerInitialSyncDelayMinutes); + + CommonConfig setPipeMetaSyncerSyncIntervalMinutes(long pipeMetaSyncerSyncIntervalMinutes); } 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 ac7e83339844b..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,6 +36,10 @@ 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; @@ -44,6 +51,7 @@ 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; @@ -63,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 @@ -529,6 +550,108 @@ 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/config/TopicConfig.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java index fe62ba622a965..c3b5f24622c43 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java @@ -40,6 +40,24 @@ public TopicConfig(final Map attributes) { super(attributes); } + private static final Map LOOSE_RANGE_TIME_CONFIG = + new HashMap() { + { + put("history.loose-range", "time"); + put("realtime.loose-range", "time"); + } + }; + + private static final Map REALTIME_BATCH_MODE_CONFIG = + Collections.singletonMap("realtime.mode", "batch"); + private static final Map REALTIME_STREAM_MODE_CONFIG = + Collections.singletonMap("realtime.mode", "stream"); + + private static final Map QUERY_MODE_CONFIG = + Collections.singletonMap("mode", "query"); + private static final Map SUBSCRIBE_MODE_CONFIG = + Collections.singletonMap("mode", "subscribe"); + /////////////////////////////// de/ser /////////////////////////////// public void serialize(final DataOutputStream stream) throws IOException { @@ -87,8 +105,7 @@ public Map getAttributesWithTimeRange(final long creationTime) { // enable loose range when using tsfile format if (TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE.equals( attributes.getOrDefault(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_DEFAULT_VALUE))) { - attributesWithTimeRange.put("history.loose-range", "time"); - attributesWithTimeRange.put("realtime.loose-range", "time"); + attributesWithTimeRange.putAll(LOOSE_RANGE_TIME_CONFIG); } return attributesWithTimeRange; @@ -97,8 +114,15 @@ public Map getAttributesWithTimeRange(final long creationTime) { public Map getAttributesWithRealtimeMode() { return TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE.equals( attributes.getOrDefault(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_DEFAULT_VALUE)) - ? Collections.singletonMap("realtime.mode", "batch") - : Collections.singletonMap("realtime.mode", "hybrid"); + ? REALTIME_BATCH_MODE_CONFIG + : REALTIME_STREAM_MODE_CONFIG; + } + + public Map getAttributesWithSourceMode() { + return TopicConstant.MODE_QUERY_VALUE.equals( + attributes.getOrDefault(TopicConstant.MODE_KEY, TopicConstant.MODE_DEFAULT_VALUE)) + ? QUERY_MODE_CONFIG + : SUBSCRIBE_MODE_CONFIG; } public Map getAttributesWithProcessorPrefix() { diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java index bd44cf1bcc95d..a6e8ec901bf9f 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java @@ -30,6 +30,11 @@ public class TopicConstant { public static final String END_TIME_KEY = "end-time"; public static final String NOW_TIME_VALUE = "now"; + public static final String MODE_KEY = "mode"; + public static final String MODE_SUBSCRIBE_VALUE = "subscribe"; + public static final String MODE_QUERY_VALUE = "query"; + public static final String MODE_DEFAULT_VALUE = MODE_SUBSCRIBE_VALUE; + public static final String FORMAT_KEY = "format"; public static final String FORMAT_SESSION_DATA_SETS_HANDLER_VALUE = "SessionDataSetsHandler"; public static final String FORMAT_TS_FILE_HANDLER_VALUE = "TsFileHandler"; diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/poll/SubscriptionPollResponse.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/poll/SubscriptionPollResponse.java index d386903f124d0..01d173d274283 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/poll/SubscriptionPollResponse.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/poll/SubscriptionPollResponse.java @@ -95,6 +95,9 @@ public static SubscriptionPollResponse deserialize(final ByteBuffer buffer) { case ERROR: payload = new ErrorPayload().deserialize(buffer); break; + case TERMINATION: + payload = new TerminationPayload().deserialize(buffer); + break; default: LOGGER.warn("unexpected response type: {}, payload will be null", responseType); break; diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/poll/SubscriptionPollResponseType.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/poll/SubscriptionPollResponseType.java index a22c9590c65fc..b27791b36c538 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/poll/SubscriptionPollResponseType.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/poll/SubscriptionPollResponseType.java @@ -31,6 +31,8 @@ public enum SubscriptionPollResponseType { FILE_INIT((short) 2), FILE_PIECE((short) 3), FILE_SEAL((short) 4), + + TERMINATION((short) 5), ; private final short type; diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/poll/TerminationPayload.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/poll/TerminationPayload.java new file mode 100644 index 0000000000000..c2da3f92e6dd8 --- /dev/null +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/poll/TerminationPayload.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.payload.poll; + +import java.io.DataOutputStream; +import java.nio.ByteBuffer; + +public class TerminationPayload implements SubscriptionPollPayload { + + @Override + public void serialize(final DataOutputStream stream) { + // do nothing + } + + @Override + public SubscriptionPollPayload deserialize(final ByteBuffer buffer) { + // do nothing + return null; + } + + @Override + public String toString() { + return "TerminationPayload"; + } +} 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 a6b418f022010..410348c991b38 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 @@ -108,8 +108,12 @@ public String getConsumerGroupId() { return consumerGroupId; } - public Set getSubscribedTopicNames() { - return subscribedTopicNames; + /** + * @return When only subscribing to the query mode topics, if there is no new data to + * process, return {@code false}; otherwise, return {@code true}. + */ + public boolean hasMoreData() { + return !subscribedTopicNames.isEmpty(); } /////////////////////////////// ctor /////////////////////////////// @@ -393,6 +397,16 @@ protected List poll( } else { throw new SubscriptionRuntimeNonCriticalException(errorMessage); } + case TERMINATION: + final SubscriptionCommitContext commitContext = pollResponse.getCommitContext(); + final String topicNameToUnsubscribe = commitContext.getTopicName(); + LOGGER.info( + "Termination occurred when SubscriptionConsumer {} polling topics {}, unsubscribe topic {} automatically", + this, + topicNames, + topicNameToUnsubscribe); + unsubscribe(Collections.singleton(topicNameToUnsubscribe), false); + break; default: LOGGER.warn("unexpected response type: {}", responseType); break; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedure.java index 8710eec85225e..0894feb62ad9e 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedure.java @@ -229,6 +229,7 @@ protected void rollbackFromValidate(ConfigNodeProcedureEnv env) { protected void rollbackFromOperateOnConfigNodes(ConfigNodeProcedureEnv env) { LOGGER.info("CreateSubscriptionProcedure: rollbackFromOperateOnConfigNodes"); + // TODO: roll back from the last executed procedure to the first executed alterConsumerGroupProcedure.rollbackFromOperateOnConfigNodes(env); TSStatus response; @@ -283,6 +284,7 @@ protected void rollbackFromOperateOnConfigNodes(ConfigNodeProcedureEnv env) { protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOException { LOGGER.info("CreateSubscriptionProcedure: rollbackFromOperateOnDataNodes"); + // TODO: roll back from the last executed procedure to the first executed alterConsumerGroupProcedure.rollbackFromOperateOnDataNodes(env); // Push all topic metas to datanode, may be time-consuming diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java index efb5aa228272d..89da284ac0d95 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java @@ -62,9 +62,10 @@ public class DropSubscriptionProcedure extends AbstractOperateSubscriptionAndPip private TUnsubscribeReq unsubscribeReq; - private AlterConsumerGroupProcedure alterConsumerGroupProcedure; - private List alterTopicProcedures = new ArrayList<>(); + // NOTE: The 'drop pipe' operation should be performed before 'alter consumer group'. private List dropPipeProcedures = new ArrayList<>(); + private List alterTopicProcedures = new ArrayList<>(); + private AlterConsumerGroupProcedure alterConsumerGroupProcedure; // Record failed index of procedures to rollback properly. // We only record fail index when executing on config nodes, because when executing on data nodes @@ -76,7 +77,7 @@ public DropSubscriptionProcedure() { super(); } - public DropSubscriptionProcedure(TUnsubscribeReq unsubscribeReq) { + public DropSubscriptionProcedure(final TUnsubscribeReq unsubscribeReq) { this.unsubscribeReq = unsubscribeReq; } @@ -86,27 +87,28 @@ protected SubscriptionOperation getOperation() { } @Override - protected void executeFromValidate(ConfigNodeProcedureEnv env) throws SubscriptionException { + protected void executeFromValidate(final ConfigNodeProcedureEnv env) + throws SubscriptionException { LOGGER.info("DropSubscriptionProcedure: executeFromValidate"); subscriptionInfo.get().validateBeforeUnsubscribe(unsubscribeReq); // Construct AlterConsumerGroupProcedure - ConsumerGroupMeta updatedConsumerGroupMeta = + final ConsumerGroupMeta updatedConsumerGroupMeta = subscriptionInfo.get().deepCopyConsumerGroupMeta(unsubscribeReq.getConsumerGroupId()); // Get topics subscribed by no consumers in this group after this un-subscription - Set topicsUnsubByGroup = + final Set topicsUnsubByGroup = updatedConsumerGroupMeta.removeSubscription( unsubscribeReq.getConsumerId(), unsubscribeReq.getTopicNames()); alterConsumerGroupProcedure = new AlterConsumerGroupProcedure(updatedConsumerGroupMeta, subscriptionInfo); - for (String topic : unsubscribeReq.getTopicNames()) { + for (final String topic : unsubscribeReq.getTopicNames()) { if (topicsUnsubByGroup.contains(topic)) { // Topic will be subscribed by no consumers in this group - TopicMeta updatedTopicMeta = subscriptionInfo.get().deepCopyTopicMeta(topic); + final TopicMeta updatedTopicMeta = subscriptionInfo.get().deepCopyTopicMeta(topic); updatedTopicMeta.removeSubscribedConsumerGroup(unsubscribeReq.getConsumerGroupId()); alterTopicProcedures.add(new AlterTopicProcedure(updatedTopicMeta, subscriptionInfo)); @@ -118,38 +120,39 @@ protected void executeFromValidate(ConfigNodeProcedureEnv env) throws Subscripti } } - alterConsumerGroupProcedure.executeFromValidate(env); + // Validate DropPipeProcedureV2s + for (final DropPipeProcedureV2 dropPipeProcedure : dropPipeProcedures) { + dropPipeProcedure.executeFromValidateTask(env); + dropPipeProcedure.executeFromCalculateInfoForTask(env); + } - for (AlterTopicProcedure alterTopicProcedure : alterTopicProcedures) { + // Validate AlterTopicProcedures + for (final AlterTopicProcedure alterTopicProcedure : alterTopicProcedures) { alterTopicProcedure.executeFromValidate(env); } - for (DropPipeProcedureV2 dropPipeProcedure : dropPipeProcedures) { - dropPipeProcedure.executeFromValidateTask(env); - dropPipeProcedure.executeFromCalculateInfoForTask(env); - } + // Validate AlterConsumerGroupProcedure + alterConsumerGroupProcedure.executeFromValidate(env); } @Override - protected void executeFromOperateOnConfigNodes(ConfigNodeProcedureEnv env) + protected void executeFromOperateOnConfigNodes(final ConfigNodeProcedureEnv env) throws SubscriptionException { LOGGER.info("DropSubscriptionProcedure: executeFromOperateOnConfigNodes"); - alterConsumerGroupProcedure.executeFromOperateOnConfigNodes(env); - TSStatus response; - List alterTopicPlans = - alterTopicProcedures.stream() - .map(AlterTopicProcedure::getUpdatedTopicMeta) - .map(AlterTopicPlan::new) + // Execute DropPipeProcedureV2s + final List dropPipePlans = + dropPipeProcedures.stream() + .map(proc -> new DropPipePlanV2(proc.getPipeName())) .collect(Collectors.toList()); try { response = env.getConfigManager() .getConsensusManager() - .write(new AlterMultipleTopicsPlan(alterTopicPlans)); - } catch (ConsensusException e) { + .write(new OperateMultiplePipesPlanV2(dropPipePlans)); + } catch (final ConsensusException e) { LOGGER.warn("Failed in the write API executing the consensus layer due to: ", e); response = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()); response.setMessage(e.getMessage()); @@ -157,19 +160,21 @@ protected void executeFromOperateOnConfigNodes(ConfigNodeProcedureEnv env) if (response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode() && response.getSubStatusSize() > 0) { // Record the failed index for rollback - alterTopicProcedureFailIndexOnCN = response.getSubStatusSize() - 1; + dropPipeProcedureFailIndexOnCN = response.getSubStatusSize() - 1; } - List dropPipePlans = - dropPipeProcedures.stream() - .map(proc -> new DropPipePlanV2(proc.getPipeName())) + // Execute AlterTopicProcedures + final List alterTopicPlans = + alterTopicProcedures.stream() + .map(AlterTopicProcedure::getUpdatedTopicMeta) + .map(AlterTopicPlan::new) .collect(Collectors.toList()); try { response = env.getConfigManager() .getConsensusManager() - .write(new OperateMultiplePipesPlanV2(dropPipePlans)); - } catch (ConsensusException e) { + .write(new AlterMultipleTopicsPlan(alterTopicPlans)); + } catch (final ConsensusException e) { LOGGER.warn("Failed in the write API executing the consensus layer due to: ", e); response = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()); response.setMessage(e.getMessage()); @@ -177,34 +182,24 @@ protected void executeFromOperateOnConfigNodes(ConfigNodeProcedureEnv env) if (response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode() && response.getSubStatusSize() > 0) { // Record the failed index for rollback - dropPipeProcedureFailIndexOnCN = response.getSubStatusSize() - 1; + alterTopicProcedureFailIndexOnCN = response.getSubStatusSize() - 1; } + + // Execute AlterConsumerGroupProcedure + alterConsumerGroupProcedure.executeFromOperateOnConfigNodes(env); } @Override - protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) + protected void executeFromOperateOnDataNodes(final ConfigNodeProcedureEnv env) throws SubscriptionException, IOException { LOGGER.info("DropSubscriptionProcedure: executeFromOperateOnDataNodes"); - alterConsumerGroupProcedure.executeFromOperateOnDataNodes(env); - - // push topic meta to data nodes - List topicMetaBinaryList = new ArrayList<>(); - for (AlterTopicProcedure alterTopicProcedure : alterTopicProcedures) { - topicMetaBinaryList.add(alterTopicProcedure.getUpdatedTopicMeta().serialize()); - } - if (pushTopicMetaHasException(env.pushMultiTopicMetaToDataNodes(topicMetaBinaryList))) { - // If not all topic meta are pushed successfully, the meta can be pushed during meta sync. - LOGGER.warn( - "Failed to alter topics when creating subscription, metadata will be synchronized later."); - } - - // push pipe meta to data nodes - List pipeNames = + // Push pipe meta to data nodes + final List pipeNames = dropPipeProcedures.stream() .map(DropPipeProcedureV2::getPipeName) .collect(Collectors.toList()); - String exceptionMessage = + final String exceptionMessage = AbstractOperatePipeProcedureV2.parsePushPipeMetaExceptionForPipe( null, dropMultiPipeOnDataNodes(pipeNames, env)); if (!exceptionMessage.isEmpty()) { @@ -214,23 +209,37 @@ protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) pipeNames, exceptionMessage); } + + // Push topic meta to data nodes + final List topicMetaBinaryList = new ArrayList<>(); + for (final AlterTopicProcedure alterTopicProcedure : alterTopicProcedures) { + topicMetaBinaryList.add(alterTopicProcedure.getUpdatedTopicMeta().serialize()); + } + if (pushTopicMetaHasException(env.pushMultiTopicMetaToDataNodes(topicMetaBinaryList))) { + // If not all topic meta are pushed successfully, the meta can be pushed during meta sync. + LOGGER.warn( + "Failed to alter topics when creating subscription, metadata will be synchronized later."); + } + + // Push consumer group meta to data nodes + alterConsumerGroupProcedure.executeFromOperateOnDataNodes(env); } @Override - protected void rollbackFromValidate(ConfigNodeProcedureEnv env) { + protected void rollbackFromValidate(final ConfigNodeProcedureEnv env) { LOGGER.info("DropSubscriptionProcedure: rollbackFromLock"); } @Override - protected void rollbackFromOperateOnConfigNodes(ConfigNodeProcedureEnv env) { + protected void rollbackFromOperateOnConfigNodes(final ConfigNodeProcedureEnv env) { LOGGER.info("DropSubscriptionProcedure: rollbackFromOperateOnConfigNodes"); + // Rollback AlterConsumerGroupProcedure alterConsumerGroupProcedure.rollbackFromOperateOnConfigNodes(env); + // Rollback AlterTopicProcedures TSStatus response; - - // rollback alterTopicProcedures - List alterTopicRollbackPlans = new ArrayList<>(); + final List alterTopicRollbackPlans = new ArrayList<>(); for (int i = 0; i <= Math.min(alterTopicProcedureFailIndexOnCN, alterTopicProcedures.size()); i++) { @@ -242,23 +251,25 @@ protected void rollbackFromOperateOnConfigNodes(ConfigNodeProcedureEnv env) { env.getConfigManager() .getConsensusManager() .write(new AlterMultipleTopicsPlan(alterTopicRollbackPlans)); - } catch (ConsensusException e) { + } catch (final ConsensusException e) { LOGGER.warn("Failed in the write API executing the consensus layer due to: ", e); response = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()); response.setMessage(e.getMessage()); } - // if failed to rollback, throw exception + // If failed to rollback, throw exception if (response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { throw new SubscriptionException(response.getMessage()); } - // Do nothing to rollback dropPipeProcedures + // Do nothing to rollback DropPipeProcedureV2s } @Override - protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOException { + protected void rollbackFromOperateOnDataNodes(final ConfigNodeProcedureEnv env) + throws IOException { LOGGER.info("DropSubscriptionProcedure: rollbackFromOperateOnDataNodes"); + // Rollback AlterConsumerGroupProcedure alterConsumerGroupProcedure.rollbackFromOperateOnDataNodes(env); // Push all topic metas to datanode, may be time-consuming @@ -268,7 +279,7 @@ protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws } // Push all pipe metas to datanode, may be time-consuming - String exceptionMessage = + final String exceptionMessage = AbstractOperatePipeProcedureV2.parsePushPipeMetaExceptionForPipe( null, AbstractOperatePipeProcedureV2.pushPipeMetaToDataNodes(env, pipeTaskInfo)); if (!exceptionMessage.isEmpty()) { @@ -279,7 +290,7 @@ protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws } @Override - public void serialize(DataOutputStream stream) throws IOException { + public void serialize(final DataOutputStream stream) throws IOException { stream.writeShort(ProcedureType.DROP_SUBSCRIPTION_PROCEDURE.getTypeCode()); super.serialize(stream); @@ -289,7 +300,7 @@ public void serialize(DataOutputStream stream) throws IOException { final int size = unsubscribeReq.getTopicNamesSize(); ReadWriteIOUtils.write(size, stream); if (size != 0) { - for (String topicName : unsubscribeReq.getTopicNames()) { + for (final String topicName : unsubscribeReq.getTopicNames()) { ReadWriteIOUtils.write(topicName, stream); } } @@ -306,7 +317,7 @@ public void serialize(DataOutputStream stream) throws IOException { if (alterTopicProcedures != null) { ReadWriteIOUtils.write(true, stream); ReadWriteIOUtils.write(alterTopicProcedures.size(), stream); - for (AlterTopicProcedure topicProcedure : alterTopicProcedures) { + for (final AlterTopicProcedure topicProcedure : alterTopicProcedures) { topicProcedure.serialize(stream); } } else { @@ -317,7 +328,7 @@ public void serialize(DataOutputStream stream) throws IOException { if (dropPipeProcedures != null) { ReadWriteIOUtils.write(true, stream); ReadWriteIOUtils.write(dropPipeProcedures.size(), stream); - for (AbstractOperatePipeProcedureV2 pipeProcedure : dropPipeProcedures) { + for (final AbstractOperatePipeProcedureV2 pipeProcedure : dropPipeProcedures) { pipeProcedure.serialize(stream); } } else { @@ -326,7 +337,7 @@ public void serialize(DataOutputStream stream) throws IOException { } @Override - public void deserialize(ByteBuffer byteBuffer) { + public void deserialize(final ByteBuffer byteBuffer) { super.deserialize(byteBuffer); unsubscribeReq = @@ -355,7 +366,7 @@ public void deserialize(ByteBuffer byteBuffer) { // This readShort should return ALTER_TOPIC_PROCEDURE, and we ignore it. ReadWriteIOUtils.readShort(byteBuffer); - AlterTopicProcedure topicProcedure = new AlterTopicProcedure(); + final AlterTopicProcedure topicProcedure = new AlterTopicProcedure(); topicProcedure.deserialize(byteBuffer); alterTopicProcedures.add(topicProcedure); } @@ -366,9 +377,9 @@ public void deserialize(ByteBuffer byteBuffer) { size = ReadWriteIOUtils.readInt(byteBuffer); for (int i = 0; i < size; ++i) { // This readShort should return DROP_PIPE_PROCEDURE. - short typeCode = ReadWriteIOUtils.readShort(byteBuffer); + final short typeCode = ReadWriteIOUtils.readShort(byteBuffer); if (typeCode == ProcedureType.DROP_PIPE_PROCEDURE_V2.getTypeCode()) { - DropPipeProcedureV2 dropPipeProcedureV2 = new DropPipeProcedureV2(); + final DropPipeProcedureV2 dropPipeProcedureV2 = new DropPipeProcedureV2(); dropPipeProcedureV2.deserialize(byteBuffer); dropPipeProcedures.add(dropPipeProcedureV2); } @@ -377,14 +388,14 @@ public void deserialize(ByteBuffer byteBuffer) { } @Override - public boolean equals(Object o) { + public boolean equals(final Object o) { if (this == o) { return true; } if (o == null || getClass() != o.getClass()) { return false; } - DropSubscriptionProcedure that = (DropSubscriptionProcedure) o; + final DropSubscriptionProcedure that = (DropSubscriptionProcedure) o; return Objects.equals(getProcId(), that.getProcId()) && Objects.equals(getCurrentState(), that.getCurrentState()) && getCycles() == that.getCycles() @@ -408,7 +419,7 @@ public int hashCode() { @TestOnly public void setAlterConsumerGroupProcedure( - AlterConsumerGroupProcedure alterConsumerGroupProcedure) { + final AlterConsumerGroupProcedure alterConsumerGroupProcedure) { this.alterConsumerGroupProcedure = alterConsumerGroupProcedure; } @@ -418,7 +429,7 @@ public AlterConsumerGroupProcedure getAlterConsumerGroupProcedure() { } @TestOnly - public void setAlterTopicProcedures(List alterTopicProcedures) { + public void setAlterTopicProcedures(final List alterTopicProcedures) { this.alterTopicProcedures = alterTopicProcedures; } @@ -428,7 +439,7 @@ public List getAlterTopicProcedures() { } @TestOnly - public void setDropPipeProcedures(List dropPipeProcedures) { + public void setDropPipeProcedures(final List dropPipeProcedures) { this.dropPipeProcedures = dropPipeProcedures; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java index 37c04e26f5085..e50f1edcc6919 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java @@ -238,16 +238,22 @@ private void closeSchemaRegionListeningQueueIfNecessary( } @Override - protected void dropPipe(final String pipeName, final long creationTime) { - super.dropPipe(pipeName, creationTime); + protected boolean dropPipe(final String pipeName, final long creationTime) { + if (!super.dropPipe(pipeName, creationTime)) { + return false; + } PipeDataNodeRemainingEventAndTimeMetrics.getInstance() .deregister(pipeName + "_" + creationTime); + + return true; } @Override - protected void dropPipe(final String pipeName) { - super.dropPipe(pipeName); + protected boolean dropPipe(final String pipeName) { + if (!super.dropPipe(pipeName)) { + return false; + } final PipeMeta pipeMeta = pipeMetaKeeper.getPipeMeta(pipeName); if (Objects.nonNull(pipeMeta)) { @@ -255,6 +261,8 @@ protected void dropPipe(final String pipeName) { PipeDataNodeRemainingEventAndTimeMetrics.getInstance() .deregister(pipeName + "_" + creationTime); } + + return true; } public void stopAllPipesWithCriticalException() { @@ -612,7 +620,7 @@ public Set getPipeTaskRegionIdSet(final String pipeName, final long cre ///////////////////////// Pipe Consensus ///////////////////////// - public ProgressIndex getPipeTaskProgressIndex(String pipeName, int consensusGroupId) { + public ProgressIndex getPipeTaskProgressIndex(final String pipeName, final int consensusGroupId) { if (!tryReadLockWithTimeOut(10)) { throw new PipeException( String.format( 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 5284d295863d6..983abb0364269 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 @@ -114,6 +114,13 @@ private void handleSingleConsumerGroupMetaChangesInternal( return; } + // unbind and remove prefetching queue + final Set topicsUnsubByGroup = + ConsumerGroupMeta.getTopicsUnsubByGroup(metaInAgent, metaFromCoordinator); + for (final String topicName : topicsUnsubByGroup) { + SubscriptionAgent.broker().unbindPrefetchingQueue(consumerGroupId, topicName, true); + } + // TODO: Currently we fully replace the entire ConsumerGroupMeta without carefully checking the // changes in its fields. consumerGroupMetaKeeper.removeConsumerGroupMeta(consumerGroupId); 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 45145bd2b67e0..bc1741704224f 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 @@ -149,4 +149,16 @@ public String getTopicFormat(final String topicName) { releaseReadLock(); } } + + public String getTopicMode(final String topicName) { + acquireReadLock(); + try { + return topicMetaKeeper + .getTopicMeta(topicName) + .getConfig() + .getStringOrDefault(TopicConstant.MODE_KEY, TopicConstant.MODE_DEFAULT_VALUE); + } finally { + releaseReadLock(); + } + } } 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 d26ea8b445817..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,14 @@ 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; @@ -181,6 +189,11 @@ 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 9dec709230f5d..d93f64e2f1848 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 @@ -25,11 +25,16 @@ import org.apache.iotdb.db.subscription.event.SubscriptionEvent; import org.apache.iotdb.db.subscription.event.SubscriptionEventBinaryCache; import org.apache.iotdb.pipe.api.event.Event; +import org.apache.iotdb.rpc.subscription.payload.poll.ErrorPayload; 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.apache.iotdb.rpc.subscription.payload.poll.TerminationPayload; 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; @@ -124,7 +129,7 @@ protected SubscriptionCommitContext generateSubscriptionCommitContext() { subscriptionCommitIdGenerator.getAndIncrement()); } - protected SubscriptionCommitContext generateInvalidSubscriptionCommitContext() { + private SubscriptionCommitContext generateInvalidSubscriptionCommitContext() { return new SubscriptionCommitContext( IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), PipeAgent.runtime().getRebootTimes(), @@ -176,4 +181,23 @@ public boolean isCompleted() { public void markCompleted() { isCompleted = true; } + + public SubscriptionEvent generateSubscriptionPollTerminationResponse() { + return new SubscriptionEvent( + Collections.emptyList(), + new SubscriptionPollResponse( + SubscriptionPollResponseType.TERMINATION.getType(), + new TerminationPayload(), + generateInvalidSubscriptionCommitContext())); + } + + public SubscriptionEvent generateSubscriptionPollErrorResponse( + final String errorMessage, final boolean critical) { + return new SubscriptionEvent( + Collections.emptyList(), + new SubscriptionPollResponse( + SubscriptionPollResponseType.ERROR.getType(), + new ErrorPayload(errorMessage, critical), + generateInvalidSubscriptionCommitContext())); + } } 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 14d3315e1b79d..935494537b3c3 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; @@ -138,6 +139,17 @@ private void prefetchOnce() { continue; } + if (event instanceof PipeTerminateEvent) { + LOGGER.info( + "Subscription: SubscriptionPrefetchingTabletsQueue {} commit PipeTerminateEvent {}", + this, + event); + // commit directly + ((PipeTerminateEvent) event) + .decreaseReferenceCount(SubscriptionPrefetchingTsFileQueue.class.getName(), true); + continue; + } + if (event instanceof TabletInsertionEvent) { final List currentTablets = convertToTablets((TabletInsertionEvent) event); if (currentTablets.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 3f2dac42db6c0..dbb6880968113 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,12 +22,13 @@ 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.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.pipe.api.event.dml.insertion.TabletInsertionEvent; -import org.apache.iotdb.rpc.subscription.payload.poll.ErrorPayload; import org.apache.iotdb.rpc.subscription.payload.poll.FileInitPayload; import org.apache.iotdb.rpc.subscription.payload.poll.FilePiecePayload; import org.apache.iotdb.rpc.subscription.payload.poll.FileSealPayload; @@ -42,7 +43,6 @@ import org.slf4j.LoggerFactory; import java.io.IOException; -import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Objects; @@ -80,7 +80,7 @@ public void cleanup() { } @Override - public SubscriptionTsFileEvent poll(final String consumerId) { + public SubscriptionEvent poll(final String consumerId) { if (hasUnPollableOnTheFlySubscriptionTsFileEvent(consumerId)) { return null; } @@ -102,6 +102,17 @@ public SubscriptionTsFileEvent poll(final String consumerId) { continue; } + if (event instanceof PipeTerminateEvent) { + LOGGER.info( + "Subscription: SubscriptionPrefetchingTsFileQueue {} commit PipeTerminateEvent {}", + this, + event); + // commit directly + ((PipeTerminateEvent) event) + .decreaseReferenceCount(SubscriptionPrefetchingTsFileQueue.class.getName(), true); + continue; + } + if (event instanceof TabletInsertionEvent) { final String errorMessage = String.format( @@ -136,7 +147,7 @@ public SubscriptionTsFileEvent poll(final String consumerId) { return null; } - public synchronized @NonNull SubscriptionTsFileEvent pollTsFile( + public synchronized @NonNull SubscriptionEvent pollTsFile( final String consumerId, final String fileName, final long writingOffset) { // 1. Extract current event and check it final SubscriptionTsFileEvent event = consumerIdToCurrentEventMap.get(consumerId); @@ -281,7 +292,7 @@ public SubscriptionTsFileEvent poll(final String consumerId) { return pollTsFile(consumerId, writingOffset, event); } - private synchronized @NonNull SubscriptionTsFileEvent pollTsFile( + private synchronized @NonNull SubscriptionEvent pollTsFile( final String consumerId, final long writingOffset, final SubscriptionTsFileEvent event) { Pair newEventWithCommittable = event.matchOrResetNext(writingOffset); @@ -384,18 +395,8 @@ private synchronized SubscriptionTsFileEvent getPollableOnTheFlySubscriptionTsFi return null; } - private SubscriptionTsFileEvent generateSubscriptionPollErrorResponse( - final String errorMessage, final boolean critical) { - return new SubscriptionTsFileEvent( - Collections.emptyList(), - new SubscriptionPollResponse( - SubscriptionPollResponseType.ERROR.getType(), - new ErrorPayload(errorMessage, critical), - super.generateInvalidSubscriptionCommitContext())); - } - - private SubscriptionTsFileEvent generateSubscriptionPollErrorResponse(final String errorMessage) { + private SubscriptionEvent generateSubscriptionPollErrorResponse(final String errorMessage) { // consider non-critical by default, meaning the client can retry - return generateSubscriptionPollErrorResponse(errorMessage, false); + return super.generateSubscriptionPollErrorResponse(errorMessage, false); } } 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 43827ab85c42e..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 @@ -98,8 +98,11 @@ 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, true); + SubscriptionAgent.broker().unbindPrefetchingQueue(consumerGroupId, topicName, false); } } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java index a1844bef5722e..65c1a95aaa316 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java @@ -103,7 +103,7 @@ protected void acquireWriteLock() { pipeMetaKeeper.acquireWriteLock(); } - protected boolean tryWriteLockWithTimeOut(long timeOutInSeconds) { + protected boolean tryWriteLockWithTimeOut(final long timeOutInSeconds) { try { return pipeMetaKeeper.tryWriteLock(timeOutInSeconds); } catch (final InterruptedException e) { @@ -462,11 +462,14 @@ private boolean createPipe(final PipeMeta pipeMetaFromCoordinator) throws Illega protected abstract Map buildPipeTasks(final PipeMeta pipeMetaFromCoordinator) throws IllegalPathException; - protected void dropPipe(final String pipeName, final long creationTime) { + /** + * @return {@code true} if a pipe has indeed been dropped, otherwise {@code false}. + */ + protected boolean dropPipe(final String pipeName, final long creationTime) { final PipeMeta existedPipeMeta = pipeMetaKeeper.getPipeMeta(pipeName); if (!checkBeforeDropPipe(existedPipeMeta, pipeName, creationTime)) { - return; + return false; } // Mark pipe meta as dropped first. This will help us detect if the pipe meta has been dropped @@ -483,7 +486,7 @@ protected void dropPipe(final String pipeName, final long creationTime) { + "Skip dropping.", pipeName, creationTime); - return; + return false; } // Trigger drop() method for each pipe task by parallel stream @@ -496,13 +499,18 @@ protected void dropPipe(final String pipeName, final long creationTime) { // Remove pipe meta from pipe meta keeper pipeMetaKeeper.removePipeMeta(pipeName); + + return true; } - protected void dropPipe(final String pipeName) { + /** + * @return {@code true} if a pipe has indeed been dropped, otherwise {@code false}. + */ + protected boolean dropPipe(final String pipeName) { final PipeMeta existedPipeMeta = pipeMetaKeeper.getPipeMeta(pipeName); if (!checkBeforeDropPipe(existedPipeMeta, pipeName)) { - return; + return false; } // Mark pipe meta as dropped first. This will help us detect if the pipe meta has been dropped @@ -516,7 +524,7 @@ protected void dropPipe(final String pipeName) { if (pipeTasks == null) { LOGGER.info( "Pipe {} has already been dropped or has not been created. Skip dropping.", pipeName); - return; + return false; } // Trigger drop() method for each pipe task by parallel stream @@ -529,6 +537,8 @@ protected void dropPipe(final String pipeName) { // Remove pipe meta from pipe meta keeper pipeMetaKeeper.removePipeMeta(pipeName); + + return true; } private void startPipe(final String pipeName, final long creationTime) { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java index d53868208bf5e..764fbfca50121 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java @@ -48,7 +48,7 @@ public ConsumerGroupMeta() { } public ConsumerGroupMeta( - String consumerGroupId, long creationTime, ConsumerMeta firstConsumerMeta) { + final String consumerGroupId, final long creationTime, final ConsumerMeta firstConsumerMeta) { this.consumerGroupId = consumerGroupId; this.creationTime = creationTime; this.topicNameToSubscribedConsumerIdSet = new HashMap<>(); @@ -74,20 +74,48 @@ public long getCreationTime() { return creationTime; } + public static /* @NonNull */ Set getTopicsUnsubByGroup( + final ConsumerGroupMeta currentMeta, final ConsumerGroupMeta updatedMeta) { + if (!Objects.equals(currentMeta.consumerGroupId, updatedMeta.consumerGroupId)) { + return Collections.emptySet(); + } + if (!Objects.equals(currentMeta.creationTime, updatedMeta.creationTime)) { + return Collections.emptySet(); + } + + // no need to check consumerIdToConsumerMeta here to avoid potential inconsistent meta + + final Set unsubscribedTopicNames = new HashSet<>(); + currentMeta + .topicNameToSubscribedConsumerIdSet + .keySet() + .forEach( + (topicName) -> { + if (!updatedMeta.topicNameToSubscribedConsumerIdSet.containsKey(topicName)) { + unsubscribedTopicNames.add(topicName); + } + }); + return unsubscribedTopicNames; + } + /////////////////////////////// consumer /////////////////////////////// - public void addConsumer(ConsumerMeta consumerMeta) { + public void addConsumer(final ConsumerMeta consumerMeta) { consumerIdToConsumerMeta.put(consumerMeta.getConsumerId(), consumerMeta); } - public void removeConsumer(String consumerId) { + public void removeConsumer(final String consumerId) { consumerIdToConsumerMeta.remove(consumerId); - for (Set subscribedConsumers : topicNameToSubscribedConsumerIdSet.values()) { - subscribedConsumers.remove(consumerId); + for (final Map.Entry> entry : + topicNameToSubscribedConsumerIdSet.entrySet()) { + entry.getValue().remove(consumerId); + if (entry.getValue().isEmpty()) { + topicNameToSubscribedConsumerIdSet.remove(entry.getKey()); + } } } - public boolean containsConsumer(String consumerId) { + public boolean containsConsumer(final String consumerId) { return consumerIdToConsumerMeta.containsKey(consumerId); } @@ -106,13 +134,13 @@ public boolean isEmpty() { * @return The set of consumer IDs subscribing the given topic in this group. If no consumer is * subscribing the topic, return an empty set. */ - public Set getConsumersSubscribingTopic(String topic) { + public Set getConsumersSubscribingTopic(final String topic) { return topicNameToSubscribedConsumerIdSet.getOrDefault(topic, Collections.emptySet()); } - public Set getTopicsSubscribedByConsumer(String consumerId) { - Set topics = new HashSet<>(); - for (Map.Entry> topicNameToSubscribedConsumerId : + public Set getTopicsSubscribedByConsumer(final String consumerId) { + final Set topics = new HashSet<>(); + for (final Map.Entry> topicNameToSubscribedConsumerId : topicNameToSubscribedConsumerIdSet.entrySet()) { if (topicNameToSubscribedConsumerId.getValue().contains(consumerId)) { topics.add(topicNameToSubscribedConsumerId.getKey()); @@ -121,7 +149,7 @@ public Set getTopicsSubscribedByConsumer(String consumerId) { return topics; } - public void addSubscription(String consumerId, Set topics) { + public void addSubscription(final String consumerId, final Set topics) { if (!consumerIdToConsumerMeta.containsKey(consumerId)) { throw new SubscriptionException( String.format( @@ -129,7 +157,7 @@ public void addSubscription(String consumerId, Set topics) { consumerId, consumerGroupId)); } - for (String topic : topics) { + for (final String topic : topics) { topicNameToSubscribedConsumerIdSet .computeIfAbsent(topic, k -> new HashSet<>()) .add(consumerId); @@ -139,7 +167,7 @@ public void addSubscription(String consumerId, Set topics) { /** * @return topics subscribed by no consumers in this group after this removal. */ - public Set removeSubscription(String consumerId, Set topics) { + public Set removeSubscription(final String consumerId, final Set topics) { if (!consumerIdToConsumerMeta.containsKey(consumerId)) { throw new SubscriptionException( String.format( @@ -147,8 +175,8 @@ public Set removeSubscription(String consumerId, Set topics) { consumerId, consumerGroupId)); } - Set noSubscriptionTopicAfterRemoval = new HashSet<>(); - for (String topic : topics) { + final Set noSubscriptionTopicAfterRemoval = new HashSet<>(); + for (final String topic : topics) { if (topicNameToSubscribedConsumerIdSet.containsKey(topic)) { topicNameToSubscribedConsumerIdSet.get(topic).remove(consumerId); if (topicNameToSubscribedConsumerIdSet.get(topic).isEmpty()) { @@ -163,33 +191,34 @@ public Set removeSubscription(String consumerId, Set topics) { /////////////////////////////// de/ser /////////////////////////////// public ByteBuffer serialize() throws IOException { - PublicBAOS byteArrayOutputStream = new PublicBAOS(); - DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream); + final PublicBAOS byteArrayOutputStream = new PublicBAOS(); + final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream); serialize(outputStream); return ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size()); } - public void serialize(OutputStream outputStream) throws IOException { + public void serialize(final OutputStream outputStream) throws IOException { ReadWriteIOUtils.write(consumerGroupId, outputStream); ReadWriteIOUtils.write(creationTime, outputStream); ReadWriteIOUtils.write(topicNameToSubscribedConsumerIdSet.size(), outputStream); - for (Map.Entry> entry : topicNameToSubscribedConsumerIdSet.entrySet()) { + for (final Map.Entry> entry : + topicNameToSubscribedConsumerIdSet.entrySet()) { ReadWriteIOUtils.write(entry.getKey(), outputStream); ReadWriteIOUtils.write(entry.getValue().size(), outputStream); - for (String id : entry.getValue()) { + for (final String id : entry.getValue()) { ReadWriteIOUtils.write(id, outputStream); } } ReadWriteIOUtils.write(consumerIdToConsumerMeta.size(), outputStream); - for (Map.Entry entry : consumerIdToConsumerMeta.entrySet()) { + for (final Map.Entry entry : consumerIdToConsumerMeta.entrySet()) { ReadWriteIOUtils.write(entry.getKey(), outputStream); entry.getValue().serialize(outputStream); } } - public static ConsumerGroupMeta deserialize(InputStream inputStream) throws IOException { + public static ConsumerGroupMeta deserialize(final InputStream inputStream) throws IOException { final ConsumerGroupMeta consumerGroupMeta = new ConsumerGroupMeta(); consumerGroupMeta.consumerGroupId = ReadWriteIOUtils.readString(inputStream); @@ -220,7 +249,7 @@ public static ConsumerGroupMeta deserialize(InputStream inputStream) throws IOEx return consumerGroupMeta; } - public static ConsumerGroupMeta deserialize(ByteBuffer byteBuffer) { + public static ConsumerGroupMeta deserialize(final ByteBuffer byteBuffer) { final ConsumerGroupMeta consumerGroupMeta = new ConsumerGroupMeta(); consumerGroupMeta.consumerGroupId = ReadWriteIOUtils.readString(byteBuffer); @@ -254,14 +283,14 @@ public static ConsumerGroupMeta deserialize(ByteBuffer byteBuffer) { /////////////////////////////// Object /////////////////////////////// @Override - public boolean equals(Object obj) { + public boolean equals(final Object obj) { if (this == obj) { return true; } if (obj == null || getClass() != obj.getClass()) { return false; } - ConsumerGroupMeta that = (ConsumerGroupMeta) obj; + final ConsumerGroupMeta that = (ConsumerGroupMeta) obj; return Objects.equals(consumerGroupId, that.consumerGroupId) && creationTime == that.creationTime && Objects.equals( diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java index a0e80db04e068..70666a0718249 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java @@ -185,6 +185,8 @@ public Map generateExtractorAttributes() { extractorAttributes.putAll(config.getAttributesWithTimeRange(creationTime)); // realtime mode extractorAttributes.putAll(config.getAttributesWithRealtimeMode()); + // source mode + extractorAttributes.putAll(config.getAttributesWithSourceMode()); return extractorAttributes; }