This is an automated email from the ASF dual-hosted git repository. jackietien pushed a commit to branch rc/1.3.5 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 8d1fbbd6ff5202d7b11dd693407573fd0ad94b55 Author: Zhenyu Luo <[email protected]> AuthorDate: Fri Aug 8 10:11:50 2025 +0800 [To dev/1.3] Pipe: CN adds logic to check if Pipe is out of memory #16119 (#16120) * Pipe: CN adds logic to check if Pipe is out of memory * update AbstractOperatePipeProcedureV2 * update AbstractOperatePipeProcedureV2 * add it --- .../it/env/cluster/config/MppCommonConfig.java | 6 ++ .../env/cluster/config/MppSharedCommonConfig.java | 6 ++ .../it/env/remote/config/RemoteCommonConfig.java | 5 ++ .../org/apache/iotdb/itbase/env/CommonConfig.java | 2 + .../pipe/it/autocreate/IoTDBPipeMemoryIT.java | 98 ++++++++++++++++++++++ .../java/org/apache/iotdb/rpc/TSStatusCode.java | 1 + .../impl/pipe/AbstractOperatePipeProcedureV2.java | 28 +++++++ .../db/pipe/agent/task/PipeDataNodeTaskAgent.java | 9 +- .../impl/DataNodeInternalRPCServiceImpl.java | 21 +++-- .../commons/pipe/agent/task/PipeTaskAgent.java | 1 + 10 files changed, 169 insertions(+), 8 deletions(-) diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java index a7d8918d1c9..26e7d8df357 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java @@ -520,6 +520,12 @@ public class MppCommonConfig extends MppBaseConfig implements CommonConfig { return this; } + @Override + public CommonConfig setDatanodeMemoryProportion(String datanodeMemoryProportion) { + setProperty("datanode_memory_proportion", datanodeMemoryProportion); + return this; + } + // For part of the log directory public String getClusterConfigStr() { return fromConsensusFullNameToAbbr(properties.getProperty(CONFIG_NODE_CONSENSUS_PROTOCOL_CLASS)) diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java index e4a4ace7a44..6ad723a8013 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java @@ -533,4 +533,10 @@ public class MppSharedCommonConfig implements CommonConfig { cnConfig.setDefaultStorageGroupLevel(defaultStorageGroupLevel); return this; } + + public CommonConfig setDatanodeMemoryProportion(String datanodeMemoryProportion) { + dnConfig.setDatanodeMemoryProportion(datanodeMemoryProportion); + cnConfig.setDatanodeMemoryProportion(datanodeMemoryProportion); + return this; + } } diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java index 23a8dad317e..48f090943f5 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java @@ -370,4 +370,9 @@ public class RemoteCommonConfig implements CommonConfig { public CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled) { return this; } + + @Override + public CommonConfig setDatanodeMemoryProportion(String datanodeMemoryProportion) { + return this; + } } diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java index 178f182624b..5f39e911e39 100644 --- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java @@ -167,4 +167,6 @@ public interface CommonConfig { default CommonConfig setDefaultStorageGroupLevel(int defaultStorageGroupLevel) { return this; } + + CommonConfig setDatanodeMemoryProportion(String datanodeMemoryProportion); } diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeMemoryIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeMemoryIT.java new file mode 100644 index 00000000000..bb211a1ffe8 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeMemoryIT.java @@ -0,0 +1,98 @@ +/* + * 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.pipe.it.autocreate; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; +import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; +import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.MultiClusterIT2AutoCreateSchema; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.util.HashMap; +import java.util.Map; + +@RunWith(IoTDBTestRunner.class) +@Category({MultiClusterIT2AutoCreateSchema.class}) +public class IoTDBPipeMemoryIT extends AbstractPipeDualAutoIT { + + @Override + @Before + public void setUp() { + super.setUp(); + } + + @Override + protected void setupConfig() { + super.setupConfig(); + senderEnv + .getConfig() + .getCommonConfig() + .setPipeMemoryManagementEnabled(true) + .setIsPipeEnableMemoryCheck(true) + .setDatanodeMemoryProportion("1000:1000:1000:1000:1:1000"); + receiverEnv + .getConfig() + .getCommonConfig() + .setPipeMemoryManagementEnabled(true) + .setIsPipeEnableMemoryCheck(true) + .setDatanodeMemoryProportion("1000:1000:1000:1000:1:1000"); + } + + @Test + public void testCreatePipeMemoryManage() { + final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0); + final String receiverIp = receiverDataNode.getIp(); + final int receiverPort = receiverDataNode.getPort(); + + try (final SyncConfigNodeIServiceClient client = + (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) { + final Map<String, String> extractorAttributes = new HashMap<>(); + final Map<String, String> processorAttributes = new HashMap<>(); + final Map<String, String> connectorAttributes = new HashMap<>(); + extractorAttributes.put("user", "root"); + + connectorAttributes.put("connector", "iotdb-thrift-connector"); + connectorAttributes.put("connector.batch.enable", "false"); + connectorAttributes.put("connector.ip", receiverIp); + connectorAttributes.put("connector.port", Integer.toString(receiverPort)); + + final TSStatus status = + client.createPipe( + new TCreatePipeReq("p1", connectorAttributes) + .setExtractorAttributes(extractorAttributes) + .setProcessorAttributes(processorAttributes)); + + Assert.assertNotEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode()); + Assert.assertNotNull(status.getMessage()); + Assert.assertTrue(status.getMessage().contains("Not enough memory for pipe.")); + + } catch (Exception e) { + Assert.fail(e.getMessage()); + } + } +} diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java index e2ab4b0ff43..de1a5962d66 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java @@ -249,6 +249,7 @@ public enum TSStatusCode { PIPE_CONFIG_RECEIVER_HANDSHAKE_NEEDED(1811), PIPE_TRANSFER_SLICE_OUT_OF_ORDER(1812), PIPE_PUSH_META_TIMEOUT(1813), + PIPE_PUSH_META_NOT_ENOUGH_MEMORY(1814), // Subscription SUBSCRIPTION_VERSION_ERROR(1900), diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java index eb4ffc48c79..07059d6a0d9 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java @@ -424,11 +424,33 @@ public abstract class AbstractOperatePipeProcedureV2 public static String parsePushPipeMetaExceptionForPipe( final String pipeName, final Map<Integer, TPushPipeMetaResp> respMap) { final StringBuilder exceptionMessageBuilder = new StringBuilder(); + final StringBuilder enoughMemoryMessageBuilder = new StringBuilder(); for (final Map.Entry<Integer, TPushPipeMetaResp> respEntry : respMap.entrySet()) { final int dataNodeId = respEntry.getKey(); final TPushPipeMetaResp resp = respEntry.getValue(); + if (resp.getStatus().getCode() + == TSStatusCode.PIPE_PUSH_META_NOT_ENOUGH_MEMORY.getStatusCode()) { + exceptionMessageBuilder.append(String.format("DataNodeId: %s,", dataNodeId)); + resp.getExceptionMessages() + .forEach( + message -> { + // Ignore the timeStamp for simplicity + if (pipeName == null) { + enoughMemoryMessageBuilder.append( + String.format( + "PipeName: %s, Message: %s", + message.getPipeName(), message.getMessage())); + } else if (pipeName.equals(message.getPipeName())) { + enoughMemoryMessageBuilder.append( + String.format("Message: %s", message.getMessage())); + } + }); + enoughMemoryMessageBuilder.append("."); + continue; + } + if (resp.getStatus().getCode() == TSStatusCode.PIPE_PUSH_META_TIMEOUT.getStatusCode()) { exceptionMessageBuilder.append( String.format( @@ -472,6 +494,12 @@ public abstract class AbstractOperatePipeProcedureV2 } } } + + final String enoughMemoryMessage = enoughMemoryMessageBuilder.toString(); + if (!enoughMemoryMessage.isEmpty()) { + throw new PipeException(enoughMemoryMessage); + } + return exceptionMessageBuilder.toString(); } 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 56e4ef9159c..1ac34aaae19 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 @@ -695,7 +695,8 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent { if (freeMemorySizeInBytes < needMemory + reservedMemorySizeInBytes) { final String message = String.format( - "Not enough memory for pipe. Need memory: %d bytes, free memory: %d bytes, reserved memory: %d bytes, total memory: %d bytes", + "%s Need memory: %d bytes, free memory: %d bytes, reserved memory: %d bytes, total memory: %d bytes", + MESSAGE_PIPE_NOT_ENOUGH_MEMORY, needMemory, freeMemorySizeInBytes, freeMemorySizeInBytes, @@ -733,8 +734,10 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent { if (remainingMemory < PipeConfig.getInstance().PipeInsertNodeQueueMemory()) { final String message = String.format( - "Not enough memory for pipe. Need Floating memory: %d bytes, free Floating memory: %d bytes", - PipeConfig.getInstance().PipeInsertNodeQueueMemory(), remainingMemory); + "%s Need Floating memory: %d bytes, free Floating memory: %d bytes", + MESSAGE_PIPE_NOT_ENOUGH_MEMORY, + PipeConfig.getInstance().PipeInsertNodeQueueMemory(), + remainingMemory); LOGGER.warn(message); throw new PipeException(message); } 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 5d7fa891530..d4c35b700cb 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 @@ -59,6 +59,7 @@ import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.path.PathDeserializeUtil; import org.apache.iotdb.commons.path.PathPatternTree; import org.apache.iotdb.commons.pipe.agent.plugin.meta.PipePluginMeta; +import org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta; import org.apache.iotdb.commons.schema.SchemaConstant; import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression; @@ -1124,12 +1125,22 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface } else { throw new Exception("Invalid TPushSinglePipeMetaReq"); } - return exceptionMessage == null - ? new TPushPipeMetaResp() - .setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())) - : new TPushPipeMetaResp() - .setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode())) + if (exceptionMessage != null) { + if (exceptionMessage.message != null + && exceptionMessage.message.contains(PipeTaskAgent.MESSAGE_PIPE_NOT_ENOUGH_MEMORY)) { + return new TPushPipeMetaResp() + .setStatus( + new TSStatus(TSStatusCode.PIPE_PUSH_META_NOT_ENOUGH_MEMORY.getStatusCode())) .setExceptionMessages(Collections.singletonList(exceptionMessage)); + } + + return new TPushPipeMetaResp() + .setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode())) + .setExceptionMessages(Collections.singletonList(exceptionMessage)); + } + + return new TPushPipeMetaResp() + .setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); } catch (Exception e) { LOGGER.error("Error occurred when pushing single pipe meta", e); return new TPushPipeMetaResp() 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 d1ea8807252..ca8ada69409 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 @@ -79,6 +79,7 @@ public abstract class PipeTaskAgent { private static final Logger LOGGER = LoggerFactory.getLogger(PipeTaskAgent.class); + public static final String MESSAGE_PIPE_NOT_ENOUGH_MEMORY = "Not enough memory for pipe."; protected static final String MESSAGE_UNKNOWN_PIPE_STATUS = "Unknown pipe status %s for pipe %s"; protected static final String MESSAGE_UNEXPECTED_PIPE_STATUS = "Unexpected pipe status %s: ";
