[dev/1.3] Disable Subscription and Pipes containing data insertion under Ratis

iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java

@@ -21,6 +21,20 @@
 
 public final class ConfigNodeMessages {
 
+  public static final String
+      MESSAGE_SUBSCRIPTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_194CE431 =
+          "Subscription is disabled when data region consensus protocol is Ratis.";
+
+  public static final String
+      MESSAGE_PIPE_DATA_INSERTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_7D669ECC =
+          "Pipe synchronization with data insertion is disabled when data region consensus protocol is Ratis.";
+
+  public static final String MESSAGE_FAILED_TO_ALTER_PIPE_ARG_FAILURES_ARG_DOES_NOT_EXIST_169BDB85 =
+      "Failed to alter pipe %s. Failures: %s does not exist.";
+
+  public static final String MESSAGE_FAILED_TO_START_PIPE_ARG_FAILURES_ARG_DOES_NOT_EXIST_F13A63EE =
+      "Failed to start pipe %s. Failures: %s does not exist.";
+
   public static final String
       EXCEPTION_CONFIGNODE_SIMPLECONSENSUS_WRITES_ARE_DISABLED_AFTER_A_WAL_CLEANUP_FAILURE_RESTART_IS_REQUIRED_D52F0107 =
           "ConfigNode SimpleConsensus writes are disabled after a WAL cleanup failure; restart is required.";

iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java

@@ -21,6 +21,20 @@
 
 public final class ConfigNodeMessages {
 
+  public static final String
+      MESSAGE_SUBSCRIPTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_194CE431 =
+          "当数据区域共识协议为 Ratis 时，Subscription 功能不可用。";
+
+  public static final String
+      MESSAGE_PIPE_DATA_INSERTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_7D669ECC =
+          "当数据区域共识协议为 Ratis 时，包含 data.insert 的 Pipe 同步不可用。";
+
+  public static final String MESSAGE_FAILED_TO_ALTER_PIPE_ARG_FAILURES_ARG_DOES_NOT_EXIST_169BDB85 =
+      "无法修改 Pipe %s。原因：%s 不存在。";
+
+  public static final String MESSAGE_FAILED_TO_START_PIPE_ARG_FAILURES_ARG_DOES_NOT_EXIST_F13A63EE =
+      "无法启动 Pipe %s。原因：%s 不存在。";
+
   public static final String
       EXCEPTION_CONFIGNODE_SIMPLECONSENSUS_WRITES_ARE_DISABLED_AFTER_A_WAL_CLEANUP_FAILURE_RESTART_IS_REQUIRED_D52F0107 =
           "ConfigNode SimpleConsensus 因 WAL 清理失败已禁用写入，需要重启。";

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java

@@ -98,6 +98,7 @@
 import org.apache.iotdb.confignode.consensus.response.template.TemplateSetInfoResp;
 import org.apache.iotdb.confignode.consensus.response.ttl.ShowTTLResp;
 import org.apache.iotdb.confignode.consensus.statemachine.ConfigRegionStateMachine;
+import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
 import org.apache.iotdb.confignode.manager.consensus.ConsensusManager;
 import org.apache.iotdb.confignode.manager.cq.CQManager;
 import org.apache.iotdb.confignode.manager.load.LoadManager;
@@ -1584,6 +1585,31 @@
         : new TGetJarInListResp(status, Collections.emptyList());
   }
 
+  private TSStatus confirmSubscriptionFeatureAvailability() {
+    final TSStatus status = confirmLeader();
+    if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()
+        || PipeConfigNodeAgent.isSubscriptionFeatureEnabled()) {
+      return status;
+    }
+    return RpcUtils.getStatus(
+        TSStatusCode.UNSUPPORTED_OPERATION,
+        ConfigNodeMessages
+            .MESSAGE_SUBSCRIPTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_194CE431);
+  }
+
+  private TSStatus confirmPipeDataInsertionAvailability(
+      final Map<String, String> sourceAttributes) {
+    final TSStatus status = confirmLeader();
+    if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()
+        || PipeConfigNodeAgent.isPipeDataInsertionEnabled(sourceAttributes)) {
+      return status;
+    }
+    return RpcUtils.getStatus(
+        TSStatusCode.UNSUPPORTED_OPERATION,
+        ConfigNodeMessages
+            .MESSAGE_PIPE_DATA_INSERTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_7D669ECC);
+  }
+
   @Override
   public TSStatus merge() {
     TSStatus status = confirmLeader();
@@ -2154,7 +2180,7 @@
 
   @Override
   public TSStatus createPipe(TCreatePipeReq req) {
-    TSStatus status = confirmLeader();
+    TSStatus status = confirmPipeDataInsertionAvailability(req.getExtractorAttributes());
     return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
         ? pipeManager.getPipeTaskCoordinator().createPipe(req)
         : status;
@@ -2210,7 +2236,7 @@
 
   @Override
   public TSStatus createTopic(TCreateTopicReq req) {
-    TSStatus status = confirmLeader();
+    TSStatus status = confirmSubscriptionFeatureAvailability();
     return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
         ? subscriptionManager.getSubscriptionCoordinator().createTopic(req)
         : status;
@@ -2242,7 +2268,7 @@
 
   @Override
   public TSStatus createConsumer(TCreateConsumerReq req) {
-    TSStatus status = confirmLeader();
+    TSStatus status = confirmSubscriptionFeatureAvailability();
     return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
         ? subscriptionManager.getSubscriptionCoordinator().createConsumer(req)
         : status;
@@ -2258,7 +2284,7 @@
 
   @Override
   public TSStatus createSubscription(TSubscribeReq req) {
-    TSStatus status = confirmLeader();
+    TSStatus status = confirmSubscriptionFeatureAvailability();
     return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
         ? subscriptionManager.getSubscriptionCoordinator().createSubscription(req)
         : status;

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/PipeConfigNodeAgent.java

@@ -19,15 +19,49 @@
 
 package org.apache.iotdb.confignode.manager.pipe.agent;
 
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.pipe.datastructure.options.PipeInclusionOptions;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
 import org.apache.iotdb.confignode.manager.pipe.agent.plugin.PipeConfigNodePluginAgent;
 import org.apache.iotdb.confignode.manager.pipe.agent.receiver.IoTDBConfigNodeReceiverAgent;
 import org.apache.iotdb.confignode.manager.pipe.agent.runtime.PipeConfigNodeRuntimeAgent;
 import org.apache.iotdb.confignode.manager.pipe.agent.task.PipeConfigNodeTaskAgent;
 import org.apache.iotdb.confignode.service.ConfigNode;
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+
+import java.util.Collections;
+import java.util.Map;
 
 /** {@link PipeConfigNodeAgent} is the entry point of the pipe module in {@link ConfigNode}. */
 public class PipeConfigNodeAgent {
 
+  public static boolean isRatisConsensus() {
+    return ConsensusFactory.RATIS_CONSENSUS.equals(
+        ConfigNodeDescriptor.getInstance().getConf().getDataRegionConsensusProtocolClass());
+  }
+
+  public static boolean isSubscriptionFeatureEnabled() {
+    return !isRatisConsensus();
+  }
+
+  public static boolean isPipeDataInsertionEnabled(final Map<String, String> sourceAttributes) {
+    return isPipeDataInsertionEnabled(
+        new PipeParameters(sourceAttributes == null ? Collections.emptyMap() : sourceAttributes));
+  }
+
+  public static boolean isPipeDataInsertionEnabled(final PipeParameters sourceParameters) {
+    if (!isRatisConsensus()) {
+      return true;
+    }
+    try {
+      return !PipeInclusionOptions.isDataInsertionIncluded(sourceParameters);
+    } catch (final IllegalPathException e) {
+      // Let the normal pipe parameter validation report malformed inclusion options.
+      return true;
+    }
+  }
+
   private final PipeConfigNodeTaskAgent pipeConfigNodeTaskAgent;
   private final PipeConfigNodePluginAgent pipeConfigNodePluginAgent;
   private final PipeConfigNodeRuntimeAgent pipeConfigNodeRuntimeAgent;

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgent.java

@@ -66,6 +66,11 @@ protected boolean isShutdown() {
     return PipeConfigNodeAgent.runtime().isShutdown();
   }
 
+  @Override
+  protected boolean isPipeDataInsertionEnabled(final PipeStaticMeta pipeStaticMeta) {
+    return PipeConfigNodeAgent.isPipeDataInsertionEnabled(pipeStaticMeta.getExtractorParameters());
+  }
+
   @Override
   protected void thawRate(final String pipeName, final long creationTime) {
     PipeConfigNodeRemainingTimeMetrics.getInstance().thawRate(pipeName + "_" + creationTime);

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java

@@ -20,9 +20,12 @@
 package org.apache.iotdb.confignode.manager.pipe.coordinator.task;
 
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
 import org.apache.iotdb.confignode.consensus.request.read.pipe.task.ShowPipePlanV2;
 import org.apache.iotdb.confignode.consensus.response.pipe.task.PipeTableResp;
+import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
 import org.apache.iotdb.confignode.manager.ConfigManager;
+import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent;
 import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo;
 import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq;
 import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
@@ -31,6 +34,7 @@
 import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
 import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp;
 import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
 import org.apache.iotdb.rpc.RpcUtils;
 import org.apache.iotdb.rpc.TSStatusCode;
 
@@ -39,6 +43,7 @@
 
 import java.io.IOException;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.concurrent.atomic.AtomicReference;
 
 public class PipeTaskCoordinator {
@@ -93,6 +98,10 @@
 
   /** Caller should ensure that the method is called in the lock {@link #lock()}. */
   public TSStatus createPipe(TCreatePipeReq req) {
+    if (!PipeConfigNodeAgent.isPipeDataInsertionEnabled(req.getExtractorAttributes())) {
+      return pipeDataInsertionDisabledStatus();
+    }
+
     final TSStatus status = configManager.getProcedureManager().createPipe(req);
     if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
       LOGGER.warn("Failed to create pipe {}. Result status: {}.", req.getPipeName(), status);
@@ -108,6 +117,22 @@
           TSStatusCode.PIPE_ERROR,
           String.format("Failed to alter pipe %s, the pipe is being dropped", pipeName));
     }
+    final PipeMeta currentPipeMeta = pipeTaskInfo.getPipeMetaByPipeName(pipeName);
+    if (currentPipeMeta == null) {
+      return req.isSetIfExistsCondition() && req.isIfExistsCondition()
+          ? RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)
+          : RpcUtils.getStatus(
+              TSStatusCode.PIPE_NOT_EXIST_ERROR,
+              String.format(
+                  ConfigNodeMessages
+                      .MESSAGE_FAILED_TO_ALTER_PIPE_ARG_FAILURES_ARG_DOES_NOT_EXIST_169BDB85,
+                  pipeName,
+                  pipeName));
+    }
+    if (!PipeConfigNodeAgent.isPipeDataInsertionEnabled(
+        getUpdatedSourceParameters(req, currentPipeMeta))) {
+      return pipeDataInsertionDisabledStatus();
+    }
     final TSStatus status = configManager.getProcedureManager().alterPipe(req);
     if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
       LOGGER.warn("Failed to alter pipe {}. Result status: {}.", req.getPipeName(), status);
@@ -122,13 +147,53 @@
           TSStatusCode.PIPE_ERROR,
           String.format("Failed to start pipe %s, the pipe is being dropped", pipeName));
     }
+    final PipeMeta pipeMeta = pipeTaskInfo.getPipeMetaByPipeName(pipeName);
+    if (pipeMeta == null) {
+      return RpcUtils.getStatus(
+          TSStatusCode.PIPE_ERROR,
+          String.format(
+              ConfigNodeMessages
+                  .MESSAGE_FAILED_TO_START_PIPE_ARG_FAILURES_ARG_DOES_NOT_EXIST_F13A63EE,
+              pipeName,
+              pipeName));
+    }
+    if (!PipeConfigNodeAgent.isPipeDataInsertionEnabled(
+        pipeMeta.getStaticMeta().getExtractorParameters())) {
+      return pipeDataInsertionDisabledStatus();
+    }
     final TSStatus status = configManager.getProcedureManager().startPipe(pipeName);
     if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
       LOGGER.warn("Failed to start pipe {}. Result status: {}.", pipeName, status);
     }
     return status;
   }
 
+  private PipeParameters getUpdatedSourceParameters(
+      final TAlterPipeReq req, final PipeMeta currentPipeMeta) {
+    if (req.isReplaceAllExtractorAttributes) {
+      return new PipeParameters(
+          req.getExtractorAttributes() == null
+              ? Collections.emptyMap()
+              : req.getExtractorAttributes());
+    }
+
+    final PipeParameters sourceParameters =
+        new PipeParameters(
+            new HashMap<>(currentPipeMeta.getStaticMeta().getExtractorParameters().getAttribute()));
+    if (req.getExtractorAttributes() != null && !req.getExtractorAttributes().isEmpty()) {
+      sourceParameters.addOrReplaceEquivalentAttributes(
+          new PipeParameters(req.getExtractorAttributes()));
+    }
+    return sourceParameters;
+  }
+
+  private TSStatus pipeDataInsertionDisabledStatus() {
+    return RpcUtils.getStatus(
+        TSStatusCode.UNSUPPORTED_OPERATION,
+        ConfigNodeMessages
+            .MESSAGE_PIPE_DATA_INSERTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_7D669ECC);
+  }
+
   /** Caller should ensure that the method is called in the lock {@link #lock()}. */
   public TSStatus stopPipe(String pipeName) {
     if (pipeTaskInfo.isPipeBeingDropped(pipeName)) {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinator.java

@@ -25,6 +25,7 @@
 import org.apache.iotdb.confignode.consensus.response.subscription.SubscriptionTableResp;
 import org.apache.iotdb.confignode.consensus.response.subscription.TopicTableResp;
 import org.apache.iotdb.confignode.manager.ConfigManager;
+import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent;
 import org.apache.iotdb.confignode.manager.pipe.coordinator.task.PipeTaskCoordinatorLock;
 import org.apache.iotdb.confignode.persistence.subscription.SubscriptionInfo;
 import org.apache.iotdb.confignode.rpc.thrift.TCloseConsumerReq;
@@ -116,7 +117,9 @@ public boolean isLocked() {
   /////////////////////////////// Meta sync ///////////////////////////////
 
   public void startSubscriptionMetaSync() {
-    subscriptionMetaSyncer.start();
+    if (PipeConfigNodeAgent.isSubscriptionFeatureEnabled()) {
+      subscriptionMetaSyncer.start();
+    }
   }
 
   public void stopSubscriptionMetaSync() {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java

@@ -50,7 +50,9 @@
 import org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusPlanV2;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusWithStoppedByRuntimeExceptionPlanV2;
 import org.apache.iotdb.confignode.consensus.response.pipe.task.PipeTableResp;
+import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
 import org.apache.iotdb.confignode.i18n.ProcedureMessages;
+import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent;
 import org.apache.iotdb.confignode.manager.pipe.resource.PipeConfigNodeResourceManager;
 import org.apache.iotdb.confignode.procedure.impl.pipe.runtime.PipeHandleMetaChangeProcedure;
 import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq;
@@ -178,6 +180,13 @@
 
   private boolean checkBeforeCreatePipeInternal(final TCreatePipeReq createPipeRequest)
       throws PipeException {
+    checkPipeDataInsertionAvailability(
+        new PipeStaticMeta(
+            createPipeRequest.getPipeName(),
+            System.currentTimeMillis(),
+            createPipeRequest.getExtractorAttributes(),
+            createPipeRequest.getProcessorAttributes(),
+            createPipeRequest.getConnectorAttributes()));
     if (!isPipeExisted(createPipeRequest.getPipeName())) {
       return true;
     }
@@ -283,6 +292,14 @@
       }
     }
 
+    checkPipeDataInsertionAvailability(
+        new PipeStaticMeta(
+            alterPipeRequest.getPipeName(),
+            System.currentTimeMillis(),
+            alterPipeRequest.getExtractorAttributes(),
+            alterPipeRequest.getProcessorAttributes(),
+            alterPipeRequest.getConnectorAttributes()));
+
     return true;
   }
 
@@ -303,6 +320,7 @@
       throw new PipeException(exceptionMessage);
     }
 
+    checkPipeDataInsertionAvailability(getPipeMetaByPipeName(pipeName).getStaticMeta());
     final PipeStatus pipeStatus = getPipeStatus(pipeName);
     if (pipeStatus == PipeStatus.DROPPED) {
       final String exceptionMessage =
@@ -318,6 +336,16 @@
     }
   }
 
+  private void checkPipeDataInsertionAvailability(final PipeStaticMeta staticMeta) {
+    // Procedures call these validators while holding the coordinator lock. Recheck the merged
+    // source parameters here because concurrent ALTER requests can change the RPC-time snapshot.
+    if (!PipeConfigNodeAgent.isPipeDataInsertionEnabled(staticMeta.getExtractorParameters())) {
+      throw new PipeException(
+          ConfigNodeMessages
+              .MESSAGE_PIPE_DATA_INSERTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_7D669ECC);
+    }
+  }
+
   public void checkBeforeStopPipe(final String pipeName) throws PipeException {
     acquireReadLock();
     try {