VGalaxies commented on code in PR #12149: URL: https://github.com/apache/iotdb/pull/12149#discussion_r1523236646
########## iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/SubscriptionInfo.java: ########## Review Comment: `SubscriptionInfo.java` can be placed in a separate subscription package. ########## iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java: ########## @@ -0,0 +1,185 @@ +/* + * 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.confignode.procedure.impl.subscription.subscription; + +import org.apache.iotdb.commons.subscription.meta.ConsumerGroupMeta; +import org.apache.iotdb.commons.subscription.meta.TopicMeta; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; +import org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2; +import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation; +import org.apache.iotdb.confignode.procedure.impl.pipe.task.StopPipeProcedureV2; +import org.apache.iotdb.confignode.procedure.impl.subscription.AbstractOperateSubscriptionProcedure; +import org.apache.iotdb.confignode.procedure.impl.subscription.consumer.AlterConsumerGroupProcedure; +import org.apache.iotdb.confignode.procedure.impl.subscription.topic.AlterTopicProcedure; +import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq; +import org.apache.iotdb.pipe.api.exception.PipeException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; + +public class DropSubscriptionProcedure extends AbstractOperateSubscriptionProcedure { + private static final Logger LOGGER = LoggerFactory.getLogger(DropSubscriptionProcedure.class); + + private TUnsubscribeReq unsubscribeReq; + private AlterConsumerGroupProcedure consumerGroupProcedure; + private List<AlterTopicProcedure> topicProcedures = new ArrayList<>(); + private List<AbstractOperatePipeProcedureV2> pipeProcedures = new ArrayList<>(); + + public DropSubscriptionProcedure() { + super(); + } + + public DropSubscriptionProcedure(TUnsubscribeReq unsubscribeReq) { + this.unsubscribeReq = unsubscribeReq; + } + + @Override + protected PipeTaskOperation getOperation() { + return PipeTaskOperation.DROP_SUBSCRIPTION; + } + + @Override + protected void executeFromValidate(ConfigNodeProcedureEnv env) throws PipeException { + LOGGER.info("DropSubscriptionProcedure: executeFromValidate"); + + // check if the consumer and all topics exists + try { + subscriptionInfo.get().validateBeforeUnsubscribe(unsubscribeReq); + } catch (PipeException e) { + LOGGER.error( + "DropSubscriptionProcedure: executeFromValidate, validateBeforeUnsubscribe failed", e); + throw e; + } + + // Construct AlterConsumerGroupProcedure + ConsumerGroupMeta updatedConsumerGroupMeta = + subscriptionInfo.get().getConsumerGroupMeta(unsubscribeReq.getConsumerGroupId()).copy(); + + // Get topics subscribed by no consumers in this group after this un-subscription + Set<String> topicsUnsubByGroup = + updatedConsumerGroupMeta.removeSubscription( + unsubscribeReq.getConsumerId(), unsubscribeReq.getTopicNames()); + consumerGroupProcedure = new AlterConsumerGroupProcedure(updatedConsumerGroupMeta); + + for (String topic : unsubscribeReq.getTopicNames()) { + if (topicsUnsubByGroup.contains(topic)) { + // Topic will be subscribed by no consumers in this group + + TopicMeta updatedTopicMeta = subscriptionInfo.get().getTopicMeta(topic).copy(); + updatedTopicMeta.removeSubscribedConsumerGroup(unsubscribeReq.getConsumerGroupId()); + + topicProcedures.add(new AlterTopicProcedure(updatedTopicMeta)); + pipeProcedures.add( + new StopPipeProcedureV2(topic + "_" + unsubscribeReq.getConsumerGroupId())); Review Comment: we are currently not considering implementing a TTL mechanism for subscriptions... maybe drop the pipe directly when the topic will be subscribed to by no consumers in this group, otherwise when resubscribing later, a pipe with the same name will be created. ########## iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/SubscriptionInfo.java: ########## @@ -0,0 +1,498 @@ +/* + * 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.confignode.persistence.pipe; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.snapshot.SnapshotProcessor; +import org.apache.iotdb.commons.subscription.meta.ConsumerGroupMeta; +import org.apache.iotdb.commons.subscription.meta.ConsumerGroupMetaKeeper; +import org.apache.iotdb.commons.subscription.meta.SubscriptionMeta; +import org.apache.iotdb.commons.subscription.meta.TopicMeta; +import org.apache.iotdb.commons.subscription.meta.TopicMetaKeeper; +import org.apache.iotdb.confignode.consensus.request.write.subscription.consumer.AlterConsumerGroupPlan; +import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.AlterTopicPlan; +import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.CreateTopicPlan; +import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.DropTopicPlan; +import org.apache.iotdb.confignode.consensus.response.subscription.SubscriptionTableResp; +import org.apache.iotdb.confignode.consensus.response.subscription.TopicTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TCloseConsumerReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateConsumerReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateTopicReq; +import org.apache.iotdb.confignode.rpc.thrift.TSubscribeReq; +import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq; +import org.apache.iotdb.consensus.common.DataSet; +import org.apache.iotdb.pipe.api.exception.PipeException; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.stream.Collectors; +import java.util.stream.StreamSupport; + +public class SubscriptionInfo implements SnapshotProcessor { + + private static final Logger LOGGER = LoggerFactory.getLogger(SubscriptionInfo.class); + + private static final String SNAPSHOT_FILE_NAME = "subscription_info.bin"; + + private final TopicMetaKeeper topicMetaKeeper; + private final ConsumerGroupMetaKeeper consumerGroupMetaKeeper; + private final ReentrantReadWriteLock subscriptionInfoLock = new ReentrantReadWriteLock(true); + + public SubscriptionInfo() { + this.topicMetaKeeper = new TopicMetaKeeper(); + this.consumerGroupMetaKeeper = new ConsumerGroupMetaKeeper(); + } + + /////////////////////////////// Lock /////////////////////////////// + + private void acquireReadLock() { + subscriptionInfoLock.readLock().lock(); + } + + private void releaseReadLock() { + subscriptionInfoLock.readLock().unlock(); + } + + public void acquireWriteLock() { + subscriptionInfoLock.writeLock().lock(); + } + + public void releaseWriteLock() { + subscriptionInfoLock.writeLock().unlock(); + } + + /////////////////////////////// Topic /////////////////////////////// + public void validateBeforeCreatingTopic(TCreateTopicReq createTopicReq) throws PipeException { + acquireReadLock(); + try { + checkBeforeCreateTopicInternal(createTopicReq); + } finally { + releaseReadLock(); + } + } + + private void checkBeforeCreateTopicInternal(TCreateTopicReq createTopicReq) throws PipeException { + if (!isTopicExisted(createTopicReq.getTopicName())) { + return; + } + + final String exceptionMessage = + String.format( + "Failed to create pipe topic %s, the topic with the same name has been created", + createTopicReq.getTopicName()); + LOGGER.warn(exceptionMessage); + throw new PipeException(exceptionMessage); + } + + public void validateBeforeDroppingTopic(String topicName) throws PipeException { + acquireReadLock(); + try { + checkBeforeDropTopicInternal(topicName); + } finally { + releaseReadLock(); + } + } + + private void checkBeforeDropTopicInternal(String topicName) throws PipeException { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "Check before dropping topic: {}, topic exists:{}", topicName, isTopicExisted(topicName)); + } + + // No matter whether the topic exists, we allow the drop operation executed on all nodes to + // ensure the consistency. + // DO NOTHING HERE! + } + + public void validateBeforeAlteringTopic(TopicMeta topicMeta) throws PipeException { + acquireReadLock(); + try { + checkBeforeAlteringTopicInternal(topicMeta); + } finally { + releaseReadLock(); + } + } + + private void checkBeforeAlteringTopicInternal(TopicMeta topicMeta) throws PipeException { + if (isTopicExisted(topicMeta.getTopicName())) { + return; + } + + final String exceptionMessage = + String.format( + "Failed to alter pipe topic %s, the pipe topic with the same name is not existed", + topicMeta.getTopicName()); + LOGGER.warn(exceptionMessage); + throw new PipeException(exceptionMessage); + } + + public boolean isTopicExisted(String topicName) { + acquireReadLock(); + try { + return topicMetaKeeper.containsTopicMeta(topicName); + } finally { + releaseReadLock(); + } + } + + public TopicMeta getTopicMeta(String topicName) { + acquireReadLock(); + try { + return topicMetaKeeper.getTopicMeta(topicName); + } finally { + releaseReadLock(); + } + } + + private Iterable<TopicMeta> getAllTopicMeta() { + return topicMetaKeeper.getAllTopicMeta(); + } + + public DataSet showTopics() { + acquireReadLock(); + try { + return new TopicTableResp( + new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()), + StreamSupport.stream(getAllTopicMeta().spliterator(), false) + .collect(Collectors.toList())); + } finally { + releaseReadLock(); + } + } + + public TSStatus createTopic(CreateTopicPlan plan) { + acquireWriteLock(); + try { + topicMetaKeeper.addTopicMeta(plan.getTopicMeta().getTopicName(), plan.getTopicMeta()); + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + } finally { + releaseWriteLock(); + } + } + + public TSStatus alterTopic(AlterTopicPlan plan) { + acquireWriteLock(); + try { + topicMetaKeeper.removeTopicMeta(plan.getTopicMeta().getTopicName()); + topicMetaKeeper.addTopicMeta(plan.getTopicMeta().getTopicName(), plan.getTopicMeta()); + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + } finally { + releaseWriteLock(); + } + } + + public TSStatus dropTopic(DropTopicPlan plan) { + acquireWriteLock(); + try { + topicMetaKeeper.removeTopicMeta(plan.getTopicName()); + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + } finally { + releaseWriteLock(); + } + } + + public TopicMeta getTopicMetaByTopicName(String topicName) { + acquireReadLock(); + try { + return topicMetaKeeper.getTopicMeta(topicName); + } finally { + releaseReadLock(); + } + } + + ///////////////////////////////// Consumer ///////////////////////////////// + + public void validateBeforeCreatingConsumer(TCreateConsumerReq createConsumerReq) + throws PipeException { + acquireReadLock(); + try { + checkBeforeCreateConsumerInternal(createConsumerReq); + } finally { + releaseReadLock(); + } + } + + private void checkBeforeCreateConsumerInternal(TCreateConsumerReq createConsumerReq) + throws PipeException { + if (!isConsumerGroupExisted(createConsumerReq.getConsumerGroupId()) + || !isConsumerExisted( + createConsumerReq.getConsumerGroupId(), createConsumerReq.getConsumerId())) { + return; + } + + // A consumer with same consumerId and consumerGroupId has already existed, we should end the + // procedure + final String exceptionMessage = + String.format( + "Failed to create pipe consumer %s in consumer group %s, the consumer with the same name has been created", + createConsumerReq.getConsumerId(), createConsumerReq.getConsumerGroupId()); + LOGGER.warn(exceptionMessage); + throw new PipeException(exceptionMessage); + } + + public void validateBeforeDroppingConsumer(TCloseConsumerReq dropConsumerReq) + throws PipeException { + acquireReadLock(); + try { + checkBeforeDropConsumerInternal(dropConsumerReq); + } finally { + releaseReadLock(); + } + } + + private void checkBeforeDropConsumerInternal(TCloseConsumerReq dropConsumerReq) + throws PipeException { + if (isConsumerExisted(dropConsumerReq.getConsumerGroupId(), dropConsumerReq.getConsumerId())) { + return; + } + + // There is no consumer with the same consumerId and consumerGroupId, we should end the + // procedure + final String exceptionMessage = + String.format( + "Failed to drop pipe consumer %s in consumer group %s, the consumer with the same name does not exist", + dropConsumerReq.getConsumerId(), dropConsumerReq.getConsumerGroupId()); + LOGGER.warn(exceptionMessage); + throw new PipeException(exceptionMessage); + } + + public void validateBeforeAlterConsumerGroup(ConsumerGroupMeta consumerGroupMeta) + throws PipeException { + acquireReadLock(); + try { + checkBeforeAlterConsumerGroupInternal(consumerGroupMeta); + } finally { + releaseReadLock(); + } + } + + public void checkBeforeAlterConsumerGroupInternal(ConsumerGroupMeta consumerGroupMeta) + throws PipeException { + if (!isConsumerGroupExisted(consumerGroupMeta.getConsumerGroupId())) { + // Consumer group not exist, not allowed to alter + final String exceptionMessage = + String.format( + "Failed to alter consumer group because the consumer group %s does not exist", + consumerGroupMeta.getConsumerGroupId()); + LOGGER.warn(exceptionMessage); + throw new PipeException(exceptionMessage); + } + } + + public boolean isConsumerGroupExisted(String consumerGroupName) { + acquireReadLock(); + try { + return consumerGroupMetaKeeper.containsConsumerGroupMeta(consumerGroupName); + } finally { + releaseReadLock(); + } + } + + public boolean isConsumerExisted(String consumerGroupName, String consumerId) { + acquireReadLock(); + try { + ConsumerGroupMeta consumerGroupMeta = + consumerGroupMetaKeeper.getConsumerGroupMeta(consumerGroupName); + return consumerGroupMeta == null || consumerGroupMeta.containsConsumer(consumerId); + } finally { + releaseReadLock(); + } + } + + public ConsumerGroupMeta getConsumerGroupMeta(String consumerGroupName) { + acquireReadLock(); + try { + return consumerGroupMetaKeeper.getConsumerGroupMeta(consumerGroupName); + } finally { + releaseReadLock(); + } + } + + public TSStatus alterConsumerGroup(AlterConsumerGroupPlan plan) { + acquireWriteLock(); + try { + consumerGroupMetaKeeper.removeConsumerGroupMeta( + plan.getConsumerGroupMeta().getConsumerGroupId()); + if (plan.getConsumerGroupMeta() != null) { Review Comment: what if `plan.getConsumerGroupMeta()` is null? -> NPE in line 342 ########## iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/AbstractOperateSubscriptionProcedure.java: ########## @@ -0,0 +1,267 @@ +/* + * 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.confignode.procedure.impl.subscription; + +import org.apache.iotdb.confignode.persistence.pipe.SubscriptionInfo; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; +import org.apache.iotdb.confignode.procedure.exception.ProcedureException; +import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedException; +import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException; +import org.apache.iotdb.confignode.procedure.impl.node.AbstractNodeProcedure; +import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation; +import org.apache.iotdb.confignode.procedure.state.ProcedureLockState; +import org.apache.iotdb.confignode.procedure.state.subscription.OperateSubscriptionState; +import org.apache.iotdb.pipe.api.exception.PipeException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +public abstract class AbstractOperateSubscriptionProcedure + extends AbstractNodeProcedure<OperateSubscriptionState> { + + private static final Logger LOGGER = + LoggerFactory.getLogger(AbstractOperateSubscriptionProcedure.class); + protected AtomicReference<SubscriptionInfo> subscriptionInfo; + private static final int RETRY_THRESHOLD = 1; + + @Override + protected ProcedureLockState acquireLock(ConfigNodeProcedureEnv configNodeProcedureEnv) { + LOGGER.info("ProcedureId {} try to acquire subscription lock.", getProcId()); + subscriptionInfo = + configNodeProcedureEnv + .getConfigManager() + .getSubscriptionManager() + .getSubscriptionCoordinator() + .tryLock(); + if (subscriptionInfo == null) { + LOGGER.warn("ProcedureId {} failed to acquire subscription lock.", getProcId()); + } else { + LOGGER.info("ProcedureId {} acquired subscription lock.", getProcId()); + } + + final ProcedureLockState procedureLockState = super.acquireLock(configNodeProcedureEnv); + switch (procedureLockState) { + case LOCK_ACQUIRED: + if (subscriptionInfo == null) { + LOGGER.warn( + "ProcedureId {}: LOCK_ACQUIRED. The following procedure should not be executed without subscription lock.", + getProcId()); + } else { + LOGGER.info( + "ProcedureId {}: LOCK_ACQUIRED. The following procedure should be executed with subscription lock.", + getProcId()); + } + break; + case LOCK_EVENT_WAIT: + if (subscriptionInfo == null) { + LOGGER.warn( + "ProcedureId {}: LOCK_EVENT_WAIT. Without acquiring subscription lock.", getProcId()); + } else { + LOGGER.info( + "ProcedureId {}: LOCK_EVENT_WAIT. Subscription lock will be released.", getProcId()); + configNodeProcedureEnv + .getConfigManager() + .getSubscriptionManager() + .getSubscriptionCoordinator() + .unlock(); + subscriptionInfo = null; + } + break; + default: + if (subscriptionInfo == null) { + LOGGER.error( + "ProcedureId {}: {}. Invalid lock state. Without acquiring subscription lock.", + getProcId(), + procedureLockState); + } else { + LOGGER.error( + "ProcedureId {}: {}. Invalid lock state. Subscription lock will be released.", + getProcId(), + procedureLockState); + configNodeProcedureEnv + .getConfigManager() + .getSubscriptionManager() + .getSubscriptionCoordinator() + .unlock(); + subscriptionInfo = null; + } + break; + } + return procedureLockState; + } + + @Override + protected void releaseLock(ConfigNodeProcedureEnv configNodeProcedureEnv) { + super.releaseLock(configNodeProcedureEnv); + + if (subscriptionInfo == null) { + LOGGER.warn( + "ProcedureId {} release lock. No need to release subscription lock.", getProcId()); + } else { + LOGGER.info("ProcedureId {} release lock. Subscription lock will be released.", getProcId()); + configNodeProcedureEnv + .getConfigManager() + .getSubscriptionManager() + .getSubscriptionCoordinator() + .unlock(); + subscriptionInfo = null; + } + } + + protected abstract PipeTaskOperation getOperation(); Review Comment: maybe PipeTaskOperation -> SubscriptionOperation? ########## iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/SubscriptionInfo.java: ########## @@ -0,0 +1,498 @@ +/* + * 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.confignode.persistence.pipe; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.snapshot.SnapshotProcessor; +import org.apache.iotdb.commons.subscription.meta.ConsumerGroupMeta; +import org.apache.iotdb.commons.subscription.meta.ConsumerGroupMetaKeeper; +import org.apache.iotdb.commons.subscription.meta.SubscriptionMeta; +import org.apache.iotdb.commons.subscription.meta.TopicMeta; +import org.apache.iotdb.commons.subscription.meta.TopicMetaKeeper; +import org.apache.iotdb.confignode.consensus.request.write.subscription.consumer.AlterConsumerGroupPlan; +import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.AlterTopicPlan; +import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.CreateTopicPlan; +import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.DropTopicPlan; +import org.apache.iotdb.confignode.consensus.response.subscription.SubscriptionTableResp; +import org.apache.iotdb.confignode.consensus.response.subscription.TopicTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TCloseConsumerReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateConsumerReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateTopicReq; +import org.apache.iotdb.confignode.rpc.thrift.TSubscribeReq; +import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq; +import org.apache.iotdb.consensus.common.DataSet; +import org.apache.iotdb.pipe.api.exception.PipeException; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.stream.Collectors; +import java.util.stream.StreamSupport; + +public class SubscriptionInfo implements SnapshotProcessor { + + private static final Logger LOGGER = LoggerFactory.getLogger(SubscriptionInfo.class); + + private static final String SNAPSHOT_FILE_NAME = "subscription_info.bin"; + + private final TopicMetaKeeper topicMetaKeeper; + private final ConsumerGroupMetaKeeper consumerGroupMetaKeeper; + private final ReentrantReadWriteLock subscriptionInfoLock = new ReentrantReadWriteLock(true); + + public SubscriptionInfo() { + this.topicMetaKeeper = new TopicMetaKeeper(); + this.consumerGroupMetaKeeper = new ConsumerGroupMetaKeeper(); + } + + /////////////////////////////// Lock /////////////////////////////// + + private void acquireReadLock() { + subscriptionInfoLock.readLock().lock(); + } + + private void releaseReadLock() { + subscriptionInfoLock.readLock().unlock(); + } + + public void acquireWriteLock() { + subscriptionInfoLock.writeLock().lock(); + } + + public void releaseWriteLock() { + subscriptionInfoLock.writeLock().unlock(); + } + + /////////////////////////////// Topic /////////////////////////////// + public void validateBeforeCreatingTopic(TCreateTopicReq createTopicReq) throws PipeException { + acquireReadLock(); + try { + checkBeforeCreateTopicInternal(createTopicReq); + } finally { + releaseReadLock(); + } + } + + private void checkBeforeCreateTopicInternal(TCreateTopicReq createTopicReq) throws PipeException { + if (!isTopicExisted(createTopicReq.getTopicName())) { + return; + } + + final String exceptionMessage = + String.format( + "Failed to create pipe topic %s, the topic with the same name has been created", + createTopicReq.getTopicName()); + LOGGER.warn(exceptionMessage); + throw new PipeException(exceptionMessage); + } + + public void validateBeforeDroppingTopic(String topicName) throws PipeException { + acquireReadLock(); + try { + checkBeforeDropTopicInternal(topicName); + } finally { + releaseReadLock(); + } + } + + private void checkBeforeDropTopicInternal(String topicName) throws PipeException { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "Check before dropping topic: {}, topic exists:{}", topicName, isTopicExisted(topicName)); + } + + // No matter whether the topic exists, we allow the drop operation executed on all nodes to + // ensure the consistency. + // DO NOTHING HERE! + } Review Comment: need to check whether the topic is subscribed to by any consumer group ########## iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/subscription/SubscriptionTableResp.java: ########## @@ -0,0 +1,78 @@ +/* + * 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.confignode.consensus.response.subscription; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.subscription.meta.SubscriptionMeta; +import org.apache.iotdb.confignode.rpc.thrift.TGetAllSubscriptionInfoResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionInfo; +import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionResp; +import org.apache.iotdb.consensus.common.DataSet; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; + +public class SubscriptionTableResp implements DataSet { + private final TSStatus status; + private final List<SubscriptionMeta> allSubscriptionMeta; + + public SubscriptionTableResp(TSStatus status, List<SubscriptionMeta> allSubscriptionMeta) { + this.status = status; + this.allSubscriptionMeta = allSubscriptionMeta; + } + + public SubscriptionTableResp filter(String topicName) { + if (topicName == null) { + return this; + } else { + final List<SubscriptionMeta> filteredSubscriptionMeta = new ArrayList<>(); + for (SubscriptionMeta subscriptionMeta : allSubscriptionMeta) { + if (subscriptionMeta.getTopicName().equals(topicName)) { + filteredSubscriptionMeta.add(subscriptionMeta); + break; + } + } + return new SubscriptionTableResp(status, filteredSubscriptionMeta); + } + } + + public TShowSubscriptionResp convertToTShowSubscriptionResp() { + final List<TShowSubscriptionInfo> showSubscriptionInfoList = new ArrayList<>(); + + for (SubscriptionMeta subscriptionMeta : allSubscriptionMeta) { + showSubscriptionInfoList.add( + new TShowSubscriptionInfo( + subscriptionMeta.getTopicName(), + subscriptionMeta.getConsumerGroupID(), + subscriptionMeta.getConsumerIDs())); + } + return new TShowSubscriptionResp(status); Review Comment: showSubscriptionInfoList unused? ########## iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedure.java: ########## @@ -0,0 +1,195 @@ +/* + * 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.confignode.procedure.impl.subscription.subscription; + +import org.apache.iotdb.commons.subscription.meta.ConsumerGroupMeta; +import org.apache.iotdb.commons.subscription.meta.TopicMeta; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; +import org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2; +import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation; +import org.apache.iotdb.confignode.procedure.impl.pipe.task.CreatePipeProcedureV2; +import org.apache.iotdb.confignode.procedure.impl.pipe.task.StartPipeProcedureV2; +import org.apache.iotdb.confignode.procedure.impl.subscription.AbstractOperateSubscriptionProcedure; +import org.apache.iotdb.confignode.procedure.impl.subscription.consumer.AlterConsumerGroupProcedure; +import org.apache.iotdb.confignode.procedure.impl.subscription.topic.AlterTopicProcedure; +import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TSubscribeReq; +import org.apache.iotdb.pipe.api.exception.PipeException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +public class CreateSubscriptionProcedure extends AbstractOperateSubscriptionProcedure { + private static final Logger LOGGER = LoggerFactory.getLogger(CreateSubscriptionProcedure.class); + + private TSubscribeReq subscribeReq; + + private AlterConsumerGroupProcedure consumerGroupProcedure; + private List<AlterTopicProcedure> topicProcedures = new ArrayList<>(); + private List<AbstractOperatePipeProcedureV2> pipeProcedures = new ArrayList<>(); + + public CreateSubscriptionProcedure() { + super(); + } + + public CreateSubscriptionProcedure(TSubscribeReq subscribeReq) { + this.subscribeReq = subscribeReq; + } + + @Override + protected PipeTaskOperation getOperation() { + return PipeTaskOperation.CREATE_SUBSCRIPTION; + } + + @Override + protected void executeFromOperateOnConfigNodes(ConfigNodeProcedureEnv env) throws PipeException { + LOGGER.info("CreateSubscriptionProcedure: executeFromOperateOnConfigNodes"); + + int topicCount = topicProcedures.size(); + for (int i = 0; i < topicCount; ++i) { + pipeProcedures.get(i).executeFromValidateTask(env); + } + + for (int i = 0; i < topicCount; ++i) { + pipeProcedures.get(i).executeFromCalculateInfoForTask(env); + } + + consumerGroupProcedure.executeFromOperateOnConfigNodes(env); + + for (int i = 0; i < topicCount; ++i) { + topicProcedures.get(i).executeFromOperateOnConfigNodes(env); + pipeProcedures.get(i).executeFromWriteConfigNodeConsensus(env); + } + } + + @Override + protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws PipeException { + LOGGER.info("CreateSubscriptionProcedure: executeFromOperateOnDataNodes"); + + consumerGroupProcedure.executeFromOperateOnDataNodes(env); + + int topicCount = topicProcedures.size(); + for (int i = 0; i < topicCount; ++i) { + topicProcedures.get(i).executeFromOperateOnDataNodes(env); + try { + pipeProcedures.get(i).executeFromOperateOnDataNodes(env); + } catch (IOException e) { + LOGGER.warn( + "Failed to create or start pipe task for subscription on datanode, because {}", + e.getMessage()); + throw new PipeException(e.getMessage()); + } + } + } + + @Override + protected void executeFromValidate(ConfigNodeProcedureEnv env) throws PipeException { + LOGGER.info("CreateSubscriptionProcedure: executeFromValidate"); + + // check if the consumer and all topics exists + try { + subscriptionInfo.get().validateBeforeSubscribe(subscribeReq); + } catch (PipeException e) { + LOGGER.error( + "CreateSubscriptionProcedure: executeFromValidate, validateBeforeSubscribe failed", e); + throw e; + } + + // Construct AlterConsumerGroupProcedure + ConsumerGroupMeta updatedConsumerGroupMeta = + subscriptionInfo.get().getConsumerGroupMeta(subscribeReq.getConsumerGroupId()).copy(); + updatedConsumerGroupMeta.addSubscription( + subscribeReq.getConsumerId(), subscribeReq.getTopicNames()); + consumerGroupProcedure = new AlterConsumerGroupProcedure(updatedConsumerGroupMeta); + + for (String topic : subscribeReq.getTopicNames()) { + // Construct AlterTopicProcedures + TopicMeta updatedTopicMeta = subscriptionInfo.get().getTopicMeta(topic).copy(); + + // Construct pipe procedures + if (updatedTopicMeta.addSubscribedConsumerGroup(subscribeReq.getConsumerGroupId())) { + // Consumer group not subscribed this topic + // TODO: now all configs are put to extractor attributes. need to put to processor and + // connector attributes correctly. + pipeProcedures.add( + new CreatePipeProcedureV2( + new TCreatePipeReq() + .setPipeName(topic + "_" + subscribeReq.getConsumerGroupId()) + .setExtractorAttributes(updatedTopicMeta.getConfig().getAttribute()))); + } else { + // Consumer group already subscribed this topic + pipeProcedures.add( + new StartPipeProcedureV2(topic + "_" + subscribeReq.getConsumerGroupId())); + } + + topicProcedures.add(new AlterTopicProcedure(updatedTopicMeta)); Review Comment: no need to update topic meta if consumer group already subscribed this topic -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
