gaoyunhaii commented on a change in pull request #13249:
URL: https://github.com/apache/flink/pull/13249#discussion_r482747530
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -1160,14 +1160,25 @@ public final CloseableRegistry getCancelables() {
return recordWriters;
}
+ @SuppressWarnings("unchecked")
private static <OUT>
RecordWriter<SerializationDelegate<StreamRecord<OUT>>> createRecordWriter(
StreamEdge edge,
int outputIndex,
Environment environment,
String taskName,
long bufferTimeout) {
- @SuppressWarnings("unchecked")
- StreamPartitioner<OUT> outputPartitioner =
(StreamPartitioner<OUT>) edge.getPartitioner();
+
+ StreamPartitioner<OUT> outputPartitioner = null;
+
+ // Clones the partition to avoid multiple stream edges sharing
the same stream partitioner,
+ // like the case of
https://issues.apache.org/jira/browse/FLINK-14087.
+ try {
+ outputPartitioner = InstantiationUtil.clone(
+ (StreamPartitioner<OUT>) edge.getPartitioner(),
+ Thread.currentThread().getContextClassLoader());
Review comment:
Very thanks Dawid! I also think that it would be much better that we get
the user classloader in a more deterministic way. I have updated the PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
