This is an automated email from the ASF dual-hosted git repository. rong pushed a commit to branch IOTDB-5787 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit df99304cfa408c6f3c8d32f1e761d39358a6503f Author: Steve Yurong Su <[email protected]> AuthorDate: Sun May 7 03:16:35 2023 +0800 DN: createPipe --- .../iotdb/db/pipe/agent/task/PipeTaskAgent.java | 33 +++++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java b/server/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java index 40a79cd3f26..b27dfe97f4e 100644 --- a/server/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java +++ b/server/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java @@ -42,20 +42,51 @@ public class PipeTaskAgent { final String pipeName = pipeMeta.getStaticMeta().getPipeName(); final long creationTime = pipeMeta.getStaticMeta().getCreationTime(); + // check if the pipe has already been created before final PipeMeta existedPipeMeta = pipeMetaKeeper.getPipeMeta(pipeName); if (existedPipeMeta != null) { if (existedPipeMeta.getStaticMeta().getCreationTime() == creationTime) { switch (existedPipeMeta.getRuntimeMeta().getStatus().get()) { case STOPPED: case RUNNING: + LOGGER.info( + "Pipe {} (creation time = {}) has already been created. Current status = {}. Skip creating.", + pipeName, + creationTime, + existedPipeMeta.getRuntimeMeta().getStatus().get().name()); + return; case DROPPED: + LOGGER.info( + "Pipe {} (creation time = {}) has already been dropped. Current status = {}. Recreating.", + pipeName, + creationTime, + existedPipeMeta.getRuntimeMeta().getStatus().get().name()); + // break to drop the pipe meta and recreate it + break; default: + throw new IllegalStateException( + "Unexpected status: " + existedPipeMeta.getRuntimeMeta().getStatus().get().name()); } - return; } + // drop the pipe if + // 1. the pipe with the same name but with different creation time has been created before + // 2. the pipe with the same name and the same creation time has been dropped before, but the + // pipe task meta has not been cleaned up dropPipe(pipeName, existedPipeMeta.getStaticMeta().getCreationTime()); } + + // build pipe task by consensus group + pipeMeta + .getRuntimeMeta() + .getConsensusGroupIdToTaskMetaMap() + .forEach( + ((consensusGroupId, pipeTaskMeta) -> { + createPipeTaskByConsensusGroup( + pipeName, creationTime, consensusGroupId, pipeTaskMeta); + })); + // add pipe meta to pipe meta keeper + pipeMetaKeeper.addPipeMeta(pipeMeta.getStaticMeta().getPipeName(), pipeMeta); } public void createPipeTaskByConsensusGroup(
