Github user sihuazhou commented on a diff in the pull request:
https://github.com/apache/flink/pull/5962#discussion_r186775375
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
---
@@ -706,6 +689,38 @@ public void notifyCheckpointComplete(long
checkpointId) throws Exception {
}
}
+ private void tryShutdownTimerService() {
+ if (timerService != null && !timerService.isTerminated()) {
+ try {
+
+ final long totalTimeoutMs =
getEnvironment().getTaskManagerInfo().getConfiguration().
+
getLong(TaskManagerOptions.TASK_CANCELLATION_TIMEOUT_TIMERS);
+
+ final Deadline deadline =
Deadline.fromNow(Duration.ofMillis(totalTimeoutMs));
+
+ boolean timerServiceShutdownComplete = false;
+
+ while (!timerServiceShutdownComplete &&
deadline.hasTimeLeft()) {
+ try {
--- End diff --
Shall we at least try to shutdown the time service one time? If so, I think
maybe we should use `do {...} while()` here.
---
