[Bugfix] Disable Subscription and Pipes containing data insertion under Ratis

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

@@ -51,6 +51,7 @@
 import org.apache.iotdb.commons.enums.RepairDataPartitionTableProgressState;
 import org.apache.iotdb.commons.exception.IllegalPathException;
 import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.i18n.PipeMessages;
 import org.apache.iotdb.commons.path.MeasurementPath;
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.path.PathDeserializeUtil;
@@ -1807,6 +1808,31 @@ public TGetJarInListResp getPipePluginJar(TGetJarInListReq req) {
         : 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,
+        PipeMessages
+            .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,
+        PipeMessages
+            .MESSAGE_PIPE_DATA_INSERTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_7D669ECC);
+  }
+
   @Override
   public TSStatus merge() {
     TSStatus status = confirmLeader();
@@ -2617,7 +2643,7 @@ public TSStatus alterTimeSeriesDataType(final TAlterTimeSeriesReq req) {
 
   @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;
@@ -2673,15 +2699,15 @@ public TGetAllPipeInfoResp getAllPipeInfo() {
 
   @Override
   public TSStatus createTopic(TCreateTopicReq req) {
-    TSStatus status = confirmLeader();
+    TSStatus status = confirmSubscriptionFeatureAvailability();
     return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
         ? subscriptionManager.getSubscriptionCoordinator().createTopic(req)
         : status;
   }
 
   @Override
   public TSStatus alterTopic(TAlterTopicReq req) {
-    TSStatus status = confirmLeader();
+    TSStatus status = confirmSubscriptionFeatureAvailability();
     return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
         ? subscriptionManager.getSubscriptionCoordinator().alterTopic(req)
         : status;
@@ -2713,7 +2739,7 @@ public TGetAllTopicInfoResp getAllTopicInfo() {
 
   @Override
   public TSStatus createConsumer(TCreateConsumerReq req) {
-    TSStatus status = confirmLeader();
+    TSStatus status = confirmSubscriptionFeatureAvailability();
     return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
         ? subscriptionManager.getSubscriptionCoordinator().createConsumer(req)
         : status;
@@ -2729,7 +2755,7 @@ public TSStatus closeConsumer(TCloseConsumerReq req) {
 
   @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

@@ -68,6 +68,11 @@ protected boolean isShutdown() {
     return PipeConfigNodeAgent.runtime().isShutdown();
   }
 
+  @Override
+  protected boolean isPipeDataInsertionEnabled(final PipeStaticMeta pipeStaticMeta) {
+    return PipeConfigNodeAgent.isPipeDataInsertionEnabled(pipeStaticMeta.getSourceParameters());
+  }
+
   @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

@@ -21,6 +21,7 @@
 
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
 import org.apache.iotdb.commons.i18n.PipeMessages;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
 import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
@@ -31,6 +32,7 @@
 import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
 import org.apache.iotdb.confignode.i18n.ManagerMessages;
 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;
@@ -106,6 +108,10 @@ public boolean isLocked() {
 
   /** 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;
     if (req.getPipeName().startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)) {
       status = configManager.getProcedureManager().createConsensusPipe(req);
@@ -235,7 +241,8 @@ public TSStatus alterPipe(TAlterPipeReq req) {
                   .EXCEPTION_FAILED_TO_ALTER_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_919F1E2B,
               pipeName));
     }
-    if (!pipeTaskInfo.isPipeExisted(pipeName, isTableModel)) {
+    final PipeMeta currentPipeMeta = pipeTaskInfo.getPipeMetaByPipeName(pipeName, isTableModel);
+    if (currentPipeMeta == null) {
       return isSetIfExistsCondition
           ? RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)
           : RpcUtils.getStatus(
@@ -246,6 +253,10 @@ public TSStatus alterPipe(TAlterPipeReq req) {
                   pipeName,
                   pipeName));
     }
+    if (!PipeConfigNodeAgent.isPipeDataInsertionEnabled(
+        getUpdatedSourceParameters(req, currentPipeMeta))) {
+      return pipeDataInsertionDisabledStatus();
+    }
     req.setIsTableModel(isTableModel);
     final TSStatus status = configManager.getProcedureManager().alterPipe(req);
     if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
@@ -285,17 +296,48 @@ public TSStatus startPipe(TStartPipeReq req) {
                   .EXCEPTION_FAILED_TO_START_PIPE_ARG_THE_PIPE_IS_BEING_DROPPED_B41F4638,
               pipeName));
     }
-    if (!pipeTaskInfo.isPipeExisted(pipeName, isTableModel)) {
+    final PipeMeta pipeMeta = pipeTaskInfo.getPipeMetaByPipeName(pipeName, isTableModel);
+    if (pipeMeta == null) {
       return RpcUtils.getStatus(
           TSStatusCode.PIPE_NOT_EXIST_ERROR,
           String.format(
               ManagerMessages.MESSAGE_FAILED_TO_START_PIPE_ARG_FAILURES_ARG_DOES_NOT_EXIST_F13A63EE,
               pipeName,
               pipeName));
     }
+    if (!PipeConfigNodeAgent.isPipeDataInsertionEnabled(
+        pipeMeta.getStaticMeta().getSourceParameters())) {
+      return pipeDataInsertionDisabledStatus();
+    }
     return startPipe(pipeName, isTableModel);
   }
 
+  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().getSourceParameters().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,
+        PipeMessages
+            .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()}. */
   private TSStatus stopPipe(String pipeName) {
     return stopPipe(pipeName, false);

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

@@ -31,6 +31,7 @@
 import org.apache.iotdb.confignode.consensus.response.subscription.TopicTableResp;
 import org.apache.iotdb.confignode.i18n.ManagerMessages;
 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.TAlterTopicReq;
@@ -187,8 +188,10 @@ public void unlockTopicAlteration(final String topicName, final boolean isTableM
   /////////////////////////////// Meta sync ///////////////////////////////
 
   public void startSubscriptionMetaSync() {
-    subscriptionMetaSyncer.start();
-    subscriptionOwnerLeaseSyncer.start();
+    if (PipeConfigNodeAgent.isSubscriptionFeatureEnabled()) {
+      subscriptionMetaSyncer.start();
+      subscriptionOwnerLeaseSyncer.start();
+    }
   }
 
   public void stopSubscriptionMetaSync() {

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

@@ -22,6 +22,7 @@
 import org.apache.iotdb.confignode.manager.load.subscriber.ConsensusGroupStatisticsChangeEvent;
 import org.apache.iotdb.confignode.manager.load.subscriber.IClusterStatusSubscriber;
 import org.apache.iotdb.confignode.manager.load.subscriber.NodeStatisticsChangeEvent;
+import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent;
 
 public class SubscriptionLeaderChangeHandler implements IClusterStatusSubscriber {
 
@@ -33,11 +34,15 @@ public SubscriptionLeaderChangeHandler(final SubscriptionRuntimeCoordinator runt
 
   @Override
   public void onNodeStatisticsChanged(final NodeStatisticsChangeEvent event) {
-    runtimeCoordinator.handleNodeStatisticsChange(event);
+    if (PipeConfigNodeAgent.isSubscriptionFeatureEnabled()) {
+      runtimeCoordinator.handleNodeStatisticsChange(event);
+    }
   }
 
   @Override
   public void onConsensusGroupStatisticsChanged(final ConsensusGroupStatisticsChangeEvent event) {
-    runtimeCoordinator.handleLeaderChangeEvent(event);
+    if (PipeConfigNodeAgent.isSubscriptionFeatureEnabled()) {
+      runtimeCoordinator.handleLeaderChangeEvent(event);
+    }
   }
 }

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

@@ -25,6 +25,7 @@
 import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
 import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
 import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
+import org.apache.iotdb.commons.i18n.PipeMessages;
 import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
 import org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
@@ -57,6 +58,7 @@
 import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
 import org.apache.iotdb.confignode.i18n.ManagerMessages;
 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;
@@ -198,6 +200,7 @@
             createPipeRequest.getExtractorAttributes(),
             createPipeRequest.getProcessorAttributes(),
             createPipeRequest.getConnectorAttributes());
+    checkPipeDataInsertionAvailability(pipeStaticMeta);
     if (!pipeMetaKeeper.containsPipeMetaOverlapped(pipeStaticMeta)) {
       return true;
     }
@@ -315,6 +318,7 @@
             alterPipeRequest.getExtractorAttributes(),
             alterPipeRequest.getProcessorAttributes(),
             alterPipeRequest.getConnectorAttributes());
+    checkPipeDataInsertionAvailability(updatedPipeStaticMeta);
     if (pipeMetaKeeper.containsPipeMetaOverlapped(
         updatedPipeStaticMeta, pipeStaticMetaFromCoordinator)) {
       final String exceptionMessage =
@@ -381,6 +385,7 @@
       throw new PipeException(exceptionMessage);
     }
 
+    checkPipeDataInsertionAvailability(getPipeMetaByPipeName(pipeName).getStaticMeta());
     final PipeStatus pipeStatus = getPipeStatus(pipeName);
     if (pipeStatus == PipeStatus.DROPPED) {
       final String exceptionMessage =
@@ -410,6 +415,8 @@
       throw new PipeException(exceptionMessage);
     }
 
+    checkPipeDataInsertionAvailability(
+        getPipeMetaByPipeName(pipeName, isTableModel).getStaticMeta());
     final PipeStatus pipeStatus = getPipeStatus(pipeName, isTableModel);
     if (pipeStatus == PipeStatus.DROPPED) {
       final String exceptionMessage =
@@ -429,6 +436,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.getSourceParameters())) {
+      throw new PipeException(
+          PipeMessages
+              .MESSAGE_PIPE_DATA_INSERTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_7D669ECC);
+    }
+  }
+
   public void checkBeforeStopPipe(final String pipeName) throws PipeException {
     acquireReadLock();
     try {
@@ -1171,7 +1188,7 @@
     try {
       final boolean hasException =
           recordDataNodePushPipeMetaExceptionsInternal(respMap, fallbackPipeStaticMeta);
       pipeMetaExceptionPersistencePending |= hasException;
       return hasException;
     } finally {
       releaseWriteLock();