From 9b3a9e68bca8cfef8a87542143212015b5263cea Mon Sep 17 00:00:00 2001 From: Abhishek Shivanna Date: Tue, 17 Sep 2019 11:23:19 -0700 Subject: [PATCH 1/2] Fix test to exit without waiting for timeout --- .../runtime/TestClusterBasedProcessorLifecycleListener.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/samza-core/src/test/java/org/apache/samza/runtime/TestClusterBasedProcessorLifecycleListener.java b/samza-core/src/test/java/org/apache/samza/runtime/TestClusterBasedProcessorLifecycleListener.java index 353bb0f3ff..0d93aa9a6b 100644 --- a/samza-core/src/test/java/org/apache/samza/runtime/TestClusterBasedProcessorLifecycleListener.java +++ b/samza-core/src/test/java/org/apache/samza/runtime/TestClusterBasedProcessorLifecycleListener.java @@ -79,11 +79,14 @@ public void testLifecycleListenerAfterFailure() { public void testShutdownHookInvokesShutdownHookCallback() { doNothing().when(mockShutdownHookCallback).run(); - // setup shutdown hook + // call beforeStart to setup shutdownHook clusterBasedProcessorLifecycleListener.beforeStart(); // Simulating shutdown hook invocation by JVM clusterBasedProcessorLifecycleListener.getShutdownHookThread().run(); + + // call afterStop to countdown shutdownLatch + clusterBasedProcessorLifecycleListener.afterStop(); Mockito.verify(mockShutdownHookCallback).run(); } } From a7112619f1299501666e4a581e45452bcbe8507a Mon Sep 17 00:00:00 2001 From: Abhishek Shivanna Date: Tue, 17 Sep 2019 14:16:30 -0700 Subject: [PATCH 2/2] addressing comments --- ...estClusterBasedProcessorLifecycleListener.java | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/samza-core/src/test/java/org/apache/samza/runtime/TestClusterBasedProcessorLifecycleListener.java b/samza-core/src/test/java/org/apache/samza/runtime/TestClusterBasedProcessorLifecycleListener.java index 0d93aa9a6b..777929a795 100644 --- a/samza-core/src/test/java/org/apache/samza/runtime/TestClusterBasedProcessorLifecycleListener.java +++ b/samza-core/src/test/java/org/apache/samza/runtime/TestClusterBasedProcessorLifecycleListener.java @@ -40,7 +40,7 @@ public void setup() { mockShutdownHookCallback = mock(Runnable.class); processorLifecycleListener = mock(ProcessorLifecycleListener.class); clusterBasedProcessorLifecycleListener = - spy(new ClusterBasedProcessorLifecycleListener(new MapConfig(ImmutableMap.of(TaskConfig.TASK_SHUTDOWN_MS, "1")), + spy(new ClusterBasedProcessorLifecycleListener(new MapConfig(ImmutableMap.of(TaskConfig.TASK_SHUTDOWN_MS, "30000")), processorLifecycleListener, mockShutdownHookCallback)); doNothing().when(clusterBasedProcessorLifecycleListener).addJVMShutdownHook(any(Thread.class)); doNothing().when(clusterBasedProcessorLifecycleListener).removeJVMShutdownHook(any(Thread.class)); @@ -77,16 +77,19 @@ public void testLifecycleListenerAfterFailure() { @Test public void testShutdownHookInvokesShutdownHookCallback() { - doNothing().when(mockShutdownHookCallback).run(); + doAnswer(invocation -> { + // Simulate call to container.shutdown() + clusterBasedProcessorLifecycleListener.afterStop(); + return null; + }).when(mockShutdownHookCallback).run(); // call beforeStart to setup shutdownHook clusterBasedProcessorLifecycleListener.beforeStart(); - // Simulating shutdown hook invocation by JVM + // Simulating shutdown hook invocation by JVM. + // The shutdownHookThread should return immediately and shutdown + // cleanly if mockShutdownHookCallback.run() is invoked. clusterBasedProcessorLifecycleListener.getShutdownHookThread().run(); - - // call afterStop to countdown shutdownLatch - clusterBasedProcessorLifecycleListener.afterStop(); Mockito.verify(mockShutdownHookCallback).run(); } }