[FLINK-9275] [streaming] Add taskName to the output flusher thread's name This closes #5943
Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a10f4791 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a10f4791 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a10f4791 Branch: refs/heads/release-1.5 Commit: a10f4791b1831e782fc26fe2d5c70336cfcd808c Parents: 24c3348 Author: Nico Kruber <[email protected]> Authored: Mon Apr 30 15:02:43 2018 +0200 Committer: Stephan Ewen <[email protected]> Committed: Mon Apr 30 23:25:27 2018 +0200 ---------------------------------------------------------------------- .../org/apache/flink/streaming/runtime/io/StreamRecordWriter.java | 3 ++- .../java/org/apache/flink/streaming/runtime/tasks/StreamTask.java | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/a10f4791/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamRecordWriter.java ---------------------------------------------------------------------- diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamRecordWriter.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamRecordWriter.java index dad680c..9fedf70 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamRecordWriter.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamRecordWriter.java @@ -67,7 +67,8 @@ public class StreamRecordWriter<T extends IOReadableWritable> extends RecordWrit } else { String threadName = taskName == null ? - DEFAULT_OUTPUT_FLUSH_THREAD_NAME : "Output Timeout Flusher - " + taskName; + DEFAULT_OUTPUT_FLUSH_THREAD_NAME : + DEFAULT_OUTPUT_FLUSH_THREAD_NAME + " for " + taskName; outputFlusher = new OutputFlusher(threadName, timeout); outputFlusher.start(); http://git-wip-us.apache.org/repos/asf/flink/blob/a10f4791/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java ---------------------------------------------------------------------- diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java index 6812871..6790949 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java @@ -1204,7 +1204,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> } StreamRecordWriter<SerializationDelegate<StreamRecord<OUT>>> output = - new StreamRecordWriter<>(bufferWriter, outputPartitioner, bufferTimeout); + new StreamRecordWriter<>(bufferWriter, outputPartitioner, bufferTimeout, taskName); output.setMetricGroup(environment.getMetricGroup().getIOMetricGroup()); return output; }
