StreamProcessor { public void stop() { synchronized (lock) { LOGGER.info("Stopping the stream processor: {}.", processorId); if (state != State.STOPPING && state != State.STOPPED) { state = State.STOPPING; try { LOGGER.info("Shutting down the container: {} of stream processor: {}.", container, processorId); boolean hasContainerShutdown = stopSamzaContainer(); if (!hasContainerShutdown) { LOGGER.info("Interrupting the container: {} thread to die.", container); containerExecutorService.shutdownNow(); } } catch (Throwable throwable) { LOGGER.error(String.format("Exception occurred on container: %s shutdown of stream processor: %s.", container, processorId), throwable); } LOGGER.info("Shutting down JobCoordinator of stream processor: {}.", processorId); jobCoordinator.stop(); } else { LOGGER.info("StreamProcessor state is: {}. Ignoring the stop.", state); } } } @Deprecated  StreamProcessor(String processorId, Config config, Map<String, MetricsReporter> customMetricsReporters, TaskFactory taskFactory,
      ProcessorLifecycleListener processorListener); @Deprecated  StreamProcessor(String processorId, Config config, Map<String, MetricsReporter> customMetricsReporters, TaskFactory taskFactory,
      ProcessorLifecycleListener processorListener, JobCoordinator jobCoordinator); @Deprecated  StreamProcessor(String processorId, Config config, Map<String, MetricsReporter> customMetricsReporters, TaskFactory taskFactory,
      Optional<ApplicationContainerContextFactory<ApplicationContainerContext>> applicationDefinedContainerContextFactoryOptional,
      Optional<ApplicationTaskContextFactory<ApplicationTaskContext>> applicationDefinedTaskContextFactoryOptional,
      Optional<ExternalContext> externalContextOptional, StreamProcessorLifecycleListenerFactory listenerFactory,
      JobCoordinator jobCoordinator);  StreamProcessor(String processorId, Config config, Map<String, MetricsReporter> customMetricsReporters, TaskFactory taskFactory,
      Optional<ApplicationContainerContextFactory<ApplicationContainerContext>> applicationDefinedContainerContextFactoryOptional,
      Optional<ApplicationTaskContextFactory<ApplicationTaskContext>> applicationDefinedTaskContextFactoryOptional,
      Optional<ExternalContext> externalContextOptional, StreamProcessorLifecycleListenerFactory listenerFactory,
      JobCoordinator jobCoordinator, MetadataStore metadataStore); State getState(); void start(); void stop();  }
@Test public void testJobModelExpiredDuringAnExistingRebalanceWithContainerInterruptFailed() { JobCoordinator mockJobCoordinator = Mockito.mock(JobCoordinator.class); ProcessorLifecycleListener lifecycleListener = Mockito.mock(ProcessorLifecycleListener.class); ExecutorService mockExecutorService = Mockito.mock(ExecutorService.class); MapConfig config = new MapConfig(ImmutableMap.of("task.shutdown.ms", "0")); StreamProcessor streamProcessor = new StreamProcessor("TestProcessorId", config, new HashMap<>(), null, Optional.empty(), Optional.empty(), Optional.empty(), sp -> lifecycleListener, mockJobCoordinator, Mockito.mock(MetadataStore.class)); runJobModelExpireDuringRebalance(streamProcessor, mockExecutorService, true); assertEquals(State.STOPPING, streamProcessor.state); assertEquals(mockExecutorService, streamProcessor.containerExecutorService); Mockito.verify(mockExecutorService, Mockito.times(1)).shutdownNow(); Mockito.verify(mockJobCoordinator, Mockito.times(1)).stop(); }
@Test public void testStopShouldBeIdempotent() { JobCoordinator mockJobCoordinator = Mockito.mock(JobCoordinator.class); ProcessorLifecycleListener lifecycleListener = Mockito.mock(ProcessorLifecycleListener.class); SamzaContainer mockSamzaContainer = Mockito.mock(SamzaContainer.class); MapConfig config = new MapConfig(ImmutableMap.of("task.shutdown.ms", "0")); StreamProcessor streamProcessor = PowerMockito.spy(new StreamProcessor("TestProcessorId", config, new HashMap<>(), null, Optional.empty(), Optional.empty(), Optional.empty(), sp -> lifecycleListener, mockJobCoordinator, Mockito.mock(MetadataStore.class))); Mockito.doNothing().when(mockJobCoordinator).stop(); Mockito.doNothing().when(mockSamzaContainer).shutdown(); Mockito.when(mockSamzaContainer.hasStopped()).thenReturn(false); Mockito.when(mockSamzaContainer.getStatus()) .thenReturn(SamzaContainerStatus.STARTED) .thenReturn(SamzaContainerStatus.STOPPED); streamProcessor.state = State.RUNNING; streamProcessor.stop(); assertEquals(State.STOPPING, streamProcessor.state); }