You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by zh...@apache.org on 2021/09/28 03:02:52 UTC
[flink] branch master updated: [FLINK-24375] Let name of flush
thread in RecordWriter contain subtask index.
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 5947647 [FLINK-24375] Let name of flush thread in RecordWriter contain subtask index.
5947647 is described below
commit 5947647a3390bf7804ca927976dda909fb6db295
Author: Lijie Wang <wa...@gmail.com>
AuthorDate: Sun Sep 26 22:11:24 2021 +0800
[FLINK-24375] Let name of flush thread in RecordWriter contain subtask index.
---
.../main/java/org/apache/flink/runtime/operators/BatchTask.java | 3 ++-
.../java/org/apache/flink/streaming/runtime/tasks/StreamTask.java | 8 ++++----
2 files changed, 6 insertions(+), 5 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/BatchTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/BatchTask.java
index ec78a44..8dcf770 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/BatchTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/BatchTask.java
@@ -1386,7 +1386,8 @@ public class BatchTask<S extends Function, OT> extends AbstractInvokable
final RecordWriter<SerializationDelegate<T>> recordWriter =
new RecordWriterBuilder()
.setChannelSelector(oe)
- .setTaskName(task.getEnvironment().getTaskInfo().getTaskName())
+ .setTaskName(
+ task.getEnvironment().getTaskInfo().getTaskNameWithSubtasks())
.build(task.getEnvironment().getWriter(outputOffset + i));
recordWriter.setMetricGroup(task.getEnvironment().getMetricGroup().getIOMetricGroup());
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 2872bee..eb5c5ca 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
@@ -1619,7 +1619,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
edge,
i,
environment,
- environment.getTaskInfo().getTaskName(),
+ environment.getTaskInfo().getTaskNameWithSubtasks(),
edge.getBufferTimeout()));
}
return recordWriters;
@@ -1630,7 +1630,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
StreamEdge edge,
int outputIndex,
Environment environment,
- String taskName,
+ String taskNameWithSubtask,
long bufferTimeout) {
StreamPartitioner<OUT> outputPartitioner = null;
@@ -1650,7 +1650,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
"Using partitioner {} for output {} of task {}",
outputPartitioner,
outputIndex,
- taskName);
+ taskNameWithSubtask);
ResultPartitionWriter bufferWriter = environment.getWriter(outputIndex);
@@ -1666,7 +1666,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
new RecordWriterBuilder<SerializationDelegate<StreamRecord<OUT>>>()
.setChannelSelector(outputPartitioner)
.setTimeout(bufferTimeout)
- .setTaskName(taskName)
+ .setTaskName(taskNameWithSubtask)
.build(bufferWriter);
output.setMetricGroup(environment.getMetricGroup().getIOMetricGroup());
return output;