LocalApplicationRunner implements ApplicationRunner { @Override public void kill() { processors.forEach(sp -> { sp.getLeft().stop(); if (sp.getRight() != null) { sp.getRight().close(); } }); cleanup(); } LocalApplicationRunner(SamzaApplication app, Config config);  LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory); @VisibleForTesting  LocalApplicationRunner(SamzaApplication app, Config config, CoordinationUtils coordinationUtils); private  LocalApplicationRunner(LocalApplicationRunnerContext context); Optional<String> getRunId(); @Override void run(ExternalContext externalContext); @Override void kill(); @Override ApplicationStatus status(); @Override void waitForFinish(); @Override boolean waitForFinish(Duration timeout);  }
@Test public void testKill() throws Exception { Map<String, String> cfgs = new HashMap<>(); cfgs.put(ApplicationConfig.APP_PROCESSOR_ID_GENERATOR_CLASS, UUIDGenerator.class.getName()); config = new MapConfig(cfgs); ProcessorLifecycleListenerFactory mockFactory = (pContext, cfg) -> mock(ProcessorLifecycleListener.class); mockApp = (StreamApplication) appDesc -> appDesc.withProcessorLifecycleListenerFactory(mockFactory); prepareTest(); doReturn(Collections.singletonList(new JobConfig(new MapConfig(config)))).when(localPlanner).prepareJobs(); StreamProcessor sp = mock(StreamProcessor.class); CoordinatorStreamStore coordinatorStreamStore = mock(CoordinatorStreamStore.class); ArgumentCaptor<StreamProcessor.StreamProcessorLifecycleListenerFactory> captor = ArgumentCaptor.forClass(StreamProcessor.StreamProcessorLifecycleListenerFactory.class); doAnswer(i -> { ProcessorLifecycleListener listener = captor.getValue().createInstance(sp); listener.afterStart(); return null; }).when(sp).start(); doAnswer(i -> { ProcessorLifecycleListener listener = captor.getValue().createInstance(sp); listener.afterStop(); return null; }).when(sp).stop(); ExternalContext externalContext = mock(ExternalContext.class); doReturn(sp).when(runner) .createStreamProcessor(anyObject(), anyObject(), captor.capture(), eq(Optional.of(externalContext)), any(CoordinatorStreamStore.class)); doReturn(coordinatorStreamStore).when(runner).createCoordinatorStreamStore(any(Config.class)); runner.run(externalContext); runner.kill(); verify(coordinatorStreamStore).init(); verify(coordinatorStreamStore).close(); assertEquals(runner.status(), ApplicationStatus.SuccessfulFinish); }