You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "VGalaxies (via GitHub)" <gi...@apache.org> on 2024/06/03 12:59:41 UTC
[PR] Subscription: adapt pipe completion signal for automatically drop subscription [iotdb]
VGalaxies opened a new pull request, #12647:
URL: https://github.com/apache/iotdb/pull/12647
WIP...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Subscription: adapt pipe completion signal for automatically drop subscription & improve parsing logic when using JAVA SDK client [iotdb]
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #12647:
URL: https://github.com/apache/iotdb/pull/12647#discussion_r1634145898
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java:
##########
@@ -242,44 +250,47 @@ 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
+ // Rollback AlterConsumerGroupProcedure
+ alterConsumerGroupProcedure.rollbackFromOperateOnConfigNodes(env);
}
@Override
- protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOException {
+ protected void rollbackFromOperateOnDataNodes(final ConfigNodeProcedureEnv env)
Review Comment:
> Usually when we do rollback, we roll back from the last executed to the first executed.
The DropSubscriptionProcedure has been appropriately modified. The CreateSubscriptionProcedure will be modified in subsequent metadata restructuring.
> Whether the drop subscription procedure needs to be rolled back may need to be considered twice.
Before metadata restructuring, considering that the process of changing the topic meta and consumer group meta involves multiple procedures, there could be failures during execution. Therefore, it is currently necessary to consider rolling back part of the procedure to ensure CN metadata consistency.
In the subsequent metadata restructuring, we will consider mapping the drop subscription procedure to a single procedure. At that time we can consider not performing rollback to prevent the occurrence of create subscription after the rollback of drop subscription.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Subscription: adapt pipe completion signal for automatically drop subscription [iotdb]
Posted by "Caideyipi (via GitHub)" <gi...@apache.org>.
Caideyipi commented on code in PR #12647:
URL: https://github.com/apache/iotdb/pull/12647#discussion_r1628640019
##########
integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java:
##########
@@ -543,7 +551,112 @@ 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<SubscriptionMessage> 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);
+ }
+ consumer.unsubscribe(topicName);
Review Comment:
This will work not only in the query mode since the "unsubscribe" is explicitly called...
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java:
##########
@@ -74,6 +74,31 @@ public long getCreationTime() {
return creationTime;
}
+ public static /* @NonNull */ Set<String> 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();
+ }
+ if (!Objects.equals(
+ currentMeta.consumerIdToConsumerMeta, updatedMeta.consumerIdToConsumerMeta)) {
Review Comment:
Consider closing the consumer directly, and the consumerIdToConsumerMeta is altered by the DropConsumerProcedure.... In that case the queue can not be removed.
##########
iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java:
##########
@@ -336,18 +336,23 @@ protected List<SubscriptionMessage> poll(final Set<String> 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);
}
+ case TERMINATION:
+ final SubscriptionCommitContext commitContext = pollResponse.getCommitContext();
+ final String topicNameToUnsubscribe = commitContext.getTopicName();
+ LOGGER.info(
+ "Termination occurred when SubscriptionConsumer {} polling topics {}, unsubscribe topic {} on DN {} automatically",
+ this,
+ topicNames,
+ topicNameToUnsubscribe,
+ commitContext.getDataNodeId());
+ unsubscribe(topicNameToUnsubscribe);
Review Comment:
Note that since the pipe is automatically dropped, the "DropPipeProcedure"s in "DropSubscriptionProcedure" will encounter failure in validation... Need to add some checks or else the procedure will fail.
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java:
##########
@@ -74,6 +74,31 @@ public long getCreationTime() {
return creationTime;
}
+ public static /* @NonNull */ Set<String> 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();
+ }
+ if (!Objects.equals(
+ currentMeta.consumerIdToConsumerMeta, updatedMeta.consumerIdToConsumerMeta)) {
Review Comment:
TODO: Remove "topic" key in ConsumerGroupMeta#removeConsumer.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java:
##########
@@ -162,8 +162,30 @@ public void unbindPrefetchingQueue(final String topicName) {
LOGGER.warn("Subscription: prefetching queue bound to topic [{}] does not exist", topicName);
return;
}
+
// clean up uncommitted events
prefetchingQueue.cleanup();
+ prefetchingQueue.markCompleted();
+
+ // we remove this prefetching queue and deregister metrics when consumers dropping related
+ // subscription...
+ }
+
+ public void removePrefetchingQueue(final String topicName) {
+ final SubscriptionPrefetchingQueue prefetchingQueue =
+ topicNameToPrefetchingQueue.get(topicName);
+ if (Objects.isNull(prefetchingQueue)) {
+ LOGGER.warn("Subscription: prefetching queue bound to topic [{}] does not exist", topicName);
+ return;
+ }
+
+ if (!prefetchingQueue.isCompleted()) {
Review Comment:
It seems that the "completed" is marked when the pipe is dropped. However in the normal scenarios (namely the unsubscriptions is manually triggered by the client) the "drop pipe" is executed after updating consumer group meta. That is, the "isCompleted" may be false when a topic's last subscription is removed.
##########
integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java:
##########
@@ -543,7 +551,112 @@ 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<SubscriptionMessage> 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);
+ }
+ consumer.unsubscribe(topicName);
Review Comment:
And better test the ”TerminationPayload“ as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Subscription: adapt pipe completion signal for automatically drop subscription [iotdb]
Posted by "Caideyipi (via GitHub)" <gi...@apache.org>.
Caideyipi commented on code in PR #12647:
URL: https://github.com/apache/iotdb/pull/12647#discussion_r1628846407
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/exception/SubscriptionTerminationException.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * 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 SubscriptionTerminationException extends SubscriptionRuntimeCriticalException {
+
+ public SubscriptionTerminationException(final String message) {
Review Comment:
Consider whether to use it before merge.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java:
##########
@@ -242,44 +250,47 @@ 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
+ // Rollback AlterConsumerGroupProcedure
+ alterConsumerGroupProcedure.rollbackFromOperateOnConfigNodes(env);
}
@Override
- protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOException {
+ protected void rollbackFromOperateOnDataNodes(final ConfigNodeProcedureEnv env)
Review Comment:
There are 3 points for the rollback function:
1. Usually when we do rollback, we roll back from the last executed to the first executed.
2. We do not rollback DataNode and ConfigNode execution of dropping pipe.
3. Whether the drop subscription procedure needs to be rolled back may need to be considered twice.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Subscription: adapt pipe completion signal for automatically drop subscription [iotdb]
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #12647:
URL: https://github.com/apache/iotdb/pull/12647#discussion_r1628743317
##########
integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java:
##########
@@ -543,7 +551,112 @@ 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<SubscriptionMessage> 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);
+ }
+ consumer.unsubscribe(topicName);
Review Comment:
> This will work not only in the query mode since the "unsubscribe" is explicitly called...
removed this explicit unsubscribe...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Subscription: adapt pipe completion signal for automatically drop subscription [iotdb]
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #12647:
URL: https://github.com/apache/iotdb/pull/12647#discussion_r1628743055
##########
integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java:
##########
@@ -543,7 +551,112 @@ 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<SubscriptionMessage> 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);
+ }
+ consumer.unsubscribe(topicName);
Review Comment:
> This will work not only in the query mode since the "unsubscribe" is explicitly called...
Before the Awaitility test passes, the consumer is always in a poll loop, so it can be confirmed that the drop subscription is triggered due to receiving a TerminationPayload.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Subscription: adapt pipe completion signal for automatically drop subscription [iotdb]
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #12647:
URL: https://github.com/apache/iotdb/pull/12647#discussion_r1628781572
##########
iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java:
##########
@@ -336,18 +336,23 @@ protected List<SubscriptionMessage> poll(final Set<String> 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);
}
+ case TERMINATION:
+ final SubscriptionCommitContext commitContext = pollResponse.getCommitContext();
+ final String topicNameToUnsubscribe = commitContext.getTopicName();
+ LOGGER.info(
+ "Termination occurred when SubscriptionConsumer {} polling topics {}, unsubscribe topic {} on DN {} automatically",
+ this,
+ topicNames,
+ topicNameToUnsubscribe,
+ commitContext.getDataNodeId());
+ unsubscribe(topicNameToUnsubscribe);
Review Comment:
> Okay, checked that there are not.... However there will be a warn log in PipeDataNodeRemainingEventAndTimeMetrics#deregister. Can you remove it incidentally?
added a boolean return value to `dropPipe`...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Subscription: adapt pipe completion signal for automatically drop subscription [iotdb]
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #12647:
URL: https://github.com/apache/iotdb/pull/12647#discussion_r1628756580
##########
iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java:
##########
@@ -336,18 +336,23 @@ protected List<SubscriptionMessage> poll(final Set<String> 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);
}
+ case TERMINATION:
+ final SubscriptionCommitContext commitContext = pollResponse.getCommitContext();
+ final String topicNameToUnsubscribe = commitContext.getTopicName();
+ LOGGER.info(
+ "Termination occurred when SubscriptionConsumer {} polling topics {}, unsubscribe topic {} on DN {} automatically",
+ this,
+ topicNames,
+ topicNameToUnsubscribe,
+ commitContext.getDataNodeId());
+ unsubscribe(topicNameToUnsubscribe);
Review Comment:
In DropPipeProcedureV2, no exception will be occurred even if the pipe does not exist...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Subscription: adapt pipe completion signal for automatically drop subscription [iotdb]
Posted by "Caideyipi (via GitHub)" <gi...@apache.org>.
Caideyipi commented on code in PR #12647:
URL: https://github.com/apache/iotdb/pull/12647#discussion_r1628768990
##########
iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java:
##########
@@ -336,18 +336,23 @@ protected List<SubscriptionMessage> poll(final Set<String> 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);
}
+ case TERMINATION:
+ final SubscriptionCommitContext commitContext = pollResponse.getCommitContext();
+ final String topicNameToUnsubscribe = commitContext.getTopicName();
+ LOGGER.info(
+ "Termination occurred when SubscriptionConsumer {} polling topics {}, unsubscribe topic {} on DN {} automatically",
+ this,
+ topicNames,
+ topicNameToUnsubscribe,
+ commitContext.getDataNodeId());
+ unsubscribe(topicNameToUnsubscribe);
Review Comment:
Okay, checked that there are not.... However there will be a warn log in PipeDataNodeRemainingEventAndTimeMetrics#deregister. Can you remove it incidentally?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Subscription: adapt pipe completion signal for automatically drop subscription [iotdb]
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies closed pull request #12647: Subscription: adapt pipe completion signal for automatically drop subscription
URL: https://github.com/apache/iotdb/pull/12647
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org