code: public void schedule(final Project project, final String submitUser)
        throws ProjectManagerException, IOException, SchedulerException {
  
      for (final Flow flow : project.getFlows()) {
        //todo chengren311: we should validate embedded flow shouldn't have flow trigger defined.
        if (flow.isEmbeddedFlow()) {
          // skip scheduling embedded flow since embedded flow are not allowed to have flow trigger
          continue;
        }
        final String flowFileName = flow.getId() + ".flow";
        final int latestFlowVersion = this.projectLoader
            .getLatestFlowVersion(flow.getProjectId(), flow
                .getVersion(), flowFileName);
        if (latestFlowVersion > 0) {
          final File tempDir = Files.createTempDir();
          final File flowFile;
          try {
            flowFile = this.projectLoader
                .getUploadedFlowFile(project.getId(), project.getVersion(),
                    flowFileName, latestFlowVersion, tempDir);
  
            final FlowTrigger flowTrigger = FlowLoaderUtils.getFlowTriggerFromYamlFile(flowFile);
  
            if (flowTrigger != null) {
              final Map<String, Object> contextMap = ImmutableMap
                  .of(FlowTriggerQuartzJob.SUBMIT_USER, submitUser,
                      FlowTriggerQuartzJob.FLOW_TRIGGER, flowTrigger,
                      FlowTriggerQuartzJob.FLOW_ID, flow.getId(),
                      FlowTriggerQuartzJob.FLOW_VERSION, latestFlowVersion,
                      FlowTriggerQuartzJob.PROJECT_ID, project.getId());
              final boolean scheduleSuccess = this.scheduler
                  .scheduleJobIfAbsent(flowTrigger.getSchedule().getCronExpression(),
                      new QuartzJobDescription
                      (FlowTriggerQuartzJob.class, FlowTriggerQuartzJob.JOB_NAME,
                          generateGroupName(flow), contextMap));
              if (scheduleSuccess) {
                logger.info("Successfully registered flow {}.{} to scheduler", project.getName(),
                    flow.getId());
              } else {
                logger.info("Fail to register a duplicate flow {}.{} to scheduler", project.getName(),
                    flow.getId());
              }
            }
          } catch (final SchedulerException | IOException ex) {
            logger.error("Error in registering flow {}.{}", project.getName(), flow.getId(), ex);
            throw ex;
          } finally {
            FlowLoaderUtils.cleanUpDir(tempDir);
          }
        }
      }
    }
docs: Schedule flows containing flow triggers for this project.