private void doTestUseNodeUrls(String connectorName) throws Exception {
    senderEnv
        .getConfig()
        .getCommonConfig()
        .setAutoCreateSchemaEnabled(true)
        .setPipeAirGapReceiverEnabled(true)
        .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
        .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
        .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
        .setSchemaReplicationFactor(1)
        .setDataReplicationFactor(1)
        .setEnableSeqSpaceCompaction(false)
        .setEnableUnseqSpaceCompaction(false)
        .setEnableCrossSpaceCompaction(false);
    receiverEnv
        .getConfig()
        .getCommonConfig()
        .setAutoCreateSchemaEnabled(true)
        .setPipeAirGapReceiverEnabled(true)
        .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
        .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
        .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
        .setSchemaReplicationFactor(3)
        .setDataReplicationFactor(2);

    // 10 min, assert that the operations will not time out
    senderEnv.getConfig().getCommonConfig().setDnConnectionTimeoutMs(600000);
    receiverEnv.getConfig().getCommonConfig().setDnConnectionTimeoutMs(600000);

    senderEnv.initClusterEnvironment(1, 1);
    receiverEnv.initClusterEnvironment(1, 3);

    final StringBuilder nodeUrlsBuilder = new StringBuilder();
    final Consumer<String> handleFailure =
        o -> {
          TestUtils.executeNonQueryWithRetry(senderEnv, "flush");
          TestUtils.executeNonQueryWithRetry(receiverEnv, "flush");
        };

    boolean insertResult = true;
    for (final DataNodeWrapper wrapper : receiverEnv.getDataNodeWrapperList()) {
      if (connectorName.equals(BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName())) {
        // Use default port for convenience
        nodeUrlsBuilder
            .append(wrapper.getIp())
            .append(":")
            .append(wrapper.getPipeAirGapReceiverPort())
            .append(",");
      } else {
        nodeUrlsBuilder.append(wrapper.getIpAndPortString()).append(",");
      }
    }

    try (final SyncConfigNodeIServiceClient client =
        (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {

      TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
      insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
      if (!insertResult) {
        return;
      }
      final Map<String, String> extractorAttributes = new HashMap<>();
      final Map<String, String> processorAttributes = new HashMap<>();
      final Map<String, String> connectorAttributes = new HashMap<>();

      connectorAttributes.put("connector", connectorName);
      connectorAttributes.put("connector.batch.enable", "false");
      connectorAttributes.put("connector.batch.max-delay-seconds", "1");
      connectorAttributes.put("connector.batch.size-bytes", "2048");
      connectorAttributes.put("connector.node-urls", nodeUrlsBuilder.toString());

      extractorAttributes.put("capture.table", "true");
      extractorAttributes.put("database-name", "test.*");
      extractorAttributes.put("table-name", "test.*");
      extractorAttributes.put("inclusion", "data.insert");
      extractorAttributes.put("mode.snapshot", "false");
      extractorAttributes.put("mode.strict", "true");
      extractorAttributes.put("user", "root");

      // Test forced-log mode, in open releases this might be "file"
      extractorAttributes.put("realtime.mode", "forced-log");

      TSStatus status =
          client.createPipe(
              new TCreatePipeReq("p1", connectorAttributes)
                  .setExtractorAttributes(extractorAttributes)
                  .setProcessorAttributes(processorAttributes));

      Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
      Assert.assertEquals(
          TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());

      insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
      if (!insertResult) {
        return;
      }
      TableModelUtils.assertData("test", "test", 0, 200, receiverEnv, handleFailure);

      insertResult = TableModelUtils.insertData("test", "test", 200, 300, senderEnv);
      if (!insertResult) {
        return;
      }
      TableModelUtils.assertData("test", "test", 0, 300, receiverEnv, handleFailure);

      extractorAttributes.replace("realtime.mode", "file");

      status =
          client.createPipe(
              new TCreatePipeReq("p2", connectorAttributes)
                  .setExtractorAttributes(extractorAttributes)
                  .setProcessorAttributes(processorAttributes));

      Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
      Assert.assertEquals(
          TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p2").getCode());

      insertResult = TableModelUtils.insertData("test", "test", 300, 400, senderEnv);
      if (!insertResult) {
        return;
      }
      TableModelUtils.assertData("test", "test", 0, 400, receiverEnv, handleFailure);
    }
  }
Test pipe's basic functionalities under multiple cluster and consensus protocol settings.
integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeProtocolIT.java