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 2020/04/14 10:14:11 UTC
[flink] 02/04: [FLINK-14504][runtime] List incomplete data sets
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 2ac349259c8bed50728baf8003a4318d9823101c
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Thu Apr 2 13:24:54 2020 +0200
[FLINK-14504][runtime] List incomplete data sets
We don't have a mechanism to ensure that either all partitions of a data set are registered or the remaining partitions are released at some point.
In these cases the user has to step in, but without being able to list incomplete partitions there's nothing that can be done.
This commit relaxes the filtering, and introduces an additional optional count for the number of currently registered partitions.
This count is only determined when the listing is requested for performance and complexity reasons.
---
.../io/network/partition/DataSetMetaInfo.java | 24 +++++++++-
.../partition/ResourceManagerPartitionTracker.java | 4 +-
.../ResourceManagerPartitionTrackerImpl.java | 9 ++--
.../ResourceManagerPartitionTrackerImplTest.java | 51 +++++-----------------
4 files changed, 40 insertions(+), 48 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/DataSetMetaInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/DataSetMetaInfo.java
index 36a37fb..49571dc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/DataSetMetaInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/DataSetMetaInfo.java
@@ -17,18 +17,40 @@
package org.apache.flink.runtime.io.network.partition;
+import org.apache.flink.util.Preconditions;
+
+import java.util.Optional;
+
/**
* Container for meta-data of a data set.
*/
public final class DataSetMetaInfo {
+ private static final int UNKNOWN = -1;
+ private final int numRegisteredPartitions;
private final int numTotalPartitions;
- public DataSetMetaInfo(int numTotalPartitions) {
+ public DataSetMetaInfo(int numRegisteredPartitions, int numTotalPartitions) {
+ this.numRegisteredPartitions = numRegisteredPartitions;
this.numTotalPartitions = numTotalPartitions;
}
+ public Optional<Integer> getNumRegisteredPartitions() {
+ return numRegisteredPartitions == UNKNOWN
+ ? Optional.empty()
+ : Optional.of(numRegisteredPartitions);
+ }
+
public int getNumTotalPartitions() {
return numTotalPartitions;
}
+
+ static DataSetMetaInfo withoutNumRegisteredPartitions(int numTotalPartitions) {
+ return new DataSetMetaInfo(UNKNOWN, numTotalPartitions);
+ }
+
+ static DataSetMetaInfo withNumRegisteredPartitions(int numRegisteredPartitions, int numTotalPartitions) {
+ Preconditions.checkArgument(numRegisteredPartitions > 0);
+ return new DataSetMetaInfo(numRegisteredPartitions, numTotalPartitions);
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTracker.java
index 96da998..82f8afd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTracker.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTracker.java
@@ -59,9 +59,9 @@ public interface ResourceManagerPartitionTracker {
CompletableFuture<Void> releaseClusterPartitions(IntermediateDataSetID dataSetId);
/**
- * Returns all data sets for which all partitions are being tracked.
+ * Returns all data sets for which partitions are being tracked.
*
- * @return completely tracked datasets
+ * @return tracked datasets
*/
Map<IntermediateDataSetID, DataSetMetaInfo> listDataSets();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImpl.java
index 47940ed..2c77a6e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImpl.java
@@ -159,7 +159,7 @@ public class ResourceManagerPartitionTrackerImpl implements ResourceManagerParti
clusterPartitionReport.getEntries().forEach(entry ->
dataSetMetaInfo.compute(entry.getDataSetId(), (dataSetID, dataSetMetaInfo) -> {
if (dataSetMetaInfo == null) {
- return new DataSetMetaInfo(entry.getNumTotalPartitions());
+ return DataSetMetaInfo.withoutNumRegisteredPartitions(entry.getNumTotalPartitions());
} else {
// double check that the meta data is consistent
Preconditions.checkState(dataSetMetaInfo.getNumTotalPartitions() == entry.getNumTotalPartitions());
@@ -201,7 +201,7 @@ public class ResourceManagerPartitionTrackerImpl implements ResourceManagerParti
@Override
public Map<IntermediateDataSetID, DataSetMetaInfo> listDataSets() {
return dataSetMetaInfo.entrySet().stream()
- .filter(entry -> {
+ .collect(Collectors.toMap(Map.Entry::getKey, entry -> {
final Map<ResourceID, Set<ResultPartitionID>> taskExecutorToPartitions = dataSetToTaskExecutors.get(entry.getKey());
Preconditions.checkState(taskExecutorToPartitions != null, "Have metadata entry for dataset %s, but no partition is tracked.", entry.getKey());
@@ -210,9 +210,8 @@ public class ResourceManagerPartitionTrackerImpl implements ResourceManagerParti
numTrackedPartitions += hostedPartitions.size();
}
- return numTrackedPartitions == entry.getValue().getNumTotalPartitions();
- })
- .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ return DataSetMetaInfo.withNumRegisteredPartitions(numTrackedPartitions, entry.getValue().getNumTotalPartitions());
+ }));
}
/**
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImplTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImplTest.java
index 71e1c76..f07c67b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImplTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImplTest.java
@@ -102,56 +102,27 @@ public class ResourceManagerPartitionTrackerImplTest extends TestLogger {
assertThat(tracker.listDataSets().size(), is(0));
- report(tracker, TASK_EXECUTOR_ID_1, DATA_SET_ID, 1, PARTITION_ID_1);
-
- final Map<IntermediateDataSetID, DataSetMetaInfo> listing = tracker.listDataSets();
- assertThat(listing, hasKey(DATA_SET_ID));
- DataSetMetaInfo metaInfo = listing.get(DATA_SET_ID);
- assertThat(metaInfo.getNumTotalPartitions(), is(1));
-
- reportEmpty(tracker, TASK_EXECUTOR_ID_1);
- assertThat(tracker.listDataSets().size(), is(0));
-
- assertThat(tracker.areAllMapsEmpty(), is(true));
- }
-
- @Test
- public void testListDataSetsMultiplePartitionsOnSingleTaskExecutor() {
- final ResourceManagerPartitionTrackerImpl tracker = new ResourceManagerPartitionTrackerImpl(new TestClusterPartitionReleaser());
-
- // data set consists of 2 partitions but only 1 is being tracked -> incomplete and should not be listed (yet)
report(tracker, TASK_EXECUTOR_ID_1, DATA_SET_ID, 2, PARTITION_ID_1);
- assertThat(tracker.listDataSets().size(), is(0));
+ checkListedDataSets(tracker, 1, 2);
- // start tracking another partitions, but we lost partition 1 so the data set is still incomplete
- report(tracker, TASK_EXECUTOR_ID_1, DATA_SET_ID, 2, PARTITION_ID_2);
- assertThat(tracker.listDataSets().size(), is(0));
+ report(tracker, TASK_EXECUTOR_ID_2, DATA_SET_ID, 2, PARTITION_ID_2);
+ checkListedDataSets(tracker, 2, 2);
- // dataset is considered complete since all partitions are being tracked
- report(tracker, TASK_EXECUTOR_ID_1, DATA_SET_ID, 2, PARTITION_ID_1, PARTITION_ID_2);
- final Map<IntermediateDataSetID, DataSetMetaInfo> listing = tracker.listDataSets();
- assertThat(listing, hasKey(DATA_SET_ID));
+ reportEmpty(tracker, TASK_EXECUTOR_ID_1);
+ checkListedDataSets(tracker, 1, 2);
- // dataset is no longer considered complete since partition 2 was lost
- report(tracker, TASK_EXECUTOR_ID_1, DATA_SET_ID, 2, PARTITION_ID_1);
+ reportEmpty(tracker, TASK_EXECUTOR_ID_2);
assertThat(tracker.listDataSets().size(), is(0));
- assertThat(tracker.areAllMapsEmpty(), is(false));
+ assertThat(tracker.areAllMapsEmpty(), is(true));
}
- @Test
- public void testListDataSetsMultiplePartitionsAcrossTaskExecutors() {
- final ResourceManagerPartitionTrackerImpl tracker = new ResourceManagerPartitionTrackerImpl(new TestClusterPartitionReleaser());
-
- report(tracker, TASK_EXECUTOR_ID_1, DATA_SET_ID, 2, PARTITION_ID_1);
- report(tracker, TASK_EXECUTOR_ID_2, DATA_SET_ID, 2, PARTITION_ID_2);
+ private static void checkListedDataSets(ResourceManagerPartitionTracker tracker, int expectedRegistered, int expectedTotal) {
final Map<IntermediateDataSetID, DataSetMetaInfo> listing = tracker.listDataSets();
assertThat(listing, hasKey(DATA_SET_ID));
-
- reportEmpty(tracker, TASK_EXECUTOR_ID_1);
- assertThat(tracker.listDataSets().size(), is(0));
-
- assertThat(tracker.areAllMapsEmpty(), is(false));
+ DataSetMetaInfo metaInfo = listing.get(DATA_SET_ID);
+ assertThat(metaInfo.getNumRegisteredPartitions().orElse(-1), is(expectedRegistered));
+ assertThat(metaInfo.getNumTotalPartitions(), is(expectedTotal));
}
@Test