dawidwys commented on a change in pull request #13249:
URL: https://github.com/apache/flink/pull/13249#discussion_r481928736
##########
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:
I am really sorry for yet another comment, but I've spotted it only now.
How about we use the `environment.getUserClassLoader()` instead of
`Thread.currentThread().getContextClassLoader()`? I think it is safer that way.
----------------------------------------------------------------
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
