FlowRunner extends EventHandler<Event> implements Runnable { @VisibleForTesting static void propagateMetadataFromProps(final Map<String, String> metaData, final Props inputProps, final String nodeType, final String nodeName, final Logger logger) { if (null == metaData || null == inputProps || null == logger || Strings.isNullOrEmpty(nodeType) || Strings.isNullOrEmpty(nodeName)) { throw new IllegalArgumentException("Input params should not be null or empty."); } if (!inputProps.containsKey(AZKABAN_EVENT_REPORTING_PROPERTIES_TO_PROPAGATE)) { return; } final String propsToPropagate = inputProps .getString(AZKABAN_EVENT_REPORTING_PROPERTIES_TO_PROPAGATE); if (Strings.isNullOrEmpty(propsToPropagate)) { logger.info( String.format("No properties to propagate to metadata for %s: %s", nodeType, nodeName)); return; } else { logger.info(String .format("Propagating: %s to metadata for %s: %s", propsToPropagate, nodeType, nodeName)); } final List<String> propsToPropagateList = SPLIT_ON_COMMA.splitToList(propsToPropagate); for (final String propKey : propsToPropagateList) { if (!inputProps.containsKey(propKey)) { logger.warn(String.format("%s does not contains: %s property; " + "skipping propagation to metadata", nodeName, propKey)); continue; } metaData.put(propKey, inputProps.getString(propKey)); } } FlowRunner(final ExecutableFlow flow, final ExecutorLoader executorLoader,
      final ProjectLoader projectLoader, final JobTypeManager jobtypeManager,
      final Props azkabanProps, final AzkabanEventReporter azkabanEventReporter,
      final AlerterHolder alerterHolder, final CommonMetrics commonMetrics,
      final ExecMetrics execMetrics);  FlowRunner(final ExecutableFlow flow, final ExecutorLoader executorLoader,
      final ProjectLoader projectLoader, final JobTypeManager jobtypeManager,
      final ExecutorService executorService, final Props azkabanProps,
      final AzkabanEventReporter azkabanEventReporter, final AlerterHolder alerterHolder,
      final CommonMetrics commonMetrics, final ExecMetrics execMetrics); long getFlowKillTime(); long getFlowKillDuration(); long getFlowPauseTime(); void setFlowCreateTime(final long flowCreateTime); long getFlowPauseDuration(); long getFlowCreateTime(); FlowRunner setFlowWatcher(final FlowWatcher watcher); FlowRunner setNumJobThreads(final int jobs); FlowRunner setJobLogSettings(final String jobLogFileSize, final int jobLogNumFiles); FlowRunner setValidateProxyUser(final boolean validateUserProxy); File getExecutionDir(); @Override void run(); Status getImpliedStatus(final ExecutableNode node); void pause(final String user); void resume(final String user); void kill(final String user); void kill(); void retryFailures(final String user); boolean isKilled(); boolean isRamping(); ExecutableFlow getExecutableFlow(); File getFlowLogFile(); File getJobLogFile(final String jobId, final int attempt); File getJobAttachmentFile(final String jobId, final int attempt); File getJobMetaDataFile(final String jobId, final int attempt); boolean isRunnerThreadAlive(); int getExecutionId(); Set<JobRunner> getActiveJobRunners(); FlowRunnerEventListener getFlowRunnerEventListener();  }
@Test public void addMetadataFromProperties() throws Exception { Map<String, String> metadataMap = new HashMap<>(); Props inputProps = new Props(); inputProps.put(Constants.ConfigurationKeys.AZKABAN_EVENT_REPORTING_PROPERTIES_TO_PROPAGATE, "my.prop1,my.prop2"); inputProps.put("my.prop1", "value1"); inputProps.put("my.prop2", "value2"); FlowRunner.propagateMetadataFromProps(metadataMap, inputProps, "flow", "dummyFlow", Logger.getLogger(FlowRunnerTest.class)); Assert.assertEquals("Metadata not propagated correctly.", 2, metadataMap.size()); Assert.assertEquals("Metadata not propagated correctly.", "value1", metadataMap.get("my.prop1")); Assert.assertEquals("Metadata not propagated correctly.", "value2", metadataMap.get("my.prop2")); metadataMap = new HashMap<>(); FlowRunner.propagateMetadataFromProps(metadataMap, new Props(), "flow", "dummyFlow", Logger.getLogger(FlowRunnerTest.class)); Assert.assertEquals("Metadata propagation backward compatibility has issues.", 0, metadataMap.size()); try { FlowRunner.propagateMetadataFromProps(null, inputProps, "flow", "dummyFlow", Logger.getLogger(FlowRunnerTest.class)); Assert.fail("Metadata propagation did not fail with bad data."); } catch (Exception e) { } }