cluster-configurable timeout

flink-streaming-java/src/main/java/org/apache/flink/streaming/configuration/TimerServiceOptions.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.configuration;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+/**
+ * Timer service configuration options.
+ */
+@PublicEvolving
+public class TimerServiceOptions {
+
+	/**
+	 * This configures how long we wait for the {@link org.apache.flink.streaming.runtime.tasks.ProcessingTimeService}
+	 * to finish all pending timer threads when the stream task performs a failover shutdown. See FLINK-5465.
+	 */
+	public static final ConfigOption<Long> TIMER_SERVICE_TERMINATION_AWAIT_MS = ConfigOptions
+		.key("timerservice.exceptional.shutdown.timeout")
+		.defaultValue(7500L);
+}

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/ProcessingTimeService.java

@@ -97,7 +97,7 @@ public abstract class ProcessingTimeService {
 
 	/**
 	 * Shuts down and clean up the timer service provider hard and immediately. This does wait
-	 * for all timer to complete or until the time limit is exceeded. Any call to
+	 * for all timers to complete or until the time limit is exceeded. Any call to
 	 * {@link #registerTimer(long, ProcessingTimeCallback)} will result in a hard exception after calling this method.
 	 * @param time time to wait for termination.
 	 * @param timeUnit time unit of parameter time.

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java

@@ -51,6 +51,7 @@
 import org.apache.flink.streaming.api.operators.OperatorSnapshotResult;
 import org.apache.flink.streaming.api.operators.Output;
 import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.configuration.TimerServiceOptions;
 import org.apache.flink.streaming.runtime.io.RecordWriterOutput;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.streaming.runtime.streamstatus.StreamStatusMaintainer;
@@ -123,8 +124,6 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
 	/** The logger used by the StreamTask and its subclasses. */
 	private static final Logger LOG = LoggerFactory.getLogger(StreamTask.class);
 
-	private static final long TIMER_SERVICE_TERMINATION_AWAIT_MS = 7500L;
-
 	// ------------------------------------------------------------------------
 
 	/**
@@ -325,13 +324,16 @@ public final void invoke() throws Exception {
 			if (timerService != null && !timerService.isTerminated()) {
 				try {
 
+					final long timeoutMs = getTaskConfiguration().
+						getLong(TimerServiceOptions.TIMER_SERVICE_TERMINATION_AWAIT_MS);
+
 					// wait for a reasonable time for all pending timer threads to finish
 					boolean timerShutdownComplete =
-						timerService.shutdownAndAwaitPending(TIMER_SERVICE_TERMINATION_AWAIT_MS, TimeUnit.MILLISECONDS);
+						timerService.shutdownAndAwaitPending(timeoutMs, TimeUnit.MILLISECONDS);
 
 					if (!timerShutdownComplete) {
-						LOG.warn("Timer service shutdown exceeded time limit while waiting for pending timers. " +
-							"Will continue with shutdown procedure.");
+						LOG.warn("Timer service shutdown exceeded time limit of {} ms while waiting for pending " +
+							"timers. Will continue with shutdown procedure.", timeoutMs);
 					}
 				}
 				catch (Throwable t) {