diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index 322b57ed59dc8..4e60e599e1d5c 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/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."; diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index aaaff13b47858..726a87487f44e 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/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 清理失败已禁用写入，需要重启。"; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java index 23df702ca40a7..ff11b178aaa17 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java +++ b/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 @@ 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, + ConfigNodeMessages + .MESSAGE_SUBSCRIPTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_194CE431); + } + + private TSStatus confirmPipeDataInsertionAvailability( + final Map 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 @@ public TSStatus alterLogicalView(final TAlterLogicalViewReq 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; @@ -2210,7 +2236,7 @@ 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; @@ -2242,7 +2268,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; @@ -2258,7 +2284,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; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/PipeConfigNodeAgent.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/PipeConfigNodeAgent.java index 834b4e0ce3573..2bd1d65c01622 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/PipeConfigNodeAgent.java +++ b/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 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; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgent.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgent.java index d60b8b3b4c7be..5851526611a08 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgent.java +++ b/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); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java index 502af676b3943..7ee4a7be040f4 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java +++ b/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 @@ 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 = 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 @@ public TSStatus alterPipe(TAlterPipeReq req) { 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,6 +147,20 @@ public TSStatus startPipe(String pipeName) { 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); @@ -129,6 +168,32 @@ public TSStatus startPipe(String pipeName) { 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)) { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinator.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinator.java index ac88f31f92128..847ce2ecb03e3 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinator.java +++ b/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() { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java index 02e7aafb79c31..5686a635a65bf 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java +++ b/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 @@ public boolean checkBeforeCreatePipe(final TCreatePipeReq createPipeRequest) 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 @@ private boolean checkAndUpdateRequestBeforeAlterPipeInternal(final TAlterPipeReq } } + checkPipeDataInsertionAvailability( + new PipeStaticMeta( + alterPipeRequest.getPipeName(), + System.currentTimeMillis(), + alterPipeRequest.getExtractorAttributes(), + alterPipeRequest.getProcessorAttributes(), + alterPipeRequest.getConnectorAttributes())); + return true; } @@ -303,6 +320,7 @@ private void checkBeforeStartPipeInternal(final String pipeName) throws PipeExce 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 checkBeforeStartPipeInternal(final String pipeName) throws PipeExce } } + 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 { diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgentTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgentTest.java new file mode 100644 index 0000000000000..6fa9512d62e52 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgentTest.java @@ -0,0 +1,127 @@ +/* + * 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.manager.pipe.agent.task; + +import org.apache.iotdb.commons.pipe.agent.task.PipeTask; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus; +import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; +import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent; +import org.apache.iotdb.consensus.ConsensusFactory; + +import org.junit.Assert; +import org.junit.Test; +import org.mockito.InOrder; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.mockito.Mockito.doCallRealMethod; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; + +public class PipeConfigNodeTaskAgentTest { + @Test + public void testRecoveryDropsDisallowedPipeAndKeepsMetadataPipes() { + final String originalConsensus = + ConfigNodeDescriptor.getInstance().getConf().getDataRegionConsensusProtocolClass(); + final boolean leaderReady = PipeConfigNodeAgent.runtime().isLeaderReady(); + final TestingTaskAgent agent = new TestingTaskAgent(); + try { + PipeConfigNodeAgent.runtime().notifyLeaderReady(); + ConfigNodeDescriptor.getInstance() + .getConf() + .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS); + final PipeMeta insertion = meta("old", "all"); + Assert.assertNull(agent.handleSinglePipeMetaChanges(insertion)); + final PipeTask oldTask = agent.tasks.get("old"); + verify(oldTask).start(); + ConfigNodeDescriptor.getInstance() + .getConf() + .setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS); + Assert.assertNull(agent.handleSinglePipeMetaChanges(insertion)); + verify(oldTask).drop(); + // In 1.3, task.drop() stops the stage internally. Exercise the real stage lifecycle and + // verify the same stop-before-cleanup guarantee instead of an agent-level stop() call. + final PipeConfigNodeTaskStage oldStage = agent.stages.get("old"); + final InOrder dropOrder = inOrder(oldStage); + dropOrder.verify(oldStage).stopSubtask(); + dropOrder.verify(oldStage).dropSubtask(); + Assert.assertFalse(agent.contains("old")); + Assert.assertNull(agent.handleSinglePipeMetaChanges(meta("blocked", "all"))); + Assert.assertFalse(agent.contains("blocked")); + Assert.assertFalse(agent.tasks.containsKey("blocked")); + for (String inclusion : new String[] {"schema", "data.delete", "schema,data.delete"}) { + Assert.assertNull(agent.handleSinglePipeMetaChanges(meta(inclusion, inclusion))); + verify(agent.tasks.get(inclusion)).start(); + Assert.assertTrue(agent.contains(inclusion)); + } + } finally { + agent.dropAllPipeTasks(); + ConfigNodeDescriptor.getInstance() + .getConf() + .setDataRegionConsensusProtocolClass(originalConsensus); + if (!leaderReady) { + PipeConfigNodeAgent.runtime().notifyLeaderUnavailable(); + } + } + } + + private PipeMeta meta(String name, String inclusion) { + final PipeMeta meta = + new PipeMeta( + new PipeStaticMeta( + name, + 1L, + new HashMap<>(Map.of("inclusion", inclusion)), + new HashMap<>(), + new HashMap<>()), + new PipeRuntimeMeta()); + meta.getRuntimeMeta().getStatus().set(PipeStatus.RUNNING); + return meta; + } + + private static class TestingTaskAgent extends PipeConfigNodeTaskAgent { + private final Map tasks = new HashMap<>(); + private final Map stages = new HashMap<>(); + + @Override + protected Map buildPipeTasks(PipeMeta meta) { + final PipeConfigNodeTaskStage stage = mock(PipeConfigNodeTaskStage.class); + doCallRealMethod().when(stage).create(); + doCallRealMethod().when(stage).start(); + doCallRealMethod().when(stage).stop(); + doCallRealMethod().when(stage).drop(); + final PipeTask task = spy(new PipeConfigNodeTask(stage)); + tasks.put(meta.getStaticMeta().getPipeName(), task); + stages.put(meta.getStaticMeta().getPipeName(), stage); + return Collections.singletonMap(1, task); + } + + private boolean contains(String name) { + return pipeMetaKeeper.containsPipeMeta(name); + } + } +} diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinatorTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinatorTest.java new file mode 100644 index 0000000000000..fa47bb9c3c239 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinatorTest.java @@ -0,0 +1,140 @@ +/* + * 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.manager.pipe.coordinator.task; + +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; +import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; +import org.apache.iotdb.confignode.consensus.request.write.pipe.task.CreatePipePlanV2; +import org.apache.iotdb.confignode.consensus.request.write.pipe.task.DropPipePlanV2; +import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo; +import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoMoreInteractions; + +@RunWith(Parameterized.class) +public class PipeTaskCoordinatorTest { + @Parameterized.Parameters(name = "{0}") + public static Collection parameters() { + final Collection cases = new ArrayList<>(); + for (String protocol : + new String[] { + ConsensusFactory.RATIS_CONSENSUS, + ConsensusFactory.IOT_CONSENSUS, + ConsensusFactory.IOT_CONSENSUS_V2, + ConsensusFactory.SIMPLE_CONSENSUS + }) { + cases.add(new Object[] {protocol}); + } + return cases; + } + + private final String protocol; + private String originalProtocol; + private final ConfigManager configManager = mock(ConfigManager.class); + private final PipeTaskInfo pipeTaskInfo = + new PipeTaskInfo() { + @Override + public PipeMeta getPipeMetaByPipeName(String name) { + // Schedule a real metadata removal just before lookup, after the RPC's earlier checks. + Assert.assertTrue(isPipeExisted(name)); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + dropPipe(new DropPipePlanV2(name)).getCode()); + return super.getPipeMetaByPipeName(name); + } + }; + private final PipeTaskCoordinator coordinator = + new PipeTaskCoordinator(configManager, pipeTaskInfo); + + public PipeTaskCoordinatorTest(String protocol) { + this.protocol = protocol; + } + + @Before + public void setUp() { + originalProtocol = + ConfigNodeDescriptor.getInstance().getConf().getDataRegionConsensusProtocolClass(); + ConfigNodeDescriptor.getInstance().getConf().setDataRegionConsensusProtocolClass(protocol); + final Map source = new HashMap<>(Map.of("inclusion", "data.insert")); + pipeTaskInfo.createPipe( + new CreatePipePlanV2( + new PipeStaticMeta("p", 1L, source, new HashMap<>(), new HashMap<>()), + new PipeRuntimeMeta())); + Assert.assertTrue(pipeTaskInfo.isPipeExisted("p")); + } + + @After + public void tearDown() { + ConfigNodeDescriptor.getInstance() + .getConf() + .setDataRegionConsensusProtocolClass(originalProtocol); + } + + @Test + public void testAlterHandlesConcurrentDrop() { + Assert.assertEquals( + TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), + coordinator.alterPipe(alterRequest(false)).getCode()); + verifyNoMoreInteractions(configManager); + Assert.assertFalse(pipeTaskInfo.isPipeExisted("p")); + } + + @Test + public void testAlterIfExistsHandlesConcurrentDrop() { + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + coordinator.alterPipe(alterRequest(true)).getCode()); + verifyNoMoreInteractions(configManager); + Assert.assertFalse(pipeTaskInfo.isPipeExisted("p")); + } + + @Test + public void testStartHandlesConcurrentDrop() { + Assert.assertEquals( + TSStatusCode.PIPE_ERROR.getStatusCode(), coordinator.startPipe("p").getCode()); + verifyNoMoreInteractions(configManager); + Assert.assertFalse(pipeTaskInfo.isPipeExisted("p")); + } + + private TAlterPipeReq alterRequest(boolean ifExists) { + return new TAlterPipeReq("p", new HashMap<>(), new HashMap<>(), false, false) + .setExtractorAttributes(new HashMap<>(Map.of("inclusion", "schema"))) + .setIsReplaceAllExtractorAttributes(false) + .setIfExistsCondition(ifExists); + } +} diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoRatisTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoRatisTest.java new file mode 100644 index 0000000000000..06c716106b934 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoRatisTest.java @@ -0,0 +1,145 @@ +/* + * 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.commons.pipe.agent.task.meta.PipeRuntimeMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; +import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; +import org.apache.iotdb.confignode.consensus.request.write.pipe.task.AlterPipePlanV2; +import org.apache.iotdb.confignode.consensus.request.write.pipe.task.CreatePipePlanV2; +import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; +import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent; +import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.pipe.api.exception.PipeException; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.util.HashMap; +import java.util.Map; + +public class PipeTaskInfoRatisTest { + private String originalConsensus; + private final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo(); + + @Before + public void setUp() { + originalConsensus = + ConfigNodeDescriptor.getInstance().getConf().getDataRegionConsensusProtocolClass(); + ConfigNodeDescriptor.getInstance() + .getConf() + .setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS); + } + + @After + public void tearDown() { + ConfigNodeDescriptor.getInstance() + .getConf() + .setDataRegionConsensusProtocolClass(originalConsensus); + } + + @Test + public void testCreateAndStartValidateUnderProcedureLock() { + final TCreatePipeReq insertion = createRequest("insertion", Map.of()); + assertDisabled(() -> pipeTaskInfo.checkBeforeCreatePipe(insertion)); + assertDisabled( + () -> + pipeTaskInfo.checkBeforeCreatePipe( + createRequest("suffix", Map.of("inclusion", "data.insert.extra")))); + for (final String inclusion : new String[] {"schema", "data.delete", "schema,data.delete"}) { + final String name = inclusion.replace('.', '_').replace(',', '_'); + final TCreatePipeReq allowed = createRequest(name, Map.of("inclusion", inclusion)); + Assert.assertTrue(pipeTaskInfo.checkBeforeCreatePipe(allowed)); + store(allowed); + pipeTaskInfo.checkBeforeStartPipe(name); + } + // Persisted metadata can predate the restriction and must be checked on START as well. + store(insertion); + assertDisabled(() -> pipeTaskInfo.checkBeforeStartPipe("insertion")); + } + + @Test + public void testAlterRevalidatesAgainstLatestSerializedMetadata() { + final TCreatePipeReq initial = createRequest("p", Map.of("inclusion", "schema")); + store(initial); + final Map firstAttributes = + Map.of("source.inclusion", "all", "source.inclusion.exclusion", "data.insert"); + final Map secondAttributes = Map.of("extractor.inclusion.exclusion", ""); + // Both partial updates are allowed against the original RPC-time snapshot. + Assert.assertTrue(PipeConfigNodeAgent.isPipeDataInsertionEnabled(firstAttributes)); + Assert.assertTrue( + PipeConfigNodeAgent.isPipeDataInsertionEnabled( + Map.of("inclusion", "schema", "inclusion.exclusion", ""))); + final TAlterPipeReq first = alterRequest(firstAttributes, false); + pipeTaskInfo.checkAndUpdateRequestBeforeAlterPipe(first); + final PipeStaticMeta newMeta = + new PipeStaticMeta( + "p", 2L, first.getExtractorAttributes(), new HashMap<>(), new HashMap<>()); + pipeTaskInfo.alterPipe(new AlterPipePlanV2(newMeta, new PipeRuntimeMeta())); + assertDisabled( + () -> + pipeTaskInfo.checkAndUpdateRequestBeforeAlterPipe( + alterRequest(secondAttributes, false))); + Assert.assertTrue( + PipeConfigNodeAgent.isPipeDataInsertionEnabled( + pipeTaskInfo.getPipeMetaByPipeName("p").getStaticMeta().getExtractorParameters())); + assertDisabled( + () -> pipeTaskInfo.checkAndUpdateRequestBeforeAlterPipe(alterRequest(Map.of(), true))); + pipeTaskInfo.checkAndUpdateRequestBeforeAlterPipe( + alterRequest(Map.of("inclusion", "schema,data.delete"), true)); + } + + private TCreatePipeReq createRequest(String name, Map source) { + return new TCreatePipeReq() + .setPipeName(name) + .setExtractorAttributes(new HashMap<>(source)) + .setProcessorAttributes(new HashMap<>()) + .setConnectorAttributes(new HashMap<>()); + } + + private TAlterPipeReq alterRequest(Map source, boolean replace) { + return new TAlterPipeReq("p", new HashMap<>(), new HashMap<>(), false, false) + .setExtractorAttributes(new HashMap<>(source)) + .setIsReplaceAllExtractorAttributes(replace); + } + + private void store(TCreatePipeReq request) { + pipeTaskInfo.createPipe( + new CreatePipePlanV2( + new PipeStaticMeta( + request.getPipeName(), + 1L, + request.getExtractorAttributes(), + request.getProcessorAttributes(), + request.getConnectorAttributes()), + new PipeRuntimeMeta())); + } + + private void assertDisabled(Runnable action) { + Assert.assertEquals( + ConfigNodeMessages + .MESSAGE_PIPE_DATA_INSERTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_7D669ECC, + Assert.assertThrows(PipeException.class, action::run).getMessage()); + } +} diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 7322948f8d1e0..129d073180a8d 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -21,6 +21,14 @@ public final class DataNodePipeMessages { + 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."; + private DataNodePipeMessages() {} public static final String diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 9053efc61586c..5dde8a32c4946 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -21,6 +21,14 @@ public final class DataNodePipeMessages { + 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 同步不可用。"; + private DataNodePipeMessages() {} public static final String diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/PipeDataNodeAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/PipeDataNodeAgent.java index e350f0a6f4bbb..838a41e85a1b5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/PipeDataNodeAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/PipeDataNodeAgent.java @@ -19,15 +19,49 @@ package org.apache.iotdb.db.pipe.agent; +import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.commons.pipe.datastructure.options.PipeInclusionOptions; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.pipe.agent.plugin.PipeDataNodePluginAgent; import org.apache.iotdb.db.pipe.agent.receiver.PipeDataNodeReceiverAgent; import org.apache.iotdb.db.pipe.agent.runtime.PipeDataNodeRuntimeAgent; import org.apache.iotdb.db.pipe.agent.task.PipeDataNodeTaskAgent; import org.apache.iotdb.db.service.DataNode; +import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; + +import java.util.Collections; +import java.util.Map; /** {@link PipeDataNodeAgent} is the entry point of the pipe module in {@link DataNode}. */ public class PipeDataNodeAgent { + public static boolean isRatisConsensus() { + return ConsensusFactory.RATIS_CONSENSUS.equals( + IoTDBDescriptor.getInstance().getConfig().getDataRegionConsensusProtocolClass()); + } + + public static boolean isSubscriptionFeatureEnabled() { + return !isRatisConsensus(); + } + + public static boolean isPipeDataInsertionEnabled(final Map 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 PipeDataNodePluginAgent pipeDataNodePluginAgent; private final PipeDataNodeTaskAgent pipeDataNodeTaskAgent; private final PipeDataNodeRuntimeAgent pipeDataNodeRuntimeAgent; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java index 9021253940d4e..f13657ab986ed 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java @@ -144,6 +144,11 @@ protected boolean isShutdown() { return PipeDataNodeAgent.runtime().isShutdown(); } + @Override + protected boolean isPipeDataInsertionEnabled(final PipeStaticMeta pipeStaticMeta) { + return PipeDataNodeAgent.isPipeDataInsertionEnabled(pipeStaticMeta.getExtractorParameters()); + } + @Override protected Map buildPipeTasks(final PipeMeta pipeMetaFromConfigNode) throws IllegalPathException { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java index 61edbe5d60f00..6f406e85b4fa3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java @@ -20,6 +20,7 @@ package org.apache.iotdb.db.pipe.agent.task.execution; import org.apache.iotdb.commons.subscription.config.SubscriptionConfig; +import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import org.apache.iotdb.db.pipe.consensus.PipeConsensusSubtaskExecutor; import org.apache.iotdb.db.subscription.task.execution.SubscriptionSubtaskExecutor; @@ -58,6 +59,7 @@ private PipeSubtaskExecutorManager() { connectorExecutorSupplier = PipeSinkSubtaskExecutor::new; subscriptionExecutor = SubscriptionConfig.getInstance().getSubscriptionEnabled() + && PipeDataNodeAgent.isSubscriptionFeatureEnabled() ? new SubscriptionSubtaskExecutor() : null; consensusExecutorSupplier = PipeConsensusSubtaskExecutor::new; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java index 872c967ae0162..ba00cd2223091 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java @@ -25,6 +25,7 @@ import org.apache.iotdb.commons.pipe.receiver.PipeReceiverFilePathUtils; import org.apache.iotdb.commons.utils.FileUtils; import org.apache.iotdb.db.i18n.DataNodePipeMessages; +import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import org.apache.iotdb.db.pipe.sink.payload.legacy.PipeData; import org.apache.iotdb.db.pipe.sink.payload.legacy.TsFilePipeData; import org.apache.iotdb.db.protocol.session.SessionManager; @@ -112,6 +113,13 @@ public TSStatus handshake( return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, ""); } + private TSStatus pipeDataInsertionDisabledStatus() { + return RpcUtils.getStatus( + TSStatusCode.UNSUPPORTED_OPERATION, + DataNodePipeMessages + .MESSAGE_PIPE_DATA_INSERTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_7D669ECC); + } + private boolean validatePipeName(final TSyncIdentityInfo info) { return info.isSetPipeName() && Objects.isNull(FileUtils.getIllegalError4Directory(info.getPipeName())); @@ -154,6 +162,9 @@ public TSStatus transportPipeData(final ByteBuffer buff, final SessionInfo sessi buff.get(byteArray); pipeData = PipeData.createPipeData(byteArray); if (pipeData instanceof TsFilePipeData) { + if (PipeDataNodeAgent.isRatisConsensus()) { + return pipeDataInsertionDisabledStatus(); + } TsFilePipeData tsFilePipeData = (TsFilePipeData) pipeData; tsFilePipeData.setDatabase(identityInfo.getDatabase()); handleTsFilePipeData(tsFilePipeData, fileDir); @@ -257,6 +268,9 @@ public TSStatus transportFile(TSyncTransportMetaInfo metaInfo, ByteBuffer buff) if (identityInfo == null) { throw new TException("Thrift connection is not alive."); } + if (PipeDataNodeAgent.isRatisConsensus()) { + return pipeDataInsertionDisabledStatus(); + } LOGGER.debug( "Invoke transportData method from client ip = {}", identityInfo.getRemoteAddress()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java index 79416ac9b69bc..b417b0c5fe897 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java @@ -45,6 +45,7 @@ import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.DiskSpaceInsufficientException; +import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import org.apache.iotdb.db.pipe.event.common.schema.PipeSchemaRegionSnapshotEvent; import org.apache.iotdb.db.pipe.metric.receiver.PipeDataNodeReceiverMetrics; @@ -79,6 +80,7 @@ import org.apache.iotdb.db.queryengine.plan.execution.config.executor.ClusterConfigTaskExecutor; import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.DatabaseSchemaTask; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.view.AlterLogicalViewNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; import org.apache.iotdb.db.queryengine.plan.statement.Statement; import org.apache.iotdb.db.queryengine.plan.statement.StatementType; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement; @@ -200,6 +202,9 @@ public synchronized TPipeTransferResp receive(final TPipeTransferReq req) { if (Objects.nonNull(authResp)) { return authResp; } + if (PipeDataNodeAgent.isRatisConsensus() && isDataInsertionRequest(requestType)) { + return pipeDataInsertionDisabledResponse(); + } switch (requestType) { case HANDSHAKE_DATANODE_V1: { @@ -422,6 +427,41 @@ private TPipeTransferResp handleTransferTabletInsertNode( : executeStatementAndAddRedirectInfo(statement)); } + private static TPipeTransferResp pipeDataInsertionDisabledResponse() { + return new TPipeTransferResp( + RpcUtils.getStatus( + TSStatusCode.UNSUPPORTED_OPERATION, + DataNodePipeMessages + .MESSAGE_PIPE_DATA_INSERTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_7D669ECC)); + } + + private static boolean isDataInsertionRequest(final PipeRequestType requestType) { + return switch (requestType) { + case TRANSFER_TABLET_INSERT_NODE, + TRANSFER_TABLET_RAW, + TRANSFER_TABLET_BINARY, + TRANSFER_TABLET_BATCH, + TRANSFER_TS_FILE_PIECE, + TRANSFER_TS_FILE_SEAL, + TRANSFER_TS_FILE_PIECE_WITH_MOD, + TRANSFER_TS_FILE_SEAL_WITH_MOD -> + true; + case HANDSHAKE_CONFIGNODE_V1, + HANDSHAKE_DATANODE_V1, + HANDSHAKE_CONFIGNODE_V2, + HANDSHAKE_DATANODE_V2, + TRANSFER_SCHEMA_PLAN, + TRANSFER_SCHEMA_SNAPSHOT_PIECE, + TRANSFER_SCHEMA_SNAPSHOT_SEAL, + TRANSFER_CONFIG_PLAN, + TRANSFER_CONFIG_SNAPSHOT_PIECE, + TRANSFER_CONFIG_SNAPSHOT_SEAL, + TRANSFER_COMPRESSED, + TRANSFER_SLICE -> + false; + }; + } + private TPipeTransferResp handleTransferTabletBinary(final PipeTransferTabletBinaryReq req) { final InsertBaseStatement statement = req.constructStatement(); return new TPipeTransferResp( @@ -840,6 +880,9 @@ private TSStatus createSchemaSnapshotDatabaseIfNecessary(final PartialPath datab } private TPipeTransferResp handleTransferSchemaPlan(final PipeTransferPlanNodeReq req) { + if (PipeDataNodeAgent.isRatisConsensus() && req.getPlanNode() instanceof InsertNode) { + return pipeDataInsertionDisabledResponse(); + } // We may be able to skip the alter logical view's exception parsing because // the "AlterLogicalViewNode" is itself idempotent if (req.getPlanNode() instanceof AlterLogicalViewNode) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java index d3ec952d97aee..9943db71b8035 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java @@ -35,6 +35,8 @@ import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvironment; import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern; import org.apache.iotdb.commons.pipe.event.ProgressReportEvent; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; @@ -654,6 +656,13 @@ private boolean mayTsFileContainUnprocessedData(final TsFileResource resource) { } final ProgressIndex resourceProgressIndex = resource.getMaxProgressIndex(); + // Ratis writes do not carry Pipe progress indexes. Minimum means unknown, + // so replay conservatively after a leader change. + if (resourceProgressIndex instanceof MinimumProgressIndex + && ConsensusFactory.RATIS_CONSENSUS.equals( + IoTDBDescriptor.getInstance().getConfig().getDataRegionConsensusProtocolClass())) { + return true; + } if (innerStartIndex.isEqualOrAfter(resourceProgressIndex) || isProgressIndexCoveredByTimePartitionProgressIndex( resource, resourceProgressIndex, innerStartIndex)) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java index c0f0023edf0a6..8333b77d0cd91 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java @@ -44,6 +44,7 @@ import org.apache.iotdb.db.auth.AuthorityChecker; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import org.apache.iotdb.db.protocol.basic.BasicOpenSessionResp; import org.apache.iotdb.db.protocol.client.ConfigNodeClient; @@ -2834,6 +2835,15 @@ public TPipeSubscribeResp pipeSubscribe(final TPipeSubscribeReq req) { if (!SESSION_MANAGER.checkLogin(clientSession)) { return getNotLoggedInPipeSubscribeResp(); } + if (!PipeDataNodeAgent.isSubscriptionFeatureEnabled()) { + return new TPipeSubscribeResp( + RpcUtils.getStatus( + TSStatusCode.UNSUPPORTED_OPERATION, + DataNodePipeMessages + .MESSAGE_SUBSCRIPTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_194CE431), + PipeSubscribeResponseVersion.VERSION_1.getVersion(), + PipeSubscribeResponseType.ACK.getType()); + } return SubscriptionAgent.receiver().handle(req, clientSession.getUsername()); } finally { @@ -2973,7 +2983,9 @@ public void handleClientExit() { } PipeDataNodeAgent.receiver().thrift().handleClientExit(); PipeDataNodeAgent.receiver().legacy().handleClientExit(); - SubscriptionAgent.receiver().handleClientExit(); + if (PipeDataNodeAgent.isSubscriptionFeatureEnabled()) { + SubscriptionAgent.receiver().handleClientExit(); + } } /** diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java index 8adbb3ba3bb71..98f503493652f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java @@ -1256,7 +1256,7 @@ public TPushPipeMetaRespExceptionMessage handleSinglePipeMeta(final ByteBuffer p @Override public TPushTopicMetaResp pushTopicMeta(TPushTopicMetaReq req) { - if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) { + if (!isSubscriptionFeatureEnabled()) { return new TPushTopicMetaResp() .setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); } @@ -1284,7 +1284,7 @@ public TPushTopicMetaResp pushTopicMeta(TPushTopicMetaReq req) { @Override public TPushTopicMetaResp pushSingleTopicMeta(TPushSingleTopicMetaReq req) { - if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) { + if (!isSubscriptionFeatureEnabled()) { return new TPushTopicMetaResp() .setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); } @@ -1317,7 +1317,7 @@ public TPushTopicMetaResp pushSingleTopicMeta(TPushSingleTopicMetaReq req) { @Override public TPushTopicMetaResp pushMultiTopicMeta(TPushMultiTopicMetaReq req) { - if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) { + if (!isSubscriptionFeatureEnabled()) { return new TPushTopicMetaResp() .setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); } @@ -1369,7 +1369,7 @@ public TPushTopicMetaResp pushMultiTopicMeta(TPushMultiTopicMetaReq req) { @Override public TPushConsumerGroupMetaResp pushConsumerGroupMeta(TPushConsumerGroupMetaReq req) { - if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) { + if (!isSubscriptionFeatureEnabled()) { return new TPushConsumerGroupMetaResp() .setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); } @@ -1398,7 +1398,7 @@ public TPushConsumerGroupMetaResp pushConsumerGroupMeta(TPushConsumerGroupMetaRe @Override public TPushConsumerGroupMetaResp pushSingleConsumerGroupMeta( TPushSingleConsumerGroupMetaReq req) { - if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) { + if (!isSubscriptionFeatureEnabled()) { return new TPushConsumerGroupMetaResp() .setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); } @@ -2491,6 +2491,11 @@ private TEndPoint getConsensusEndPoint( return nodeLocation.getSchemaRegionConsensusEndPoint(); } + private boolean isSubscriptionFeatureEnabled() { + return SubscriptionConfig.getInstance().getSubscriptionEnabled() + && PipeDataNodeAgent.isSubscriptionFeatureEnabled(); + } + @Override public TSStatus createPipePlugin(TCreatePipePluginInstanceReq req) { try { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java index a0c85d9a72ce9..8252426cebb60 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java @@ -140,6 +140,7 @@ import org.apache.iotdb.db.exception.metadata.PathNotExistException; import org.apache.iotdb.db.exception.metadata.SchemaQuotaExceededException; import org.apache.iotdb.db.exception.sql.SemanticException; +import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import org.apache.iotdb.db.pipe.source.dataregion.DataRegionListeningFilter; import org.apache.iotdb.db.protocol.client.ConfigNodeClient; @@ -291,6 +292,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -316,6 +318,9 @@ private ClusterConfigTaskExecutorHolder() {} } private static final SettableFuture SUBSCRIPTION_NOT_ENABLED_ERROR_FUTURE; + private static final SettableFuture SUBSCRIPTION_DISABLED_BY_RATIS_ERROR_FUTURE; + private static final SettableFuture + PIPE_DATA_INSERTION_DISABLED_BY_RATIS_ERROR_FUTURE; static { SUBSCRIPTION_NOT_ENABLED_ERROR_FUTURE = SettableFuture.create(); @@ -323,6 +328,20 @@ private ClusterConfigTaskExecutorHolder() {} new IoTDBException( "Subscription is not enabled.", TSStatusCode.SUBSCRIPTION_NOT_ENABLED_ERROR.getStatusCode())); + + SUBSCRIPTION_DISABLED_BY_RATIS_ERROR_FUTURE = SettableFuture.create(); + SUBSCRIPTION_DISABLED_BY_RATIS_ERROR_FUTURE.setException( + new IoTDBException( + DataNodePipeMessages + .MESSAGE_SUBSCRIPTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_194CE431, + TSStatusCode.UNSUPPORTED_OPERATION.getStatusCode())); + + PIPE_DATA_INSERTION_DISABLED_BY_RATIS_ERROR_FUTURE = SettableFuture.create(); + PIPE_DATA_INSERTION_DISABLED_BY_RATIS_ERROR_FUTURE.setException( + new IoTDBException( + DataNodePipeMessages + .MESSAGE_PIPE_DATA_INSERTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_7D669ECC, + TSStatusCode.UNSUPPORTED_OPERATION.getStatusCode())); } public static ClusterConfigTaskExecutor getInstance() { @@ -1826,6 +1845,10 @@ public SettableFuture unsetSchemaTemplate( @Override public SettableFuture createPipe( final CreatePipeStatement createPipeStatement) { + if (!PipeDataNodeAgent.isPipeDataInsertionEnabled(createPipeStatement.getSourceAttributes())) { + return PIPE_DATA_INSERTION_DISABLED_BY_RATIS_ERROR_FUTURE; + } + final SettableFuture future = SettableFuture.create(); // Verify that Pipe is disabled if TSFile encryption is enabled @@ -2044,6 +2067,20 @@ public SettableFuture alterPipe(AlterPipeStatement alterPipeSt // Construct temporary pipe static meta for validation final String pipeName = alterPipeStatement.getPipeName(); + final PipeParameters sourceParameters = + alterPipeStatement.isReplaceAllExtractorAttributes() + ? new PipeParameters(alterPipeStatement.getExtractorAttributes()) + : new PipeParameters( + new HashMap<>( + pipeMetaFromCoordinator + .getStaticMeta() + .getExtractorParameters() + .getAttribute())) + .addOrReplaceEquivalentAttributes( + new PipeParameters(alterPipeStatement.getExtractorAttributes())); + if (!PipeDataNodeAgent.isPipeDataInsertionEnabled(sourceParameters)) { + return PIPE_DATA_INSERTION_DISABLED_BY_RATIS_ERROR_FUTURE; + } try { if (!alterPipeStatement.getExtractorAttributes().isEmpty()) { if (alterPipeStatement.isReplaceAllExtractorAttributes()) { @@ -2312,6 +2349,9 @@ public SettableFuture createTopic(CreateTopicStatement createT if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) { return SUBSCRIPTION_NOT_ENABLED_ERROR_FUTURE; } + if (!PipeDataNodeAgent.isSubscriptionFeatureEnabled()) { + return SUBSCRIPTION_DISABLED_BY_RATIS_ERROR_FUTURE; + } final SettableFuture future = SettableFuture.create(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java index 8967504e25a04..297ea6f737fad 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java @@ -803,7 +803,9 @@ private void setUp() throws StartupException { registerManager.register(CompactionTaskManager.getInstance()); // Register subscription agent before pipe agent - registerManager.register(SubscriptionAgent.runtime()); + if (PipeDataNodeAgent.isSubscriptionFeatureEnabled()) { + registerManager.register(SubscriptionAgent.runtime()); + } registerManager.register(PipeDataNodeAgent.runtime()); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java index e2ae494aa1236..9ce6a2cacf104 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java @@ -23,6 +23,8 @@ import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil; import org.apache.iotdb.commons.subscription.config.SubscriptionConfig; +import org.apache.iotdb.db.i18n.DataNodePipeMessages; +import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import org.apache.iotdb.db.subscription.receiver.SubscriptionReceiver; import org.apache.iotdb.db.subscription.receiver.SubscriptionReceiverV1; import org.apache.iotdb.rpc.RpcUtils; @@ -59,6 +61,15 @@ public class SubscriptionReceiverAgent { PipeSubscribeResponseVersion.VERSION_1.getVersion(), PipeSubscribeResponseType.ACK.getType()); + private static final TPipeSubscribeResp SUBSCRIPTION_DISABLED_BY_RATIS_RESP = + new TPipeSubscribeResp( + RpcUtils.getStatus( + TSStatusCode.UNSUPPORTED_OPERATION, + DataNodePipeMessages + .MESSAGE_SUBSCRIPTION_IS_DISABLED_WHEN_DATA_REGION_CONSENSUS_PROTOCOL_IS_RATIS_194CE431), + PipeSubscribeResponseVersion.VERSION_1.getVersion(), + PipeSubscribeResponseType.ACK.getType()); + private final ThreadLocal receiverThreadLocal = new ThreadLocal<>(); private final Set activeReceivers = ConcurrentHashMap.newKeySet(); private final ScheduledExecutorService receiverTimeoutChecker = @@ -68,12 +79,16 @@ public class SubscriptionReceiverAgent { SubscriptionReceiverAgent() { RECEIVER_CONSTRUCTORS.put( PipeSubscribeRequestVersion.VERSION_1.getVersion(), SubscriptionReceiverV1::new); - ScheduledExecutorUtil.safelyScheduleWithFixedDelay( - receiverTimeoutChecker, - this::checkReceiverTimeouts, - Math.max(1_000L, SubscriptionConfig.getInstance().getSubscriptionDefaultTimeoutInMs() / 2L), - Math.max(1_000L, SubscriptionConfig.getInstance().getSubscriptionDefaultTimeoutInMs() / 2L), - TimeUnit.MILLISECONDS); + if (PipeDataNodeAgent.isSubscriptionFeatureEnabled()) { + ScheduledExecutorUtil.safelyScheduleWithFixedDelay( + receiverTimeoutChecker, + this::checkReceiverTimeouts, + Math.max( + 1_000L, SubscriptionConfig.getInstance().getSubscriptionDefaultTimeoutInMs() / 2L), + Math.max( + 1_000L, SubscriptionConfig.getInstance().getSubscriptionDefaultTimeoutInMs() / 2L), + TimeUnit.MILLISECONDS); + } } public TPipeSubscribeResp handle(final TPipeSubscribeReq req) { @@ -87,6 +102,9 @@ public TPipeSubscribeResp handle(final TPipeSubscribeReq req, final String usern PipeSubscribeResponseVersion.VERSION_1.getVersion(), PipeSubscribeResponseType.ACK.getType()); } + if (!PipeDataNodeAgent.isSubscriptionFeatureEnabled()) { + return SUBSCRIPTION_DISABLED_BY_RATIS_RESP; + } if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) { return SUBSCRIPTION_NOT_ENABLED_ERROR_RESP; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionRuntimeAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionRuntimeAgent.java index aec165684635a..1262ae64c5a09 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionRuntimeAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionRuntimeAgent.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.service.IService; import org.apache.iotdb.commons.service.ServiceType; import org.apache.iotdb.commons.subscription.config.SubscriptionConfig; +import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import java.util.concurrent.atomic.AtomicBoolean; @@ -62,7 +63,8 @@ SubscriptionTopicAgent topic() { @Override public void start() throws StartupException { - if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) { + if (!SubscriptionConfig.getInstance().getSubscriptionEnabled() + || !PipeDataNodeAgent.isSubscriptionFeatureEnabled()) { return; } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java index a919810cc0474..072a6df702e3e 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java @@ -34,8 +34,11 @@ import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant; +import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; +import org.apache.iotdb.db.pipe.source.dataregion.DataRegionListeningFilter; +import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; import org.apache.iotdb.pipe.api.exception.PipeException; import org.awaitility.Awaitility; @@ -45,6 +48,7 @@ import java.lang.reflect.Field; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -55,6 +59,48 @@ public class PipeDataNodeTaskAgentTest { private static final int LOCAL_NODE_ID = 1; private static final int REGION_ID = 7; + @Test + public void testRatisDetectsPipeDataInsertionOption() throws Exception { + final String originalConsensus = + IoTDBDescriptor.getInstance().getConfig().getDataRegionConsensusProtocolClass(); + try { + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS); + + Assert.assertTrue( + PipeDataNodeAgent.isPipeDataInsertionEnabled( + Map.of(PipeSourceConstant.EXTRACTOR_INCLUSION_KEY, "schema"))); + Assert.assertTrue( + PipeDataNodeAgent.isPipeDataInsertionEnabled( + Map.of(PipeSourceConstant.EXTRACTOR_INCLUSION_KEY, "data.delete"))); + Assert.assertFalse( + PipeDataNodeAgent.isPipeDataInsertionEnabled( + Map.of(PipeSourceConstant.EXTRACTOR_INCLUSION_KEY, "data.insert"))); + Assert.assertFalse( + PipeDataNodeAgent.isPipeDataInsertionEnabled( + Map.of(PipeSourceConstant.EXTRACTOR_INCLUSION_KEY, "data.insert,data.delete"))); + for (final Map attributes : + List.of( + Map.of(), + Map.of("inclusion", "schema"), + Map.of("inclusion", "data.delete"), + Map.of("source.inclusion", "data.insert.extra"), + Map.of("inclusion", "all", "inclusion.exclusion", "data.insert.extra"))) { + Assert.assertEquals( + attributes.toString(), + !DataRegionListeningFilter.parseInsertionDeletionListeningOptionPair( + new PipeParameters(attributes)) + .getLeft(), + PipeDataNodeAgent.isPipeDataInsertionEnabled(attributes)); + } + } finally { + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(originalConsensus); + } + } + @Test public void testCreateMemoryCheckStillRunsWhenNoPipeTasksNeedToBeCreated() throws Exception { final boolean originalPipeEnableMemoryCheck = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgentTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgentTest.java index 2693e182f6e7f..a0bfafa6e844b 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgentTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgentTest.java @@ -20,9 +20,15 @@ package org.apache.iotdb.db.pipe.receiver.protocol.legacy; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.cluster.NodeStatus; import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.path.MeasurementPath; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.sink.payload.legacy.TsFilePipeData; +import org.apache.iotdb.db.queryengine.common.SessionInfo; +import org.apache.iotdb.db.storageengine.dataregion.modification.Deletion; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.TSyncIdentityInfo; import org.apache.iotdb.service.rpc.thrift.TSyncTransportMetaInfo; @@ -31,17 +37,21 @@ import ch.qos.logback.classic.spi.ILoggingEvent; import ch.qos.logback.core.read.ListAppender; import org.apache.commons.io.FileUtils; +import org.apache.tsfile.utils.ReadWriteIOUtils; import org.junit.After; import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.slf4j.LoggerFactory; +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; import java.io.File; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; +import java.time.ZoneId; public class IoTDBLegacyPipeReceiverAgentTest { @@ -80,6 +90,67 @@ public void tearDown() throws Exception { } } + @Test + public void testRatisRejectsFileTransfersButAllowsDeletion() throws Exception { + final String originalConsensus = + IoTDBDescriptor.getInstance().getConfig().getDataRegionConsensusProtocolClass(); + final NodeStatus originalStatus = CommonDescriptor.getInstance().getConfig().getNodeStatus(); + final String fileName = "ratis.tsfile"; + try { + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + agent + .handshake( + new TSyncIdentityInfo(PIPE_NAME, CREATE_TIME, "UNKNOWN", ""), + REMOTE_ADDRESS, + null, + null) + .getCode()); + final ByteBuffer bytes = ByteBuffer.wrap(new byte[] {1, 2, 3}); + Assert.assertEquals( + TSStatusCode.UNSUPPORTED_OPERATION.getStatusCode(), + agent.transportFile(new TSyncTransportMetaInfo(fileName, 0), bytes).getCode()); + Assert.assertEquals(0, bytes.position()); + Assert.assertFalse(Files.exists(getFileDataDir().resolve(fileName + ".patch"))); + + final Path patch = getFileDataDir().resolve(fileName + ".patch"); + Files.write(patch, new byte[] {4, 5, 6}); + Assert.assertEquals( + TSStatusCode.UNSUPPORTED_OPERATION.getStatusCode(), + agent + .transportPipeData(ByteBuffer.wrap(new TsFilePipeData("", fileName, 1).serialize())) + .getCode()); + Assert.assertTrue(Files.exists(patch)); + Assert.assertFalse(Files.exists(getFileDataDir().resolve(fileName))); + + final ByteArrayOutputStream deletionBytes = new ByteArrayOutputStream(); + try (final DataOutputStream output = new DataOutputStream(deletionBytes)) { + output.writeByte(1); + output.writeLong(2L); + ReadWriteIOUtils.write("root.sg", output); + new Deletion(new MeasurementPath("root.sg.d.s"), 0, 0, 1) + .serializeWithoutFileOffset(output); + } + // The real deletion loader must still be reached under Ratis. Read-only mode gives it a + // deterministic error without requiring a running cluster or executing an actual deletion. + CommonDescriptor.getInstance().getConfig().setNodeStatus(NodeStatus.ReadOnly); + final TSStatus deletionStatus = + agent.transportPipeData( + ByteBuffer.wrap(deletionBytes.toByteArray()), + new SessionInfo(1L, "root", ZoneId.systemDefault(), null)); + Assert.assertEquals(TSStatusCode.PIPESERVER_ERROR.getStatusCode(), deletionStatus.getCode()); + Assert.assertTrue(deletionStatus.getMessage().contains("storage engine readonly")); + } finally { + CommonDescriptor.getInstance().getConfig().setNodeStatus(originalStatus); + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(originalConsensus); + } + } + @Test public void testTransportFileRejectsPathTraversal() throws Exception { final String traversal = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java index d616c77eabb52..67189f2019d7a 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java @@ -19,23 +19,217 @@ package org.apache.iotdb.db.pipe.receiver.protocol.thrift; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.path.MeasurementPath; +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.pipe.sink.compressor.PipeCompressor; +import org.apache.iotdb.commons.pipe.sink.compressor.PipeCompressorFactory; +import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapPseudoTPipeTransferRequest; +import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferCompressedReq; +import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferSliceReq; +import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferPlanNodeReq; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.CreateTimeSeriesNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement; import org.apache.iotdb.db.storageengine.load.active.ActiveLoadPathHelper; import org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator; +import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.enums.CompressionType; +import org.apache.tsfile.file.metadata.enums.TSEncoding; import org.junit.Assert; import org.junit.Test; +import java.nio.ByteBuffer; import java.nio.file.Files; import java.nio.file.Path; import java.util.Arrays; +import java.util.Collections; +import java.util.List; import java.util.Map; public class IoTDBDataNodeReceiverTest { + @Test + public void testRatisRejectsInsertionInPlanNodeRequests() throws Exception { + final String originalProtocol = + IoTDBDescriptor.getInstance().getConfig().getDataRegionConsensusProtocolClass(); + try { + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS); + for (final PlanNode node : insertionPlans()) { + assertPlanRequestEnvelopes(node, true); + } + } finally { + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(originalProtocol); + } + } + + @Test + public void testRatisAllowsMetadataAndDeletionPlanNodeRequests() throws Exception { + final String originalProtocol = + IoTDBDescriptor.getInstance().getConfig().getDataRegionConsensusProtocolClass(); + try { + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS); + assertPlanRequestEnvelopes( + new DeleteDataNode( + new PlanNodeId("delete"), + Collections.singletonList(new MeasurementPath("root.sg.d.s")), + 0, + 1), + false); + assertPlanRequestEnvelopes( + new CreateTimeSeriesNode( + new PlanNodeId("schema"), + new MeasurementPath("root.sg.d.s"), + TSDataType.INT32, + TSEncoding.PLAIN, + CompressionType.UNCOMPRESSED, + null, + null, + null, + null), + false); + } finally { + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(originalProtocol); + } + } + + @Test + public void testNonRatisAllowsInsertionPlanNodeRequests() throws Exception { + final String originalProtocol = + IoTDBDescriptor.getInstance().getConfig().getDataRegionConsensusProtocolClass(); + try { + for (final String protocol : + new String[] { + ConsensusFactory.IOT_CONSENSUS, + ConsensusFactory.IOT_CONSENSUS_V2, + ConsensusFactory.SIMPLE_CONSENSUS + }) { + IoTDBDescriptor.getInstance().getConfig().setDataRegionConsensusProtocolClass(protocol); + for (final PlanNode node : insertionPlans()) { + assertPlanRequestEnvelopes(node, false); + } + } + } finally { + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(originalProtocol); + } + } + + private List insertionPlans() throws Exception { + final InsertRowNode row = + new InsertRowNode( + new PlanNodeId("row"), + new PartialPath("root.sg.d"), + false, + new String[] {"s"}, + new TSDataType[] {TSDataType.INT32}, + 1L, + new Object[] {1}, + false); + return Arrays.asList( + row, + new InsertRowsNode( + new PlanNodeId("rows"), Collections.singletonList(0), Collections.singletonList(row)), + new InsertTabletNode( + new PlanNodeId("tablet"), + new PartialPath("root.sg.d"), + false, + new String[] {"s"}, + new TSDataType[] {TSDataType.INT32}, + new long[] {1}, + null, + new Object[] {new int[] {1}}, + 1)); + } + + private void assertPlanRequestEnvelopes(final PlanNode node, final boolean rejected) + throws Exception { + final AdmissionTestReceiver receiver = new AdmissionTestReceiver(); + final int expectedCode = + rejected + ? TSStatusCode.UNSUPPORTED_OPERATION.getStatusCode() + : TSStatusCode.PIPE_HANDSHAKE_ERROR.getStatusCode(); + Assert.assertEquals( + expectedCode, + receiver.receive(PipeTransferPlanNodeReq.toTPipeTransferReq(node)).getStatus().getCode()); + + final TPipeTransferReq plain = PipeTransferPlanNodeReq.toTPipeTransferReq(node); + final TPipeTransferReq airGap = + new AirGapPseudoTPipeTransferRequest() + .setVersion(plain.getVersion()) + .setType(plain.getType()) + .setBody(plain.body.duplicate()); + Assert.assertEquals(expectedCode, receiver.receive(airGap).getStatus().getCode()); + Assert.assertEquals( + expectedCode, + receiver + .receive( + PipeTransferCompressedReq.toTPipeTransferReq( + PipeTransferPlanNodeReq.toTPipeTransferReq(node), + Collections.singletonList( + PipeCompressorFactory.getCompressor( + PipeCompressor.PipeCompressionType.GZIP.getIndex())))) + .getStatus() + .getCode()); + + final TPipeTransferReq sliced = PipeTransferPlanNodeReq.toTPipeTransferReq(node); + final ByteBuffer body = sliced.body.duplicate(); + final int midpoint = body.limit() / 2; + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + receiver + .receive( + PipeTransferSliceReq.toTPipeTransferReq( + 1, sliced.getType(), 0, 2, body, 0, midpoint)) + .getStatus() + .getCode()); + Assert.assertEquals( + expectedCode, + receiver + .receive( + PipeTransferSliceReq.toTPipeTransferReq( + 1, sliced.getType(), 1, 2, body, midpoint, body.limit())) + .getStatus() + .getCode()); + Assert.assertEquals(rejected ? 0 : 4, receiver.executionAttempts); + } + + private static class AdmissionTestReceiver extends IoTDBDataNodeReceiver { + private int executionAttempts; + + private AdmissionTestReceiver() { + hasPipeHandshakeCredential = true; + } + + @Override + protected TSStatus loginIfNecessary() { + // Record arrival at statement execution without starting a cluster or writing data. + executionAttempts++; + return new TSStatus(TSStatusCode.PIPE_HANDSHAKE_ERROR.getStatusCode()); + } + } + @Test public void testLoadTsFileSyncStatementUsesTreeDatabaseLevelFromDatabaseName() throws Exception { final Path tsFile = Files.createTempFile("pipe-load-tree-database-level", ".tsfile"); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java index 3b264f749a249..0f33640fec16d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java @@ -34,6 +34,8 @@ import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixPipePattern; import org.apache.iotdb.commons.pipe.event.ProgressReportEvent; import org.apache.iotdb.commons.utils.FileUtils; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.pipe.agent.task.connection.PipeEventCollector; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; @@ -610,6 +612,50 @@ public void testQueryPriorityOrderReportsProgressAfterAllHistoricalResources() { Assert.assertFalse((Boolean) getPrivateField(source, "shouldReportMaxHistoricalProgressIndex")); } + @Test + public void testRatisFilesWithoutPipeProgressAreNotCoveredByCheckpoint() throws Exception { + final String originalConsensus = + IoTDBDescriptor.getInstance().getConfig().getDataRegionConsensusProtocolClass(); + final File tempDir = Files.createTempDirectory("pipeHistoricalRatisCoverage").toFile(); + try { + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS); + assertMayTsFileContainUnprocessedData( + tempDir, + "unknown-initial.tsfile", + MinimumProgressIndex.INSTANCE, + MinimumProgressIndex.INSTANCE, + true); + assertMayTsFileContainUnprocessedData( + tempDir, + "unknown-recovery.tsfile", + new SimpleProgressIndex(0, 100), + MinimumProgressIndex.INSTANCE, + true); + assertMayTsFileContainUnprocessedData( + tempDir, + "indexed-covered.tsfile", + new SimpleProgressIndex(0, 100), + new SimpleProgressIndex(0, 50), + false); + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS); + assertMayTsFileContainUnprocessedData( + tempDir, + "other-consensus.tsfile", + MinimumProgressIndex.INSTANCE, + MinimumProgressIndex.INSTANCE, + false); + } finally { + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(originalConsensus); + FileUtils.deleteFileOrDirectory(tempDir); + } + } + @Test public void testMayTsFileContainUnprocessedDataUsesTimePartitionProgressCoverage() throws Exception { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgentTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgentTest.java new file mode 100644 index 0000000000000..958e9588307e2 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgentTest.java @@ -0,0 +1,72 @@ +/* + * 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.db.subscription.agent; + +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.rpc.subscription.config.ConsumerConfig; +import org.apache.iotdb.rpc.subscription.payload.request.PipeSubscribeHandshakeReq; + +import org.junit.Assert; +import org.junit.Test; + +import java.lang.reflect.Field; +import java.util.Map; +import java.util.concurrent.ScheduledExecutorService; + +public class SubscriptionReceiverAgentTest { + @Test + public void testRatisRejectsSubscriptionBeforeCreatingReceiver() throws Exception { + final String originalConsensus = + IoTDBDescriptor.getInstance().getConfig().getDataRegionConsensusProtocolClass(); + SubscriptionReceiverAgent agent = null; + try { + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS); + agent = new SubscriptionReceiverAgent(); + Assert.assertEquals( + TSStatusCode.UNSUPPORTED_OPERATION.getStatusCode(), + agent + .handle( + PipeSubscribeHandshakeReq.toTPipeSubscribeReq( + new ConsumerConfig( + Map.of("consumer-group-id", "group", "consumer-id", "consumer"))), + "root") + .getStatus() + .getCode()); + final Field receiver = + SubscriptionReceiverAgent.class.getDeclaredField("receiverThreadLocal"); + receiver.setAccessible(true); + Assert.assertNull(((ThreadLocal) receiver.get(agent)).get()); + } finally { + if (agent != null) { + final Field checker = + SubscriptionReceiverAgent.class.getDeclaredField("receiverTimeoutChecker"); + checker.setAccessible(true); + ((ScheduledExecutorService) checker.get(agent)).shutdownNow(); + } + IoTDBDescriptor.getInstance() + .getConfig() + .setDataRegionConsensusProtocolClass(originalConsensus); + } + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java index e2324f5c95b5e..adf9e50cf0f28 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java @@ -204,6 +204,10 @@ protected TPushPipeMetaRespExceptionMessage handleSinglePipeMetaChangesInternal( protected abstract boolean isShutdown(); + protected boolean isPipeDataInsertionEnabled(final PipeStaticMeta pipeStaticMeta) { + return true; + } + private void executeSinglePipeMetaChanges(final PipeMeta metaFromCoordinator) throws IllegalPathException { final String pipeName = metaFromCoordinator.getStaticMeta().getPipeName(); @@ -220,6 +224,14 @@ private void executeSinglePipeMetaChanges(final PipeMeta metaFromCoordinator) return; } + if (!isPipeDataInsertionEnabled(metaFromCoordinator.getStaticMeta())) { + final PipeMeta metaInAgent = pipeMetaKeeper.getPipeMeta(pipeName); + if (metaInAgent != null) { + dropPipe(pipeName, metaInAgent.getStaticMeta().getCreationTime()); + } + return; + } + final PipeMeta metaInAgent = pipeMetaKeeper.getPipeMeta(pipeName); // If pipe meta does not exist on local agent, create a new pipe diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/options/PipeInclusionOptions.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/options/PipeInclusionOptions.java index 8564b0116b086..832626707c152 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/options/PipeInclusionOptions.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/options/PipeInclusionOptions.java @@ -208,6 +208,18 @@ public static String getExclusionString(final PipeParameters parameters) { PipeSourceConstant.EXTRACTOR_EXCLUSION_DEFAULT_VALUE); } + public static boolean isDataInsertionIncluded(final PipeParameters parameters) + throws IllegalPathException { + // Match the data-region listening filter, including prefixes longer than the concrete option + // that the existing option validator accepts. + final PartialPath insertion = + new PartialPath(PipeSourceConstant.EXTRACTOR_INCLUSION_DEFAULT_VALUE); + return parseOptions(getInclusionString(parameters)).stream() + .anyMatch(insertion::overlapWithFullPathPrefix) + && parseOptions(getExclusionString(parameters)).stream() + .noneMatch(insertion::overlapWithFullPathPrefix); + } + public static boolean areOptionsEnabled(final PipeParameters parameters, final String... options) throws IllegalPathException { final Set inclusionOptions = parseOptions(getInclusionString(parameters)); diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/options/PipeInclusionOptionsTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/options/PipeInclusionOptionsTest.java index f3f2290db6176..f50b85e7e7979 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/options/PipeInclusionOptionsTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/options/PipeInclusionOptionsTest.java @@ -20,10 +20,14 @@ package org.apache.iotdb.commons.pipe.datastructure.options; import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; import org.junit.Assert; import org.junit.Test; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; import java.util.Set; public class PipeInclusionOptionsTest { @@ -45,4 +49,41 @@ public void testSchemaDeleteAliasIncludesOrdinaryTimeseriesDrop() throws Excepti Assert.assertFalse(options.contains(new PartialPath("schema.timeseries.ordinary.delete"))); Assert.assertTrue(PipeInclusionOptions.optionsAreAllLegal("schema.delete")); } + + @Test + public void testDataInsertionOptionSelection() throws Exception { + Assert.assertTrue( + PipeInclusionOptions.areOptionsEnabled( + new PipeParameters(Collections.emptyMap()), "data.insert")); + Assert.assertTrue( + PipeInclusionOptions.areOptionsEnabled( + new PipeParameters(Map.of("extractor.inclusion", "data.insert,data.delete")), + "data.insert")); + Assert.assertFalse( + PipeInclusionOptions.areOptionsEnabled( + new PipeParameters(Map.of("extractor.inclusion", "schema")), "data.insert")); + Assert.assertFalse( + PipeInclusionOptions.areOptionsEnabled( + new PipeParameters(Map.of("extractor.inclusion", "data.delete")), "data.insert")); + + final Map attributes = new HashMap<>(); + attributes.put("extractor.inclusion", "all"); + attributes.put("extractor.inclusion.exclusion", "data.insert"); + Assert.assertFalse( + PipeInclusionOptions.areOptionsEnabled(new PipeParameters(attributes), "data.insert")); + } + + @Test + public void testDataInsertionUsesListeningPrefixSemantics() throws Exception { + Assert.assertTrue(PipeInclusionOptions.optionsAreAllLegal("data.insert.extra")); + Assert.assertTrue( + PipeInclusionOptions.isDataInsertionIncluded( + new PipeParameters(Map.of("source.inclusion", "data.insert.extra")))); + Assert.assertFalse( + PipeInclusionOptions.isDataInsertionIncluded( + new PipeParameters( + Map.of( + "source.inclusion", "all", + "source.inclusion.exclusion", "data.insert.extra")))); + } }