Merge pull request #533 from zalando/bugfix/aruha-485

aruha-485: added shutdown hook to free subscription resources

Dockerfile

@@ -7,4 +7,4 @@ ADD build/libs/nakadi.jar nakadi.jar
 
 EXPOSE 8080
 
-ENTRYPOINT java -Djava.security.egd=file:/dev/./urandom -jar nakadi.jar
+ENTRYPOINT exec java -Djava.security.egd=file:/dev/./urandom -jar nakadi.jar

src/main/java/org/zalando/nakadi/service/subscription/StreamingContext.java

@@ -92,9 +92,16 @@ public long getKafkaPollTimeout() {
 
     @Override
     public void stream() throws InterruptedException {
+        // bugfix ARUHA-485
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> onNodeShutdown()));
         streamInternal(new StartingState());
     }
 
+    void onNodeShutdown() {
+        log.info("Shutdown hook called. Trying to terminate subscription gracefully");
+        switchState(new CleanupState(null));
+    }
+
     void streamInternal(final State firstState) throws InterruptedException {
         // Add first task - switch to starting state.
         switchState(firstState);

src/test/java/org/zalando/nakadi/service/subscription/StreamingContextTest.java

@@ -2,7 +2,10 @@
 
 import org.junit.Assert;
 import org.junit.Test;
+import org.mockito.Mockito;
 import org.zalando.nakadi.service.subscription.model.Session;
+import org.zalando.nakadi.service.subscription.state.CleanupState;
+import org.zalando.nakadi.service.subscription.state.DummyState;
 import org.zalando.nakadi.service.subscription.state.State;
 
 import java.io.IOException;
@@ -126,4 +129,28 @@ public void onExit() {
         // Check that onExit called even if onEnter throws exception.
         Assert.assertArrayEquals(new boolean[]{true, true}, onExitCalls);
     }
+
+    @Test
+    public void testOnNodeShutdown() throws Exception {
+        final StreamingContext ctxSpy = Mockito.spy(createTestContext(null));
+
+        final Thread t = new Thread(() -> {
+            try {
+                ctxSpy.streamInternal(new State() {
+                    @Override
+                    public void onEnter() {
+                    }
+                });
+            } catch (final InterruptedException ignore) {
+            }
+        });
+        t.start();
+        t.join(1000);
+
+        ctxSpy.onNodeShutdown();
+
+        Mockito.verify(ctxSpy).switchState(Mockito.isA(CleanupState.class));
+        Mockito.verify(ctxSpy).unregisterSession();
+        Mockito.verify(ctxSpy).switchState(Mockito.isA(DummyState.class));
+    }
 }