diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/DataNodeHeartbeatHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/DataNodeHeartbeatHandler.java index 9c7810dabe2d5..14e1dcbb86fd4 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/DataNodeHeartbeatHandler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/DataNodeHeartbeatHandler.java @@ -192,7 +192,8 @@ private void cachePipeHeartbeat(TDataNodeHeartbeatResp heartbeatResp) { heartbeatResp.getPipeRemainingEventCountList(), heartbeatResp.getPipeRemainingTimeList(), heartbeatResp.getPipeDegradedStatusList(), - heartbeatResp.getPipeRecentFailureList()); + heartbeatResp.getPipeRecentFailureList(), + heartbeatResp.getPipeCompletedDataRegionList()); } } 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 c0276fd2cd540..7ff219745408a 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 @@ -3339,7 +3339,8 @@ public TSStatus pushHeartbeat(final int dataNodeId, final TPipeHeartbeatResp res resp.getPipeRemainingEventCountList(), resp.getPipeRemainingTimeList(), resp.getPipeDegradedStatusList(), - resp.getPipeRecentFailureList()); + resp.getPipeRecentFailureList(), + resp.getPipeCompletedDataRegionList()); return StatusUtils.OK; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeRuntimeCoordinator.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeRuntimeCoordinator.java index ec00adcd30224..cb0da3e0948cc 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeRuntimeCoordinator.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeRuntimeCoordinator.java @@ -19,6 +19,7 @@ package org.apache.iotdb.confignode.manager.pipe.coordinator.runtime; +import org.apache.iotdb.common.rpc.thrift.TPipeCompletedDataRegion; import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.confignode.manager.ConfigManager; @@ -98,7 +99,8 @@ public void parseHeartbeat( /* @Nullable */ final List pipeRemainingEventCountListFromAgent, /* @Nullable */ final List pipeRemainingTimeListFromAgent, /* @Nullable */ final List pipeDegradedStatusListFromAgent, - /* @Nullable */ final List> pipeRecentFailureListFromAgent) { + /* @Nullable */ final List> pipeRecentFailureListFromAgent, + /* @Nullable */ final List pipeCompletedDataRegionListFromAgent) { pipeHeartbeatScheduler.parseHeartbeat( dataNodeId, new PipeHeartbeat( @@ -107,6 +109,7 @@ public void parseHeartbeat( pipeRemainingEventCountListFromAgent, pipeRemainingTimeListFromAgent, pipeDegradedStatusListFromAgent, - pipeRecentFailureListFromAgent)); + pipeRecentFailureListFromAgent, + pipeCompletedDataRegionListFromAgent)); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeat.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeat.java index 7aa75b2d78d79..dd31751bb17d9 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeat.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeat.java @@ -19,6 +19,7 @@ package org.apache.iotdb.confignode.manager.pipe.coordinator.runtime.heartbeat; +import org.apache.iotdb.common.rpc.thrift.TPipeCompletedDataRegion; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMeta; @@ -32,11 +33,11 @@ public class PipeHeartbeat { private final Map pipeMetaMap = new HashMap<>(); - private final Map isCompletedMap = new HashMap<>(); private final Map remainingEventCountMap = new HashMap<>(); private final Map remainingTimeMap = new HashMap<>(); private final Map isDegradedMap = new HashMap<>(); private final Map> recentFailuresMap = new HashMap<>(); + private final Map> completedDataRegionIdsMap = new HashMap<>(); public PipeHeartbeat( final List pipeMetaByteBufferListFromAgent, @@ -50,6 +51,7 @@ public PipeHeartbeat( pipeRemainingEventCountListFromAgent, pipeRemainingTimeListFromAgent, pipeDegradedStatusListFromAgent, + null, null); } @@ -60,6 +62,24 @@ public PipeHeartbeat( /* @Nullable */ final List pipeRemainingTimeListFromAgent, /* @Nullable */ final List pipeDegradedStatusListFromAgent, /* @Nullable */ final List> pipeRecentFailureListFromAgent) { + this( + pipeMetaByteBufferListFromAgent, + pipeCompletedListFromAgent, + pipeRemainingEventCountListFromAgent, + pipeRemainingTimeListFromAgent, + pipeDegradedStatusListFromAgent, + pipeRecentFailureListFromAgent, + null); + } + + public PipeHeartbeat( + final List pipeMetaByteBufferListFromAgent, + /* @Nullable */ final List pipeCompletedListFromAgent, + /* @Nullable */ final List pipeRemainingEventCountListFromAgent, + /* @Nullable */ final List pipeRemainingTimeListFromAgent, + /* @Nullable */ final List pipeDegradedStatusListFromAgent, + /* @Nullable */ final List> pipeRecentFailureListFromAgent, + /* @Nullable */ final List completedDataRegionListFromAgent) { // Shall not reach here, just in case if (Objects.isNull(pipeMetaByteBufferListFromAgent)) { return; @@ -68,11 +88,6 @@ public PipeHeartbeat( final PipeMeta pipeMeta = PipeMeta.deserialize4TaskAgent(pipeMetaByteBufferListFromAgent.get(i)); pipeMetaMap.put(pipeMeta.getStaticMeta(), pipeMeta); - isCompletedMap.put( - pipeMeta.getStaticMeta(), - Objects.nonNull(pipeCompletedListFromAgent) - && i < pipeCompletedListFromAgent.size() - && pipeCompletedListFromAgent.get(i)); // If remaining event count & remaining time can not be got, it implies that the heartbeat is // from an ancient version of DataNode. Here we guarantee that "0" will not affect both of // the final results and namely these dataNodes are omitted in calculation. @@ -102,6 +117,17 @@ public PipeHeartbeat( && Objects.nonNull(pipeRecentFailureListFromAgent.get(i)) ? new HashMap<>(pipeRecentFailureListFromAgent.get(i)) : Collections.emptyMap()); + if (completedDataRegionListFromAgent != null) { + for (final TPipeCompletedDataRegion completedDataRegion : + completedDataRegionListFromAgent) { + if (pipeMeta.getStaticMeta().getPipeName().equals(completedDataRegion.getPipeName()) + && pipeMeta.getStaticMeta().getCreationTime() + == completedDataRegion.getCreationTime()) { + completedDataRegionIdsMap.put( + pipeMeta.getStaticMeta(), completedDataRegion.getCompletedDataRegionIds()); + } + } + } } } @@ -113,8 +139,12 @@ public PipeMeta getPipeMeta(final PipeStaticMeta pipeStaticMeta) { return pipeMetaMap.get(pipeStaticMeta); } - public Boolean isCompleted(final PipeStaticMeta pipeStaticMeta) { - return isCompletedMap.get(pipeStaticMeta); + public List getCompletedDataRegionIds(final PipeStaticMeta pipeStaticMeta) { + return completedDataRegionIdsMap.getOrDefault(pipeStaticMeta, Collections.emptyList()); + } + + public boolean hasCompletedDataRegionReport(final PipeStaticMeta pipeStaticMeta) { + return completedDataRegionIdsMap.containsKey(pipeStaticMeta); } public Long getRemainingEventCount(final PipeStaticMeta pipeStaticMeta) { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java index a8734469d0c8d..1149c8e44442a 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java @@ -19,6 +19,8 @@ package org.apache.iotdb.confignode.manager.pipe.coordinator.runtime.heartbeat; +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; import org.apache.iotdb.commons.consensus.index.ProgressIndex; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException; @@ -39,6 +41,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.HashSet; import java.util.Map; import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; @@ -156,41 +159,48 @@ private void parseHeartbeatAndSaveMetaChangeLocally( final PipeTemporaryMetaInCoordinator temporaryMeta = (PipeTemporaryMetaInCoordinator) pipeMetaFromCoordinator.getTemporaryMeta(); - // Remove completed pipes - final Boolean isPipeCompletedFromAgent = pipeHeartbeat.isCompleted(staticMeta); - if (Boolean.TRUE.equals(isPipeCompletedFromAgent)) { + // Aggregate completed DataRegion ids reported by DataNodes. Only the DataNodes that own the + // target region can report it, so the coordinator can compare the union against all required + // DataRegion ids without trusting any DataNode's single per-pipe completion boolean. + if (pipeHeartbeat.hasCompletedDataRegionReport(staticMeta)) { + for (final Integer completedDataRegionId : + pipeHeartbeat.getCompletedDataRegionIds(staticMeta)) { + temporaryMeta.markDataRegionCompleted(completedDataRegionId); + } + } + + final Set requiredDataRegionIds = new HashSet<>(); + for (final Map.Entry entry : + pipeMetaFromCoordinator.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().entrySet()) { + if (configManager + .getPartitionManager() + .isRegionGroupExists( + new TConsensusGroupId(TConsensusGroupType.DataRegion, entry.getKey()))) { + requiredDataRegionIds.add(entry.getKey()); + } + } - temporaryMeta.markDataNodeCompleted(nodeId); + // Remove completed pipes only when every required DataRegion has been reported complete. + // Relying on the region-level reports (instead of the DataNode-level boolean) prevents a + // leader-change / task-creation failure from being treated as a successful snapshot transfer. + if (!requiredDataRegionIds.isEmpty() + && temporaryMeta.getCompletedDataRegionIds().containsAll(requiredDataRegionIds)) { PipeLogger.log( LOGGER::info, - ManagerMessages.DETECTED_HISTORICAL_PIPE_COMPLETION_REPORT_FROM_DATANODE, - nodeId, + ManagerMessages.ALL_DATANODES_REPORTED_HISTORICAL_PIPE_COMPLETED, staticMeta.getPipeName(), - pipeHeartbeat.getRemainingEventCount(staticMeta), - pipeHeartbeat.getRemainingTime(staticMeta), - temporaryMeta.getCompletedDataNodeIds()); - - final Set uncompletedDataNodeIds = - configManager.getNodeManager().getRegisteredDataNodeLocations().keySet(); - uncompletedDataNodeIds.removeAll(temporaryMeta.getCompletedDataNodeIds()); - if (uncompletedDataNodeIds.isEmpty()) { - PipeLogger.log( - LOGGER::info, - ManagerMessages.ALL_DATANODES_REPORTED_HISTORICAL_PIPE_COMPLETED, - staticMeta.getPipeName(), - temporaryMeta.getGlobalRemainingEvents(), - temporaryMeta.getGlobalRemainingTime(), - staticMeta); - pipeTaskInfo.get().removePipeMeta(staticMeta); - PipeLogger.log( - LOGGER::info, - ManagerMessages.DETECTED_COMPLETION_OF_PIPE_STATIC_META_REMOVE_IT, - staticMeta.getPipeName(), - staticMeta); - needWriteConsensusOnConfigNodes.set(true); - needPushPipeMetaToDataNodes.set(true); - continue; - } + temporaryMeta.getGlobalRemainingEvents(), + temporaryMeta.getGlobalRemainingTime(), + staticMeta); + pipeTaskInfo.get().removePipeMeta(staticMeta); + PipeLogger.log( + LOGGER::info, + ManagerMessages.DETECTED_COMPLETION_OF_PIPE_STATIC_META_REMOVE_IT, + staticMeta.getPipeName(), + staticMeta); + needWriteConsensusOnConfigNodes.set(true); + needPushPipeMetaToDataNodes.set(true); + continue; } // Record statistics diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatScheduler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatScheduler.java index 209b08cff1557..79d944ad87bab 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatScheduler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatScheduler.java @@ -117,7 +117,8 @@ private synchronized void heartbeat() { resp.getPipeRemainingEventCountList(), resp.getPipeRemainingTimeList(), resp.getPipeDegradedStatusList(), - resp.getPipeRecentFailureList()))); + resp.getPipeRecentFailureList(), + resp.getPipeCompletedDataRegionList()))); // config node heartbeat try { 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 978567110501e..2427e7f3329af 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 @@ -34,7 +34,6 @@ 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.agent.task.meta.PipeTemporaryMeta; -import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMetaInCoordinator; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeType; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant; @@ -917,9 +916,6 @@ private TSStatus handleLeaderChangeInternal(final PipeHandleLeaderChangePlan pla consensusGroupIdToTaskMetaMap .get(consensusGroupId.getId()) .setLeaderNodeId(newLeader); - // New region leader may contain un-transferred events - ((PipeTemporaryMetaInCoordinator) pipeMeta.getTemporaryMeta()) - .markDataNodeUncompleted(newLeader); } else { consensusGroupIdToTaskMetaMap.remove(consensusGroupId.getId()); } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java index c7d4e3b5d8d2d..95c29a5633661 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java @@ -19,6 +19,8 @@ package org.apache.iotdb.confignode.manager.pipe.coordinator.runtime.heartbeat; +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; +import org.apache.iotdb.common.rpc.thrift.TPipeCompletedDataRegion; import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException; @@ -33,6 +35,7 @@ import org.apache.iotdb.confignode.manager.ConfigManager; import org.apache.iotdb.confignode.manager.ProcedureManager; import org.apache.iotdb.confignode.manager.node.NodeManager; +import org.apache.iotdb.confignode.manager.partition.PartitionManager; import org.apache.iotdb.confignode.manager.pipe.coordinator.PipeManager; import org.apache.iotdb.confignode.manager.pipe.coordinator.runtime.PipeRuntimeCoordinator; import org.apache.iotdb.confignode.manager.pipe.coordinator.task.PipeTaskCoordinator; @@ -47,6 +50,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.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -360,6 +364,105 @@ public void testPipeHeartbeatTreatsNullRecentFailureMapAsEmpty() throws Exceptio Assert.assertTrue(heartbeat.getRecentFailures(pipeMeta.getStaticMeta()).isEmpty()); } + @Test + public void testParseHeartbeatDoesNotCompleteWhenRequiredDataRegionMissing() throws Exception { + CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false); + + final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo(); + final PipeMeta pipeMeta = createPipeMeta(); + pipeTaskInfo.createPipe( + new CreatePipePlanV2(pipeMeta.getStaticMeta(), pipeMeta.getRuntimeMeta())); + + final ParserTestContext context = createParserTestContext(1, pipeTaskInfo); + context.parser.parseHeartbeat( + 1, + new PipeHeartbeat( + Collections.singletonList(pipeMeta.serialize()), + Collections.singletonList(true), + Collections.singletonList(0L), + Collections.singletonList(0d), + null, + null, + Collections.singletonList( + new TPipeCompletedDataRegion( + pipeMeta.getStaticMeta().getPipeName(), + pipeMeta.getStaticMeta().getCreationTime(), + Collections.emptyList())))); + + Assert.assertTrue(getTemporaryMeta(pipeTaskInfo).getCompletedDataRegionIds().isEmpty()); + Assert.assertNotNull(pipeTaskInfo.getPipeMetaByPipeName("test_pipe")); + verify(context.procedureManager, never()).pipeHandleMetaChange(anyBoolean(), anyBoolean()); + } + + @Test + public void testParseHeartbeatDoesNotTrustDataNodeBooleanForCompletion() throws Exception { + CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false); + + final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo(); + final PipeMeta pipeMeta = createPipeMeta(1); + pipeTaskInfo.createPipe( + new CreatePipePlanV2(pipeMeta.getStaticMeta(), pipeMeta.getRuntimeMeta())); + + final ParserTestContext context = createParserTestContext(1, pipeTaskInfo); + // The DataNode's boolean is false, but the required DataRegion is reported complete. The + // coordinator should still complete the pipe because it no longer trusts the boolean. + context.parser.parseHeartbeat( + 1, createPipeHeartbeatWithCompletedRegions(pipeMeta, false, Collections.singletonList(1))); + + Assert.assertNull(pipeTaskInfo.getPipeMetaByPipeName("test_pipe")); + } + + @Test + public void testParseHeartbeatCompletesOnlyAfterAllRequiredDataRegionsReported() + throws Exception { + CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false); + + final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo(); + final PipeMeta pipeMeta = createPipeMeta(1, 2); + pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().get(2).setLeaderNodeId(2); + pipeTaskInfo.createPipe( + new CreatePipePlanV2(pipeMeta.getStaticMeta(), pipeMeta.getRuntimeMeta())); + + final ParserTestContext context = createParserTestContext(2, pipeTaskInfo); + + context.parser.parseHeartbeat( + 1, createPipeHeartbeatWithCompletedRegions(pipeMeta, true, Collections.singletonList(1))); + Assert.assertNotNull(pipeTaskInfo.getPipeMetaByPipeName("test_pipe")); + + context.parser.parseHeartbeat( + 2, createPipeHeartbeatWithCompletedRegions(pipeMeta, true, Collections.singletonList(2))); + Assert.assertNull(pipeTaskInfo.getPipeMetaByPipeName("test_pipe")); + // After CN decides the pipe is complete, the next heartbeat round pushes the updated meta so + // DataNodes will drop their local pipe tasks. + verify(context.procedureManager, times(1)).pipeHandleMetaChange(true, true); + } + + @Test + public void testParseHeartbeatKeepsCompletedDataRegionAfterLeaderChange() throws Exception { + CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false); + + final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo(); + final PipeMeta pipeMeta = createPipeMeta(1, 2); + pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().get(2).setLeaderNodeId(2); + pipeTaskInfo.createPipe( + new CreatePipePlanV2(pipeMeta.getStaticMeta(), pipeMeta.getRuntimeMeta())); + + final ParserTestContext context = createParserTestContext(2, pipeTaskInfo); + + // The old leader of region 1 reports it complete before the leader changes. + context.parser.parseHeartbeat( + 1, createPipeHeartbeatWithCompletedRegions(pipeMeta, true, Collections.singletonList(1))); + Assert.assertNotNull(pipeTaskInfo.getPipeMetaByPipeName("test_pipe")); + + // Region 1's leader moves to node 2, which only reports region 2. Region 1's completion is + // still valid because its historical data was already transferred by the old leader. + pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().get(1).setLeaderNodeId(2); + context.parser.parseHeartbeat( + 2, createPipeHeartbeatWithCompletedRegions(pipeMeta, true, Collections.singletonList(2))); + + Assert.assertNull(pipeTaskInfo.getPipeMetaByPipeName("test_pipe")); + } + private ParserTestContext createParserTestContext(final int registeredDataNodeCount) { return createParserTestContext(registeredDataNodeCount, new PipeTaskInfo()); } @@ -373,6 +476,7 @@ private ParserTestContext createParserTestContext( final PipeRuntimeCoordinator pipeRuntimeCoordinator = Mockito.mock(PipeRuntimeCoordinator.class); final PipeTaskCoordinator pipeTaskCoordinator = Mockito.mock(PipeTaskCoordinator.class); + final PartitionManager partitionManager = Mockito.mock(PartitionManager.class); final ExecutorService procedureSubmitter = Mockito.mock(ExecutorService.class); when(configManager.getNodeManager()).thenReturn(nodeManager); @@ -382,6 +486,8 @@ private ParserTestContext createParserTestContext( when(pipeManager.getPipeRuntimeCoordinator()).thenReturn(pipeRuntimeCoordinator); when(pipeManager.getPipeTaskCoordinator()).thenReturn(pipeTaskCoordinator); when(pipeRuntimeCoordinator.getProcedureSubmitter()).thenReturn(procedureSubmitter); + when(configManager.getPartitionManager()).thenReturn(partitionManager); + when(partitionManager.isRegionGroupExists(any(TConsensusGroupId.class))).thenReturn(true); when(pipeTaskCoordinator.tryLock()).thenReturn(new AtomicReference<>(pipeTaskInfo)); when(procedureManager.pipeHandleMetaChange(anyBoolean(), anyBoolean())).thenReturn(true); Mockito.doAnswer( @@ -467,15 +573,38 @@ private void setAtomicBooleanField( } private PipeMeta createPipeMeta() { + return createPipeMeta(1); + } + + private PipeMeta createPipeMeta(final int... regionIds) { final PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta(); - pipeRuntimeMeta - .getConsensusGroupId2TaskMetaMap() - .put(1, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)); + for (final int regionId : regionIds) { + pipeRuntimeMeta + .getConsensusGroupId2TaskMetaMap() + .put(regionId, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)); + } return new PipeMeta( new PipeStaticMeta("test_pipe", 1L, new HashMap<>(), new HashMap<>(), new HashMap<>()), pipeRuntimeMeta); } + private PipeHeartbeat createPipeHeartbeatWithCompletedRegions( + final PipeMeta pipeMeta, final boolean isCompleted, final List completedRegionIds) + throws Exception { + return new PipeHeartbeat( + Collections.singletonList(pipeMeta.serialize()), + Collections.singletonList(isCompleted), + Collections.singletonList(0L), + Collections.singletonList(0d), + null, + null, + Collections.singletonList( + new TPipeCompletedDataRegion( + pipeMeta.getStaticMeta().getPipeName(), + pipeMeta.getStaticMeta().getCreationTime(), + completedRegionIds))); + } + private PipeHeartbeat emptyHeartbeat() { return new PipeHeartbeat(Collections.emptyList(), null, null, null, null); } 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 5241ea0c916d7..1f6416ad20fb9 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 @@ -19,6 +19,7 @@ package org.apache.iotdb.db.pipe.agent.task; +import org.apache.iotdb.common.rpc.thrift.TPipeCompletedDataRegion; import org.apache.iotdb.common.rpc.thrift.TPipeHeartbeatResp; import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.concurrent.IoTThreadFactory; @@ -501,7 +502,7 @@ private void collectPipeMetaListInternal(final TDataNodeHeartbeatResp resp) thro PipeConfig.getInstance().getPipeMetaReportMaxLogIntervalRounds(), pipeMetaKeeper.getPipeMetaCount()); - collectPipeMetaReport(logger, true).setTo(resp); + collectPipeMetaReport(logger).setTo(resp); PipeInsertionDataNodeListener.getInstance().listenToHeartbeat(true); } @@ -523,17 +524,11 @@ protected void collectPipeMetaListInternal( LOGGER.debug( DataNodePipeMessages.RECEIVED_PIPE_HEARTBEAT_REQUEST_FROM_CONFIG_NODE, req.heartbeatId); - collectPipeMetaReport(logger, false).setTo(resp); + collectPipeMetaReport(logger).setTo(resp); PipeInsertionDataNodeListener.getInstance().listenToHeartbeat(true); } - private PipeMetaReport collectPipeMetaReport( - final Optional logger, final boolean includeQueryMode) throws TException { - final Set dataRegionIds = - StorageEngine.getInstance().getAllDataRegionIds().stream() - .map(DataRegionId::getId) - .collect(Collectors.toSet()); - + private PipeMetaReport collectPipeMetaReport(final Optional logger) throws TException { final PipeMetaReport report = new PipeMetaReport(); try { for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) { @@ -542,17 +537,23 @@ private PipeMetaReport collectPipeMetaReport( final PipeStaticMeta staticMeta = pipeMeta.getStaticMeta(); final Map pipeTaskMap = pipeTaskManager.getPipeTasks(staticMeta); - final boolean isAllDataRegionCompleted = - pipeTaskMap == null - || pipeTaskMap.entrySet().stream() - .filter(entry -> dataRegionIds.contains(entry.getKey())) - .allMatch(entry -> ((PipeDataNodeTask) entry.getValue()).isCompleted()); - final boolean isCompleted = - isAllDataRegionCompleted && includeDataAndNeedDrop(pipeMeta, includeQueryMode); + final Set expectedDataRegionIds = getExpectedDataRegionIds(pipeMeta); + final List completedDataRegionIds = new ArrayList<>(); + if (pipeTaskMap != null) { + for (final Integer regionId : expectedDataRegionIds) { + final PipeTask pipeTask = pipeTaskMap.get(regionId); + if (pipeTask instanceof PipeDataNodeTask + && ((PipeDataNodeTask) pipeTask).isCompleted()) { + completedDataRegionIds.add(regionId); + } + } + } + report.pipeCompletedDataRegionList.add( + new TPipeCompletedDataRegion( + staticMeta.getPipeName(), staticMeta.getCreationTime(), completedDataRegionIds)); final Pair remainingEventAndTime = PipeDataNodeSinglePipeMetrics.getInstance() .getRemainingEventAndTime(staticMeta.getPipeName(), staticMeta.getCreationTime()); - report.pipeCompletedList.add(isCompleted); report.pipeRemainingEventCountList.add(remainingEventAndTime.getLeft()); report.pipeRemainingTimeList.add(remainingEventAndTime.getRight()); report.pipeDegradedStatusList.add( @@ -561,16 +562,6 @@ private PipeMetaReport collectPipeMetaReport( .getGlobalTsFileEpochDegraded())); report.pipeRecentFailureList.add( ((PipeTemporaryMetaInAgent) pipeMeta.getTemporaryMeta()).getRecentFailures()); - - logger.ifPresent( - l -> - PipeLogger.log( - l::info, - DataNodePipeMessages - .LOG_REPORTING_PIPE_META_ARG_ISCOMPLETED_ARG_REMAININGEVENTCOUNT_ARG_8F996DF3, - pipeMeta.coreReportMessage(), - isCompleted, - remainingEventAndTime.getLeft())); } logger.ifPresent( l -> @@ -578,56 +569,68 @@ private PipeMetaReport collectPipeMetaReport( l::info, DataNodePipeMessages.LOG_REPORTED_ARG_PIPE_METAS_12068FC6, report.pipeMetaBinaryList.size())); - } catch (final IOException | IllegalPathException e) { + } catch (final IOException e) { throw new TException(e); } return report; } - private boolean includeDataAndNeedDrop(final PipeMeta pipeMeta, final boolean includeQueryMode) - throws IllegalPathException { - final PipeParameters sourceParameters = pipeMeta.getStaticMeta().getSourceParameters(); - if (!DataRegionListeningFilter.parseInsertionDeletionListeningOptionPair(sourceParameters) - .getLeft()) { - return false; - } - if (!includeQueryMode) { - return isSnapshotMode(sourceParameters); + // Returns the DataRegion ids that this DataNode is expected to transfer for the given pipe. + // A region is included only when it is owned by this DataNode, is led by this DataNode according + // to the pipe's runtime metadata, and is selected by the pipe's source parameters. This expected + // set is used instead of the already-created PipeTask map so that a failed task initialization is + // not silently treated as a completed region. + private Set getExpectedDataRegionIds(final PipeMeta pipeMeta) { + final PipeStaticMeta staticMeta = pipeMeta.getStaticMeta(); + final PipeParameters sourceParameters = staticMeta.getSourceParameters(); + final Set localDataRegionIds = + StorageEngine.getInstance().getAllDataRegionIds().stream() + .map(DataRegionId::getId) + .collect(Collectors.toSet()); + final Set expectedDataRegionIds = new HashSet<>(); + for (final Map.Entry entry : + pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().entrySet()) { + final int regionId = entry.getKey(); + if (entry.getValue().getLeaderNodeId() != CONFIG.getDataNodeId() + || !localDataRegionIds.contains(regionId)) { + continue; + } + try { + if (DataRegionListeningFilter.shouldDataRegionBeListened( + sourceParameters, new DataRegionId(regionId), staticMeta.getPipeType())) { + expectedDataRegionIds.add(regionId); + } + } catch (final IllegalPathException e) { + throw new PipeException(e.toString()); + } } - - final String sourceModeValue = - sourceParameters.getStringOrDefault( - Arrays.asList( - PipeSourceConstant.EXTRACTOR_MODE_KEY, PipeSourceConstant.SOURCE_MODE_KEY), - PipeSourceConstant.EXTRACTOR_MODE_DEFAULT_VALUE); - return sourceModeValue.equalsIgnoreCase(PipeSourceConstant.EXTRACTOR_MODE_QUERY_VALUE) - || sourceModeValue.equalsIgnoreCase(PipeSourceConstant.EXTRACTOR_MODE_SNAPSHOT_VALUE); + return expectedDataRegionIds; } private static class PipeMetaReport { private final List pipeMetaBinaryList = new ArrayList<>(); - private final List pipeCompletedList = new ArrayList<>(); private final List pipeRemainingEventCountList = new ArrayList<>(); private final List pipeRemainingTimeList = new ArrayList<>(); private final List pipeDegradedStatusList = new ArrayList<>(); private final List> pipeRecentFailureList = new ArrayList<>(); + private final List pipeCompletedDataRegionList = new ArrayList<>(); private void setTo(final TDataNodeHeartbeatResp resp) { resp.setPipeMetaList(pipeMetaBinaryList); - resp.setPipeCompletedList(pipeCompletedList); resp.setPipeRemainingEventCountList(pipeRemainingEventCountList); resp.setPipeRemainingTimeList(pipeRemainingTimeList); resp.setPipeDegradedStatusList(pipeDegradedStatusList); resp.setPipeRecentFailureList(pipeRecentFailureList); + resp.setPipeCompletedDataRegionList(pipeCompletedDataRegionList); } private void setTo(final TPipeHeartbeatResp resp) { resp.setPipeMetaList(pipeMetaBinaryList); - resp.setPipeCompletedList(pipeCompletedList); resp.setPipeRemainingEventCountList(pipeRemainingEventCountList); resp.setPipeRemainingTimeList(pipeRemainingTimeList); resp.setPipeDegradedStatusList(pipeDegradedStatusList); resp.setPipeRecentFailureList(pipeRecentFailureList); + resp.setPipeCompletedDataRegionList(pipeCompletedDataRegionList); } } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTemporaryMetaInCoordinator.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTemporaryMetaInCoordinator.java index 5c3a1ea17eb24..79fc8b6a5b2f9 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTemporaryMetaInCoordinator.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTemporaryMetaInCoordinator.java @@ -33,7 +33,7 @@ public class PipeTemporaryMetaInCoordinator implements PipeTemporaryMeta { // ConfigNode statistics - private final Set completedDataNodeIds = + private final Set completedDataRegionIds = Collections.newSetFromMap(new ConcurrentHashMap<>()); private final ConcurrentMap nodeId2RemainingEventMap = new ConcurrentHashMap<>(); private final ConcurrentMap nodeId2RemainingTimeMap = new ConcurrentHashMap<>(); @@ -41,12 +41,12 @@ public class PipeTemporaryMetaInCoordinator implements PipeTemporaryMeta { private final ConcurrentMap nodeId2RecentFailuresMap = new ConcurrentHashMap<>(); - public void markDataNodeCompleted(final int dataNodeId) { - completedDataNodeIds.add(dataNodeId); + public void markDataRegionCompleted(final int dataRegionId) { + completedDataRegionIds.add(dataRegionId); } - public void markDataNodeUncompleted(final int dataNodeId) { - completedDataNodeIds.remove(dataNodeId); + public Set getCompletedDataRegionIds() { + return completedDataRegionIds; } public void setRemainingEvent(final int dataNodeId, final long remainingEventCount) { @@ -86,10 +86,6 @@ public void setRecentFailures(final int dataNodeId, final Map rece } } - public Set getCompletedDataNodeIds() { - return completedDataNodeIds; - } - public long getGlobalRemainingEvents() { return nodeId2RemainingEventMap.values().stream().reduce(Long::sum).orElse(0L); } @@ -131,7 +127,7 @@ public boolean equals(final Object o) { return false; } final PipeTemporaryMetaInCoordinator that = (PipeTemporaryMetaInCoordinator) o; - return Objects.equals(this.completedDataNodeIds, that.completedDataNodeIds) + return Objects.equals(this.completedDataRegionIds, that.completedDataRegionIds) && Objects.equals(this.nodeId2RemainingEventMap, that.nodeId2RemainingEventMap) && Objects.equals(this.nodeId2RemainingTimeMap, that.nodeId2RemainingTimeMap) && Objects.equals(this.nodeId2IsDegradedMap, that.nodeId2IsDegradedMap) @@ -141,7 +137,7 @@ public boolean equals(final Object o) { @Override public int hashCode() { return Objects.hash( - completedDataNodeIds, + completedDataRegionIds, nodeId2RemainingEventMap, nodeId2RemainingTimeMap, nodeId2IsDegradedMap, @@ -151,8 +147,8 @@ public int hashCode() { @Override public String toString() { return "PipeTemporaryMeta{" - + "completedDataNodeIds=" - + completedDataNodeIds + + "completedDataRegionIds=" + + completedDataRegionIds + ", nodeId2RemainingEventMap=" + nodeId2RemainingEventMap + ", nodeId2RemainingTimeMap=" diff --git a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift index edc824a6b433c..16b40a853541c 100644 --- a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift +++ b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift @@ -197,6 +197,12 @@ struct TSetThrottleQuotaReq { 2: required TThrottleQuota throttleQuota } +struct TPipeCompletedDataRegion { + 1: required string pipeName + 2: required i64 creationTime + 3: required list completedDataRegionIds +} + struct TPipeHeartbeatResp { 1: required list pipeMetaList 2: optional list pipeCompletedList @@ -204,6 +210,7 @@ struct TPipeHeartbeatResp { 4: optional list pipeRemainingTimeList 5: optional list pipeDegradedStatusList 6: optional list> pipeRecentFailureList + 7: optional list pipeCompletedDataRegionList } struct TLicense { diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift index d66ed10ccf9cd..1c3fd78470d3c 100644 --- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift +++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift @@ -319,6 +319,7 @@ struct TDataNodeHeartbeatResp { 17: optional map dataRegionRawDataSize 18: optional list pipeDegradedStatusList 19: optional list> pipeRecentFailureList + 20: optional list pipeCompletedDataRegionList } struct TPipeHeartbeatReq {