diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java index 28928653ef96..2d1d7d51f336 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java @@ -137,6 +137,87 @@ public void testAutoDropInHistoricalTransfer() throws Exception { } } + @Test + public void testAutoDropFinitePipesWithoutDataRegion() throws Exception { + final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0); + final Map sinkAttributes = new HashMap<>(); + sinkAttributes.put("sink", "iotdb-thrift-sink"); + sinkAttributes.put("sink.batch.enable", "false"); + sinkAttributes.put("sink.ip", receiverDataNode.getIp()); + sinkAttributes.put("sink.port", Integer.toString(receiverDataNode.getPort())); + + try (final SyncConfigNodeIServiceClient client = + (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) { + final Map querySourceAttributes = new HashMap<>(); + querySourceAttributes.put("source.mode", "query"); + querySourceAttributes.put("user", SessionConfig.DEFAULT_USER); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + client + .createPipe( + new TCreatePipeReq("query_pipe_without_data_region", sinkAttributes) + .setExtractorAttributes(querySourceAttributes)) + .getCode()); + + final Map historySourceAttributes = new HashMap<>(); + historySourceAttributes.put("source.realtime.enable", Boolean.FALSE.toString()); + historySourceAttributes.put("user", SessionConfig.DEFAULT_USER); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + client + .createPipe( + new TCreatePipeReq("history_pipe_without_data_region", sinkAttributes) + .setExtractorAttributes(historySourceAttributes)) + .getCode()); + + final Map realtimeSourceAttributes = new HashMap<>(); + realtimeSourceAttributes.put("source.history.enable", Boolean.FALSE.toString()); + realtimeSourceAttributes.put("user", SessionConfig.DEFAULT_USER); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + client + .createPipe( + new TCreatePipeReq("realtime_pipe_without_data_region", sinkAttributes) + .setExtractorAttributes(realtimeSourceAttributes)) + .getCode()); + + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + client.startPipe("query_pipe_without_data_region").getCode()); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + client.startPipe("history_pipe_without_data_region").getCode()); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + client.startPipe("realtime_pipe_without_data_region").getCode()); + + await() + .pollInSameThread() + .pollInterval(1L, TimeUnit.SECONDS) + .atMost(600, TimeUnit.SECONDS) + .untilAsserted( + () -> { + final List pipeInfoList = + client.showPipe(new TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER)) + .pipeInfoList; + Assert.assertFalse( + pipeInfoList.stream() + .anyMatch(info -> info.getId().equals("query_pipe_without_data_region"))); + Assert.assertFalse( + pipeInfoList.stream() + .anyMatch(info -> info.getId().equals("history_pipe_without_data_region"))); + Assert.assertTrue( + pipeInfoList.stream() + .anyMatch( + info -> info.getId().equals("realtime_pipe_without_data_region"))); + }); + + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + client.dropPipe("realtime_pipe_without_data_region").getCode()); + } + } + @Test public void testAutoDropIgnoredUnmatchedDataRegions() throws Exception { final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0); diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipePermissionIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipePermissionIT.java index 45fa762b02bf..96870b0329b9 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipePermissionIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipePermissionIT.java @@ -37,7 +37,11 @@ import java.sql.SQLException; import java.sql.Statement; import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import static org.awaitility.Awaitility.await; import static org.junit.Assert.fail; @RunWith(IoTDBTestRunner.class) @@ -257,4 +261,66 @@ public void testSinkPermissionWithHistoricalDataAndTablePattern() { TableModelUtils.assertCountData("test", "test", 0, env); TableModelUtils.assertCountData("test", "test1", 100, env); } + + @Test + public void testAuthenticatedRealtimePipesRemainWithoutDataRegion() throws Exception { + final Set expectedPipeNames = + new HashSet<>(Arrays.asList("root_pipe", "source_pipe", "sink_pipe", "source_sink_pipe")); + + try (final Connection connection = env.getConnection(BaseEnv.TABLE_SQL_DIALECT); + final Statement statement = connection.createStatement()) { + statement.execute("CREATE USER user_source 'paSs1234@56789'"); + statement.execute("CREATE USER user_sink 'paSs1234@56789'"); + statement.execute("CREATE DATABASE test_pipe_authentication"); + + statement.execute( + "create pipe root_pipe " + + "with source ('forwarding-pipe-requests'='false', " + + "'database-name'='test_pipe_authentication', 'table-name'='table_0') " + + "with processor ('processor'='rename-database-processor', " + + "'new-db-name'='pipe_newDB1') " + + "with sink ('sink'='write-back-sink')"); + statement.execute( + "create pipe source_pipe " + + "with source ('forwarding-pipe-requests'='false', " + + "'database-name'='test_pipe_authentication', 'table-name'='table_0', " + + "'user'='user_source', 'password'='paSs1234@56789') " + + "with processor ('processor'='rename-database-processor', " + + "'new-db-name'='pipe_newDB2') " + + "with sink ('sink'='write-back-sink')"); + statement.execute( + "create pipe sink_pipe " + + "with source ('forwarding-pipe-requests'='false', " + + "'database-name'='test_pipe_authentication', 'table-name'='table_0') " + + "with processor ('processor'='rename-database-processor', " + + "'new-db-name'='pipe_newDB3') " + + "with sink ('sink'='write-back-sink', 'user'='user_sink', " + + "'password'='paSs1234@56789')"); + statement.execute( + "create pipe source_sink_pipe " + + "with source ('forwarding-pipe-requests'='false', " + + "'database-name'='test_pipe_authentication', 'table-name'='table_0', " + + "'user'='user_source', 'password'='paSs1234@56789') " + + "with processor ('processor'='rename-database-processor', " + + "'new-db-name'='pipe_newDB4') " + + "with sink ('sink'='write-back-sink', 'user'='user_sink', " + + "'password'='paSs1234@56789')"); + + await() + .pollInSameThread() + .pollInterval(1, TimeUnit.SECONDS) + .during(15, TimeUnit.SECONDS) + .atMost(30, TimeUnit.SECONDS) + .untilAsserted( + () -> { + final Set actualPipeNames = new HashSet<>(); + try (final ResultSet resultSet = statement.executeQuery("SHOW PIPES")) { + while (resultSet.next()) { + actualPipeNames.add(resultSet.getString("ID")); + } + } + Assert.assertEquals(expectedPipeNames, actualPipeNames); + }); + } + } } 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 db3d67e86c3a..f702312e5dc7 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 @@ -25,6 +25,7 @@ import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkCriticalException; +import org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent; 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; @@ -178,10 +179,18 @@ private void parseHeartbeatAndSaveMetaChangeLocally( final Set requiredDataRegionIds = collectRequiredDataRegionIds(pipeMetaFromCoordinator); - // 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() + // A history-only internal pipe is finite and may be removed when all required DataRegions + // complete, or when CN determines that no DataRegion matched at creation time. An explicit + // region-level report proves that a DataNode has received the pipe meta, preventing an empty + // task map from completing the pipe before its initial push. Realtime and external-source + // pipes must remain alive because they may receive work in the future. + final boolean isFiniteInternalPipe = + !staticMeta.isSourceExternal() + && PipeTaskAgent.isHistoryOnlyPipe(staticMeta.getSourceParameters()); + final boolean hasReliableDataRegionReport = + pipeHeartbeat.hasCompletedDataRegionReport(staticMeta); + if (isFiniteInternalPipe + && hasReliableDataRegionReport && temporaryMeta.getCompletedDataRegionIds().containsAll(requiredDataRegionIds)) { PipeLogger.log( LOGGER::info, 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 95c29a563366..c36a09a7def8 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 @@ -369,7 +369,7 @@ public void testParseHeartbeatDoesNotCompleteWhenRequiredDataRegionMissing() thr CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false); final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo(); - final PipeMeta pipeMeta = createPipeMeta(); + final PipeMeta pipeMeta = createHistoryOnlyPipeMeta(1); pipeTaskInfo.createPipe( new CreatePipePlanV2(pipeMeta.getStaticMeta(), pipeMeta.getRuntimeMeta())); @@ -394,12 +394,67 @@ public void testParseHeartbeatDoesNotCompleteWhenRequiredDataRegionMissing() thr verify(context.procedureManager, never()).pipeHandleMetaChange(anyBoolean(), anyBoolean()); } + @Test + public void testParseHeartbeatCompletesHistoryOnlyPipeWithoutRequiredDataRegion() + throws Exception { + CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false); + + final Map sourceAttributes = new HashMap<>(); + sourceAttributes.put("source.realtime.enable", Boolean.FALSE.toString()); + final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo(); + final PipeMeta pipeMeta = createPipeMeta(sourceAttributes); + pipeTaskInfo.createPipe( + new CreatePipePlanV2(pipeMeta.getStaticMeta(), pipeMeta.getRuntimeMeta())); + + final ParserTestContext context = createParserTestContext(1, pipeTaskInfo); + context.parser.parseHeartbeat( + 1, createPipeHeartbeatWithCompletedRegions(pipeMeta, false, Collections.emptyList())); + + Assert.assertNull(pipeTaskInfo.getPipeMetaByPipeName("test_pipe")); + verify(context.procedureManager, times(1)).pipeHandleMetaChange(true, true); + } + + @Test + public void testParseHeartbeatKeepsHistoryOnlyPipeWithoutDataRegionReport() throws Exception { + CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false); + + final Map sourceAttributes = new HashMap<>(); + sourceAttributes.put("source.realtime.enable", Boolean.FALSE.toString()); + final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo(); + final PipeMeta pipeMeta = createPipeMeta(sourceAttributes); + pipeTaskInfo.createPipe( + new CreatePipePlanV2(pipeMeta.getStaticMeta(), pipeMeta.getRuntimeMeta())); + + final ParserTestContext context = createParserTestContext(1, pipeTaskInfo); + context.parser.parseHeartbeat(1, createPipeHeartbeat(pipeMeta, false)); + + Assert.assertNotNull(pipeTaskInfo.getPipeMetaByPipeName("test_pipe")); + verify(context.procedureManager, never()).pipeHandleMetaChange(anyBoolean(), anyBoolean()); + } + + @Test + public void testParseHeartbeatKeepsRealtimePipeWithoutRequiredDataRegion() throws Exception { + CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false); + + final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo(); + final PipeMeta pipeMeta = createPipeMeta(Collections.emptyMap()); + pipeTaskInfo.createPipe( + new CreatePipePlanV2(pipeMeta.getStaticMeta(), pipeMeta.getRuntimeMeta())); + + final ParserTestContext context = createParserTestContext(1, pipeTaskInfo); + context.parser.parseHeartbeat( + 1, createPipeHeartbeatWithCompletedRegions(pipeMeta, false, Collections.emptyList())); + + 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); + final PipeMeta pipeMeta = createHistoryOnlyPipeMeta(1); pipeTaskInfo.createPipe( new CreatePipePlanV2(pipeMeta.getStaticMeta(), pipeMeta.getRuntimeMeta())); @@ -418,7 +473,7 @@ public void testParseHeartbeatCompletesOnlyAfterAllRequiredDataRegionsReported() CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false); final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo(); - final PipeMeta pipeMeta = createPipeMeta(1, 2); + final PipeMeta pipeMeta = createHistoryOnlyPipeMeta(1, 2); pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().get(2).setLeaderNodeId(2); pipeTaskInfo.createPipe( new CreatePipePlanV2(pipeMeta.getStaticMeta(), pipeMeta.getRuntimeMeta())); @@ -442,7 +497,7 @@ public void testParseHeartbeatKeepsCompletedDataRegionAfterLeaderChange() throws CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false); final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo(); - final PipeMeta pipeMeta = createPipeMeta(1, 2); + final PipeMeta pipeMeta = createHistoryOnlyPipeMeta(1, 2); pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().get(2).setLeaderNodeId(2); pipeTaskInfo.createPipe( new CreatePipePlanV2(pipeMeta.getStaticMeta(), pipeMeta.getRuntimeMeta())); @@ -577,6 +632,17 @@ private PipeMeta createPipeMeta() { } private PipeMeta createPipeMeta(final int... regionIds) { + return createPipeMeta(Collections.emptyMap(), regionIds); + } + + private PipeMeta createHistoryOnlyPipeMeta(final int... regionIds) { + final Map sourceAttributes = new HashMap<>(); + sourceAttributes.put("source.realtime.enable", Boolean.FALSE.toString()); + return createPipeMeta(sourceAttributes, regionIds); + } + + private PipeMeta createPipeMeta( + final Map sourceAttributes, final int... regionIds) { final PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta(); for (final int regionId : regionIds) { pipeRuntimeMeta @@ -584,7 +650,7 @@ private PipeMeta createPipeMeta(final int... regionIds) { .put(regionId, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)); } return new PipeMeta( - new PipeStaticMeta("test_pipe", 1L, new HashMap<>(), new HashMap<>(), new HashMap<>()), + new PipeStaticMeta("test_pipe", 1L, sourceAttributes, new HashMap<>(), new HashMap<>()), pipeRuntimeMeta); }