From 234eab86b6aa0a23689ab1b3111e16de858089b1 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 13 Aug 2026 15:52:33 +0800 Subject: [PATCH] Subscription: skip runtime when disabled --- .../thrift/impl/ClientRPCServiceImpl.java | 4 +- .../impl/DataNodeInternalRPCServiceImpl.java | 25 ++++++-- .../org/apache/iotdb/db/service/DataNode.java | 5 +- .../agent/SubscriptionReceiverAgent.java | 2 +- .../ConsensusSubscriptionSetupHandler.java | 9 +++ ...PCServiceImplSubscriptionDisabledTest.java | 64 +++++++++++++++++++ .../agent/SubscriptionReceiverAgentTest.java | 47 ++++++++++++++ ...ConsensusSubscriptionSetupHandlerTest.java | 15 +++++ 8 files changed, 163 insertions(+), 8 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplSubscriptionDisabledTest.java 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 790e022ded36..be0a00bddf82 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 @@ -3635,7 +3635,9 @@ public void handleClientExit() { } PipeDataNodeAgent.receiver().thrift().handleClientExit(); PipeDataNodeAgent.receiver().legacy().handleClientExit(); - SubscriptionAgent.receiver().handleClientExit(); + if (COMMON_CONFIG.getSubscriptionEnabled()) { + 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 cde5b09b4578..9ebaeaa44f8e 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 @@ -1627,6 +1627,11 @@ public TPushConsumerGroupMetaResp pushSingleConsumerGroupMeta( @Override public TPullCommitProgressResp pullCommitProgress(TPullCommitProgressReq req) { + if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) { + return new TPullCommitProgressResp(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())) + .setCommitRegionProgress(Collections.emptyMap()); + } + try { final int dataNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId(); final Map regionProgress = @@ -1643,6 +1648,10 @@ public TPullCommitProgressResp pullCommitProgress(TPullCommitProgressReq req) { @Override public TSStatus syncSubscriptionProgress(TSyncSubscriptionProgressReq req) { + if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) { + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + } + try { SubscriptionAgent.broker() .receiveSubscriptionProgress( @@ -1664,6 +1673,10 @@ public TSStatus syncSubscriptionProgress(TSyncSubscriptionProgressReq req) { @Override public TSStatus pushSubscriptionRuntime(TPushSubscriptionRuntimeReq req) { + if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) { + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + } + try { for (final TSubscriptionRuntimeStateEntry runtimeStateEntry : req.getRuntimeStates()) { ConsensusSubscriptionSetupHandler.applyRuntimeState( @@ -2437,8 +2450,12 @@ public TDataNodeHeartbeatResp getDataNodeHeartBeat(TDataNodeHeartbeatReq req) th @Override public TSStatus updateRegionCache(TRegionRouteReq req) { - boolean result = ClusterPartitionFetcher.getInstance().updateRegionCache(req); - if (result) { + final boolean result = ClusterPartitionFetcher.getInstance().updateRegionCache(req); + if (!result) { + return RpcUtils.getStatus(TSStatusCode.PARTITION_CACHE_UPDATE_ERROR); + } + + if (SubscriptionConfig.getInstance().getSubscriptionEnabled()) { // Notify consensus subscription queues of any preferred-writer changes try { ConsensusSubscriptionSetupHandler.onRegionRouteChanged( @@ -2449,10 +2466,8 @@ public TSStatus updateRegionCache(TRegionRouteReq req) { .MISC_LOG_FAILED_TO_PROCESS_CONSENSUS_SUBSCRIPTION_ROUTE_UPDATE_80D73E2B, e); } - return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); - } else { - return RpcUtils.getStatus(TSStatusCode.PARTITION_CACHE_UPDATE_ERROR); } + return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); } private Map getJudgedLeaders() { 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 484842686232..760c5747c1af 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 @@ -51,6 +51,7 @@ import org.apache.iotdb.commons.service.RegisterManager; import org.apache.iotdb.commons.service.ServiceType; import org.apache.iotdb.commons.service.metric.MetricService; +import org.apache.iotdb.commons.subscription.config.SubscriptionConfig; import org.apache.iotdb.commons.trigger.TriggerInformation; import org.apache.iotdb.commons.trigger.exception.TriggerManagementException; import org.apache.iotdb.commons.trigger.service.TriggerExecutableManager; @@ -941,7 +942,9 @@ private void setUp() throws StartupException, IOException { registerInternalRPCService(); // Register subscription agent before pipe agent - registerManager.register(SubscriptionAgent.runtime()); + if (SubscriptionConfig.getInstance().getSubscriptionEnabled()) { + registerManager.register(SubscriptionAgent.runtime()); + } registerManager.register(PipeDataNodeAgent.runtime()); // Start GRASS Service 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 d918b4aed474..1f26397cbcc5 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 @@ -75,7 +75,7 @@ public class SubscriptionReceiverAgent { private final ScheduledExecutorService receiverTimeoutChecker; SubscriptionReceiverAgent() { - this(SubscriptionReceiverV1::new, true); + this(SubscriptionReceiverV1::new, SubscriptionConfig.getInstance().getSubscriptionEnabled()); } SubscriptionReceiverAgent( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java index 3fea7020e72b..0706ddd65377 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java @@ -28,6 +28,7 @@ import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern; import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern; +import org.apache.iotdb.commons.subscription.config.SubscriptionConfig; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.consensus.IConsensus; import org.apache.iotdb.consensus.iot.IoTConsensus; @@ -741,6 +742,10 @@ public static void handleNewSubscriptions( public static void applyRuntimeState( final TConsensusGroupId groupId, final ConsensusRegionRuntimeState runtimeState) { + if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) { + return; + } + final int newPreferredNodeId = runtimeState.getPreferredWriterNodeId(); final Integer oldPreferredBoxed = lastKnownPreferredWriter.put(groupId, newPreferredNodeId); final int oldPreferredNodeId = (oldPreferredBoxed != null) ? oldPreferredBoxed : -1; @@ -772,6 +777,10 @@ public static void applyRuntimeState( public static void onRegionRouteChanged( final Map newMap, final long routingTimestamp) { + if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) { + return; + } + final int myNodeId = IOTDB_CONFIG.getDataNodeId(); for (final Map.Entry newEntry : newMap.entrySet()) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplSubscriptionDisabledTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplSubscriptionDisabledTest.java new file mode 100644 index 000000000000..27c217b1061e --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplSubscriptionDisabledTest.java @@ -0,0 +1,64 @@ +/* + * 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.protocol.thrift.impl; + +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.service.DataNode.DataNodeContext; +import org.apache.iotdb.mpp.rpc.thrift.TPullCommitProgressResp; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.mockito.Mockito; + +public class DataNodeInternalRPCServiceImplSubscriptionDisabledTest { + + @BeforeClass + public static void setUp() { + IoTDBDescriptor.getInstance().getConfig().setDataNodeId(0); + } + + @Test + public void testSubscriptionRuntimeRPCsAreNoOpWhenSubscriptionIsDisabled() { + final boolean subscriptionEnabled = + CommonDescriptor.getInstance().getConfig().getSubscriptionEnabled(); + try { + CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(false); + final DataNodeInternalRPCServiceImpl service = + new DataNodeInternalRPCServiceImpl(Mockito.mock(DataNodeContext.class)); + + final TPullCommitProgressResp pullResp = service.pullCommitProgress(null); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), pullResp.getStatus().getCode()); + Assert.assertTrue(pullResp.isSetCommitRegionProgress()); + Assert.assertTrue(pullResp.getCommitRegionProgress().isEmpty()); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + service.syncSubscriptionProgress(null).getCode()); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + service.pushSubscriptionRuntime(null).getCode()); + } finally { + CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(subscriptionEnabled); + } + } +} 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 index a151e83e21d0..29d4f4f47f81 100644 --- 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 @@ -20,6 +20,7 @@ package org.apache.iotdb.db.subscription.agent; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.db.subscription.receiver.SubscriptionReceiver; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; @@ -38,10 +39,12 @@ import org.junit.Test; import java.io.IOException; +import java.lang.reflect.Field; import java.util.HashMap; import java.util.Map; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -49,6 +52,43 @@ public class SubscriptionReceiverAgentTest { + @Test + public void testTimeoutCheckerIsNotScheduledWhenSubscriptionIsDisabled() throws Exception { + final boolean subscriptionEnabled = + CommonDescriptor.getInstance().getConfig().getSubscriptionEnabled(); + try { + CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(false); + + final SubscriptionReceiverAgent agent = new SubscriptionReceiverAgent(); + + Assert.assertNull(getReceiverTimeoutChecker(agent)); + } finally { + CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(subscriptionEnabled); + } + } + + @Test + public void testTimeoutCheckerIsScheduledWhenSubscriptionIsEnabled() throws Exception { + final boolean subscriptionEnabled = + CommonDescriptor.getInstance().getConfig().getSubscriptionEnabled(); + SubscriptionReceiverAgent agent = null; + try { + CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(true); + + agent = new SubscriptionReceiverAgent(); + + Assert.assertNotNull(getReceiverTimeoutChecker(agent)); + } finally { + if (agent != null) { + final ScheduledExecutorService receiverTimeoutChecker = getReceiverTimeoutChecker(agent); + if (receiverTimeoutChecker != null) { + receiverTimeoutChecker.shutdownNow(); + } + } + CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(subscriptionEnabled); + } + } + @Test public void testDisconnectedReceiverIsRetainedUntilTimeout() throws IOException { final CopyOnWriteArrayList receivers = new CopyOnWriteArrayList<>(); @@ -176,6 +216,13 @@ private SubscriptionReceiverAgent createAgent( return new SubscriptionReceiverAgent(constructor, false); } + private ScheduledExecutorService getReceiverTimeoutChecker(final SubscriptionReceiverAgent agent) + throws Exception { + final Field field = SubscriptionReceiverAgent.class.getDeclaredField("receiverTimeoutChecker"); + field.setAccessible(true); + return (ScheduledExecutorService) field.get(agent); + } + private TPipeSubscribeReq createHandshakeRequest( final String consumerGroupId, final String consumerId) throws IOException { final Map attributes = new HashMap<>(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java index 97ee9df9d3c5..d8d30f2eb87e 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.subscription.broker.consensus; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; import org.apache.iotdb.db.conf.IoTDBDescriptor; @@ -49,6 +50,20 @@ public class ConsensusSubscriptionSetupHandlerTest { @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + @Test + public void testRuntimeUpdatesAreIgnoredWhenSubscriptionIsDisabled() { + final boolean subscriptionEnabled = + CommonDescriptor.getInstance().getConfig().getSubscriptionEnabled(); + try { + CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(false); + + ConsensusSubscriptionSetupHandler.applyRuntimeState(null, null); + ConsensusSubscriptionSetupHandler.onRegionRouteChanged(null, 0); + } finally { + CommonDescriptor.getInstance().getConfig().setSubscriptionEnabled(subscriptionEnabled); + } + } + @Test public void testSingleTopicSetupFailurePropagates() { SubscriptionException failure = null;