FlowRunner extends EventHandler<Event> implements Runnable { public FlowRunnerEventListener getFlowRunnerEventListener() { return this.flowListener; } 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 flowEventMetadata() throws Exception { final EventCollectorListener eventCollector = new EventCollectorListener(); eventCollector.setEventFilterOut(EventType.JOB_FINISHED, EventType.JOB_STARTED, EventType.JOB_STATUS_CHANGED); this.runner = this.testUtil.createFromFlowFile(eventCollector, "exec1"); FlowRunner.FlowRunnerEventListener flowRunnerEventListener = this.runner.getFlowRunnerEventListener(); Map<String, String> flowMetadata = flowRunnerEventListener.getFlowMetadata(this.runner); Assert.assertEquals("Event metadata not created as expected.", "localhost", flowMetadata.get("azkabanWebserver")); Assert.assertEquals("Event metadata not created as expected.", "unknown", flowMetadata.get("azkabanHost")); Assert.assertNull("Event metadata not created as expected.", flowMetadata.get("submitUser")); Assert.assertEquals("Event metadata not created as expected.", "test", flowMetadata.get("projectName")); Assert.assertEquals("Event metadata not created as expected.", "derived-member-data", flowMetadata.get("flowName")); Assert.assertEquals("Event metadata not created as expected.", "testUser", flowMetadata.get("projectFileUploadUser")); Assert.assertEquals("Event metadata not created as expected.", "111.111.111.111", flowMetadata.get("projectFileUploaderIpAddr")); Assert.assertEquals("Event metadata not created as expected.", "test.zip", flowMetadata.get("projectFileName")); Assert.assertEquals("Event metadata not created as expected.", "1", flowMetadata.get("projectFileUploadTime")); Assert.assertEquals("Event metadata not created as expected.", "null", flowMetadata.get("slaOptions")); }