Runner { public void run(Config config) throws Exception { this.baseConfig = config; config = ConfigUtils.mergeLoadedConfiguration(config); validateConfigurations(config); Set<Step> steps = StepUtils.extractSteps(config, true, true); ExecutionMode mode = getExecutionMode(steps); Contexts.initialize(config, mode); initializeSecurity(config, steps); initializeEventHandlers(config); initializeAccumulators(steps); initializeUDFs(config); initializeThreadPool(config); notifyPipelineStarted(); try { if (mode == ExecutionMode.STREAMING) { runStreaming(steps); } else { runBatch(steps); } } catch (Exception e) { notifyPipelineException(e); throw e; } finally { shutdownThreadPool(); shutdownSecurity(); } notifyPipelineFinished(); }  void run(Config config); static final String STEPS_SECTION_CONFIG; static final String TYPE_PROPERTY; static final String DATA_TYPE; static final String LOOP_TYPE; static final String DECISION_TYPE; static final String TASK_TYPE; static final String UDFS_SECTION_CONFIG; static final String UDFS_NAME; static final String UDFS_CLASS; static final String EVENT_HANDLERS_CONFIG; static final String CONFIG_LOADER_PROPERTY; static final String PIPELINE_THREADS_PROPERTY; }
@Test public void testExpectedCoreEvents() throws Exception { EventManager.reset(); String executionKey = UUID.randomUUID().toString(); Config executionKeyConfig = ConfigFactory.parseString("execution_key = " + executionKey); Config config = ConfigUtils.configFromResource("/event/expected_core_events.conf"); config = config.withFallback(executionKeyConfig).resolve(); new Runner().run(config); List<Event> events = TestingEventHandler.getHandledEvents(executionKey); List<String> eventTypes = Lists.newArrayList(); for (Event event : events) { eventTypes.add(event.getEventType()); } List<String> expectedEventTypes = Lists.newArrayList( CoreEventTypes.PIPELINE_STARTED, CoreEventTypes.PIPELINE_FINISHED, CoreEventTypes.DATA_STEP_DATA_GENERATED, CoreEventTypes.DATA_STEP_WRITTEN_TO_OUTPUT, CoreEventTypes.EXECUTION_MODE_DETERMINED, CoreEventTypes.STEPS_EXTRACTED ); for (String expectedEventType : expectedEventTypes) { assertTrue("Does not contain event type: " + expectedEventType, eventTypes.contains(expectedEventType)); } }
@Test public void testExceptionEvent() { EventManager.reset(); String executionKey = UUID.randomUUID().toString(); Config executionKeyConfig = ConfigFactory.parseString("execution_key = " + executionKey); Config config = ConfigUtils.configFromResource("/event/exception_event.conf"); config = config.withFallback(executionKeyConfig).resolve(); try { new Runner().run(config); } catch (Exception e) { } List<Event> events = TestingEventHandler.getHandledEvents(executionKey); List<String> eventTypes = Lists.newArrayList(); for (Event event : events) { eventTypes.add(event.getEventType()); } assertTrue(eventTypes.contains(CoreEventTypes.PIPELINE_EXCEPTION_OCCURRED)); }