Changes per PR feedback

line · Feb 22, 2021 · cd84d5e · cd84d5e
1 parent f26b7ec
commit cd84d5e
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 6 deletions.
diff --git a/processor/src/main/java/com/linecorp/decaton/processor/runtime/ProcessorSubscription.java b/processor/src/main/java/com/linecorp/decaton/processor/runtime/ProcessorSubscription.java
@@ -235,8 +235,9 @@ public void run() {
                 metrics.commitOffsetTime.record(timer.duration());
             }
             updateState(SubscriptionStateListener.State.SHUTTING_DOWN);
-            waitForRemainingTasksCompletion(
-                    scope.props().get(ProcessorProperties.CONFIG_SHUTDOWN_TIMEOUT_MS).value());
+            final long timeoutMillis =
+                    scope.props().get(ProcessorProperties.CONFIG_SHUTDOWN_TIMEOUT_MS).value();
+            if (timeoutMillis > 0) { waitForRemainingTasksCompletion(timeoutMillis); }
         } catch (RuntimeException e) {
             log.error("Unknown exception thrown at subscription loop, thread will be terminated: {}", scope, e);
             updateState(SubscriptionStateListener.State.SHUTTING_DOWN);

diff --git a/processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/ProcessorUnit.java b/processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/ProcessorUnit.java
@@ -52,10 +52,7 @@ public ProcessorUnit(ThreadScope scope, ProcessPipeline<?> pipeline) {
                                    "partition", String.valueOf(tp.partition()),
                                    "subpartition", String.valueOf(scope.threadId()))
                 .new ResourceUtilizationMetrics();
-        shutdownFuture = executorShutdownFuture.thenAccept(v -> {
-            pipeline.close();
-            metrics.close();
-        });
+        shutdownFuture = executorShutdownFuture.thenAccept(v -> metrics.close());
     }
 
     public void putTask(TaskRequest request) {
@@ -95,6 +92,7 @@ public void initiateShutdown() {
         // finish
         executor.submit(() -> executorShutdownFuture.complete(null));
         executor.shutdown();
+        pipeline.close();
     }
 
     @Override