You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2019/06/06 08:52:42 UTC
[flink] 01/02: [hotfix][network] Move partition release under
registeredPartitions modification lock in
ResultPartitionManager.onConsumedPartition
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a6d44a78f94bf70ff642ca2e3ba219623a7945a7
Author: Andrey Zagrebin <az...@gmail.com>
AuthorDate: Wed Jun 5 14:34:06 2019 +0200
[hotfix][network] Move partition release under registeredPartitions modification lock in ResultPartitionManager.onConsumedPartition
---
.../io/network/partition/ResultPartitionManager.java | 18 ++++++++----------
1 file changed, 8 insertions(+), 10 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionManager.java
index 8d96e2b..4050252 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionManager.java
@@ -106,19 +106,17 @@ public class ResultPartitionManager implements ResultPartitionProvider {
// ------------------------------------------------------------------------
void onConsumedPartition(ResultPartition partition) {
- final ResultPartition previous;
-
LOG.debug("Received consume notification from {}.", partition);
synchronized (registeredPartitions) {
- previous = registeredPartitions.remove(partition.getPartitionId());
- }
-
- // Release the partition if it was successfully removed
- if (partition == previous) {
- partition.release();
-
- LOG.debug("Released {}.", partition);
+ final ResultPartition previous = registeredPartitions.remove(partition.getPartitionId());
+ // Release the partition if it was successfully removed
+ if (partition == previous) {
+ partition.release();
+ ResultPartitionID partitionId = partition.getPartitionId();
+ LOG.debug("Released partition {} produced by {}.",
+ partitionId.getPartitionId(), partitionId.getProducerId());
+ }
}
}