You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/03/18 14:23:08 UTC
[GitHub] [flink] zentol opened a new pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
zentol opened a new pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443
Based on #10362 (FLINK-14792).
With this PR the ResourceManager tracks cluster partitions hosted on task executors, based on the `ClusterPartitionReports` that are already submitted via TE heartbeats.
The entire partition handling logic is encapsulated in a `ResourceManagerPartitionTracker`; the RM does not really do anything and just informs the tracker about certain events (arrival of a new report, shutdown of a TM, requested release of a partition via the REST API).
The tracker maintains the following mappings
a) data set -> TE
b) TE -> data set
c) TE -> partitions
Mapping a) is required for the release of partitions via the REST API.
Mapping b) is required for figuring out which data sets may be affected by a TE shutdown.
Mapping c) is required for detecting when all partitions of a partition are being tracked and whether a TE has lost a single partition.
The tracker not only tracks partitions but also searches for data sets that have been corrupted by the loss of partitions (e.g., because of a TE shutdown). In this case the tracker will issue release calls for the remaining partitions to the hosting task executors.
Note that `ResourceManagerPartitionTracker#listDataSets` and `#releastPartitions` are unused at this time, but are required for the upcoming REST API.
This PR does _NOT_ handle cases where a partition of a cluster partition is never being tracked, for example because the TE died just after the job finished. In this case we will currently never release partitions of the corresponding data set.
This will be tackled in a follow-up, and requires setting up a timeout for the maximum duration between the first and last registration of a partition.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zentol commented on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
zentol commented on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-603819561
They are in the intended order; they will be squashed during the merge, I just kept them separate since the constructor changes are rather boring and noisy.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zentol commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r400033791
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImpl.java
##########
@@ -0,0 +1,232 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+/**
+ * Default {@link ResourceManagerPartitionTracker} implementation.
+ *
+ * <p>Internal tracking info must only be updated upon reception of a {@link ClusterPartitionReport}, as the task
+ * executor state is the source of truth.
+ */
+public class ResourceManagerPartitionTrackerImpl implements ResourceManagerPartitionTracker {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ResourceManagerPartitionTrackerImpl.class);
+
+ private final Map<ResourceID, Set<IntermediateDataSetID>> taskExecutorToDataSets = new HashMap<>();
+ private final Map<IntermediateDataSetID, Map<ResourceID, Set<ResultPartitionID>>> dataSetToTaskExecutors = new HashMap<>();
+ private final Map<IntermediateDataSetID, DataSetMetaInfo> dataSetMetaInfo = new HashMap<>();
+ private final Map<IntermediateDataSetID, CompletableFuture<Void>> partitionReleaseCompletionFutures = new HashMap<>();
+
+ private final ClusterPartitionReleaser clusterPartitionReleaser;
+
+ public ResourceManagerPartitionTrackerImpl(ClusterPartitionReleaser clusterPartitionReleaser) {
+ this.clusterPartitionReleaser = clusterPartitionReleaser;
+ }
+
+ @Override
+ public void processTaskExecutorClusterPartitionReport(ResourceID taskExecutorId, ClusterPartitionReport clusterPartitionReport) {
+ Preconditions.checkNotNull(taskExecutorId);
+ Preconditions.checkNotNull(clusterPartitionReport);
+ LOG.debug("Processing cluster partition report from task executor {}: {}.", taskExecutorId, clusterPartitionReport);
+
+ internalProcessClusterPartitionReport(taskExecutorId, clusterPartitionReport);
+ }
+
+ @Override
+ public void processTaskExecutorShutdown(ResourceID taskExecutorId) {
+ Preconditions.checkNotNull(taskExecutorId);
+ LOG.debug("Processing shutdown of task executor {}.", taskExecutorId);
+
+ internalProcessClusterPartitionReport(taskExecutorId, new ClusterPartitionReport(Collections.emptyList()));
+ }
+
+ @Override
+ public CompletableFuture<Void> releaseClusterPartitions(IntermediateDataSetID dataSetId) {
+ Preconditions.checkNotNull(dataSetId);
+ LOG.debug("Releasing cluster partitions for data set {}.", dataSetId);
+
+ CompletableFuture<Void> partitionReleaseCompletionFuture = partitionReleaseCompletionFutures.computeIfAbsent(dataSetId, ignored -> new CompletableFuture<>());
+ internalReleasePartitions(Collections.singleton(dataSetId));
Review comment:
I will add a check at the start of the method that the partition is being tracked. if it isn't we will log something and return a completed future.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0240c441a2ea285e49af3172c2e5829207be011b Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153925949) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365)
* df635bda88e745a98993533481f9a343603bf919 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156483467",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6821",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156536980",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "14b3f8a73814fa648afc42be527a565821a262a8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "14b3f8a73814fa648afc42be527a565821a262a8",
"triggerType" : "PUSH"
}, {
"hash" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"triggerType" : "PUSH"
}, {
"hash" : "8a86aa89a221085749362a508718bd4959ac8190",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6830",
"triggerID" : "8a86aa89a221085749362a508718bd4959ac8190",
"triggerType" : "PUSH"
}, {
"hash" : "8a86aa89a221085749362a508718bd4959ac8190",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156550433",
"triggerID" : "8a86aa89a221085749362a508718bd4959ac8190",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 14b3f8a73814fa648afc42be527a565821a262a8 UNKNOWN
* 38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72 UNKNOWN
* 8a86aa89a221085749362a508718bd4959ac8190 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/156550433) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6830)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0240c441a2ea285e49af3172c2e5829207be011b Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153925949) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r398383525
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImplTest.java
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.collection.IsEmptyCollection.empty;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for the {@link ResourceManagerPartitionTrackerImpl}.
+ */
+public class ResourceManagerPartitionTrackerImplTest extends TestLogger {
+
+ private static final ClusterPartitionReport EMPTY_PARTITION_REPORT = new ClusterPartitionReport(Collections.emptySet());
+
+ private static final ResourceID TASK_EXECUTOR_ID_1 = ResourceID.generate();
+ private static final ResourceID TASK_EXECUTOR_ID_2 = ResourceID.generate();
+ private static final IntermediateDataSetID DATA_SET_ID = new IntermediateDataSetID();
+ private static final ResultPartitionID PARTITION_ID_1 = new ResultPartitionID();
+ private static final ResultPartitionID PARTITION_ID_2 = new ResultPartitionID();
+
+ @Test
+ public void testProcessEmptyClusterPartitionReport() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(TASK_EXECUTOR_ID_1, EMPTY_PARTITION_REPORT);
+ assertThat(partitionReleaser.releaseCalls, empty());
+ }
+
+ /**
+ * Verifies that a task executor hosting multiple partitions of a data set receives a release call if a subset of
+ * its partitions is lost.
+ */
+ @Test
+ public void testReportProcessingWithPartitionLossOnSameTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_1, Collections.singleton(DATA_SET_ID))));
+ }
+
+ /**
+ * Verifies that a task executor hosting partitions of a data set receives a release call if a partition of the
+ * data set is lost on another task executor.
+ */
+ @Test
+ public void testReportProcessingWithPartitionLossOnOtherTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_2,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(TASK_EXECUTOR_ID_1, EMPTY_PARTITION_REPORT);
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_2, Collections.singleton(DATA_SET_ID))));
+ }
+
+ @Test
+ public void testListDataSetsBasics() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ assertEquals(0, tracker.listDataSets().size());
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(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);
+ assertEquals(1, metaInfo.getNumTotalPartitions());
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ EMPTY_PARTITION_REPORT);
+ assertEquals(0, tracker.listDataSets().size());
+ }
+
+ @Test
+ public void testListDataSetsMultiplePartitionsOnSingleTaskExecutor() {
Review comment:
minor: this test tests 3 transitions:
incomplete -> incomplete
incomplete -> complete
complete -> incomplete
I would prefer 3 test cases
some other tests also test multiple things
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r400261308
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerPartitionLifecycleTest.java
##########
@@ -0,0 +1,247 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.resourcemanager;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.heartbeat.HeartbeatServices;
+import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
+import org.apache.flink.runtime.instance.HardwareDescription;
+import org.apache.flink.runtime.io.network.partition.ResourceManagerPartitionTrackerImpl;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.leaderelection.TestingLeaderElectionService;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.registration.RegistrationResponse;
+import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager;
+import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManagerBuilder;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorHeartbeatPayload;
+import org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGatewayBuilder;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.runtime.util.TestingFatalErrorHandler;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for the partition-lifecycle logic in the {@link ResourceManager}.
+ */
+public class ResourceManagerPartitionLifecycleTest extends TestLogger {
Review comment:
Indeed, we want good utility methods/resources and it is an effort. It looks like we have accumulated some technical debt in tests. At this point, it may be only feasible to approach the ideal state of code stepwise. I leave it up to you then whether we can do here any steps or not, it is indeed time-consuming.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r398063719
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImpl.java
##########
@@ -0,0 +1,232 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+/**
+ * Default {@link ResourceManagerPartitionTracker} implementation.
+ *
+ * <p>Internal tracking info must only be updated upon reception of a {@link ClusterPartitionReport}, as the task
+ * executor state is the source of truth.
+ */
+public class ResourceManagerPartitionTrackerImpl implements ResourceManagerPartitionTracker {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ResourceManagerPartitionTrackerImpl.class);
+
+ private final Map<ResourceID, Set<IntermediateDataSetID>> taskExecutorToDataSets = new HashMap<>();
+ private final Map<IntermediateDataSetID, Map<ResourceID, Set<ResultPartitionID>>> dataSetToTaskExecutors = new HashMap<>();
+ private final Map<IntermediateDataSetID, DataSetMetaInfo> dataSetMetaInfo = new HashMap<>();
+ private final Map<IntermediateDataSetID, CompletableFuture<Void>> partitionReleaseCompletionFutures = new HashMap<>();
+
+ private final ClusterPartitionReleaser clusterPartitionReleaser;
+
+ public ResourceManagerPartitionTrackerImpl(ClusterPartitionReleaser clusterPartitionReleaser) {
+ this.clusterPartitionReleaser = clusterPartitionReleaser;
+ }
+
+ @Override
+ public void processTaskExecutorClusterPartitionReport(ResourceID taskExecutorId, ClusterPartitionReport clusterPartitionReport) {
+ Preconditions.checkNotNull(taskExecutorId);
+ Preconditions.checkNotNull(clusterPartitionReport);
+ LOG.debug("Processing cluster partition report from task executor {}: {}.", taskExecutorId, clusterPartitionReport);
+
+ internalProcessClusterPartitionReport(taskExecutorId, clusterPartitionReport);
+ }
+
+ @Override
+ public void processTaskExecutorShutdown(ResourceID taskExecutorId) {
+ Preconditions.checkNotNull(taskExecutorId);
+ LOG.debug("Processing shutdown of task executor {}.", taskExecutorId);
+
+ internalProcessClusterPartitionReport(taskExecutorId, new ClusterPartitionReport(Collections.emptyList()));
+ }
+
+ @Override
+ public CompletableFuture<Void> releaseClusterPartitions(IntermediateDataSetID dataSetId) {
+ Preconditions.checkNotNull(dataSetId);
+ LOG.debug("Releasing cluster partitions for data set {}.", dataSetId);
+
+ CompletableFuture<Void> partitionReleaseCompletionFuture = partitionReleaseCompletionFutures.computeIfAbsent(dataSetId, ignored -> new CompletableFuture<>());
+ internalReleasePartitions(Collections.singleton(dataSetId));
+ return partitionReleaseCompletionFuture;
+ }
+
+ private void internalProcessClusterPartitionReport(ResourceID taskExecutorId, ClusterPartitionReport clusterPartitionReport) {
+ final Set<IntermediateDataSetID> dataSetsWithLostPartitions = clusterPartitionReport.getEntries().isEmpty()
+ ? processEmptyReport(taskExecutorId)
+ : setHostedDataSetsAndCheckCorruption(taskExecutorId, clusterPartitionReport.getEntries());
+
+ updateDataSetMetaData(clusterPartitionReport);
+
+ checkForFullyLostDatasets(dataSetsWithLostPartitions);
+
+ internalReleasePartitions(dataSetsWithLostPartitions);
+ }
+
+ private void internalReleasePartitions(Set<IntermediateDataSetID> dataSetsToRelease) {
+ Map<ResourceID, Set<IntermediateDataSetID>> releaseCalls = prepareReleaseCalls(dataSetsToRelease);
+ releaseCalls.forEach(clusterPartitionReleaser::releaseClusterPartitions);
+ }
+
+ private Set<IntermediateDataSetID> processEmptyReport(ResourceID taskExecutorId) {
+ Set<IntermediateDataSetID> previouslyHostedDatasets = taskExecutorToDataSets.remove(taskExecutorId);
+ if (previouslyHostedDatasets == null) {
+ // default path for task executors that never have any cluster partitions
+ previouslyHostedDatasets = Collections.emptySet();
+ } else {
+ previouslyHostedDatasets.forEach(dataSetId -> removeInnerKey(dataSetId, taskExecutorId, dataSetToTaskExecutors));
+ }
+ return previouslyHostedDatasets;
+ }
+
+ /**
+ * Updates the data sets for which the given task executor is hosting partitions and returns data sets that were
+ * corrupted due to a loss of partitions.
+ *
+ * @param taskExecutorId ID of the hosting TaskExecutor
+ * @param reportEntries IDs of data sets for which partitions are hosted
+ * @return corrupted data sets
+ */
+ private Set<IntermediateDataSetID> setHostedDataSetsAndCheckCorruption(ResourceID taskExecutorId, Collection<ClusterPartitionReport.ClusterPartitionReportEntry> reportEntries) {
+ final Set<IntermediateDataSetID> currentlyHostedDatasets = reportEntries
+ .stream()
+ .map(ClusterPartitionReport.ClusterPartitionReportEntry::getDataSetId)
+ .collect(Collectors.toSet());
+
+ final Set<IntermediateDataSetID> previouslyHostedDataSets = taskExecutorToDataSets.put(
+ taskExecutorId,
+ currentlyHostedDatasets);
+
+ // previously tracked data sets may be corrupted since we may be tracking less partitions than before
+ final Set<IntermediateDataSetID> potentiallyCorruptedDataSets = Optional
+ .ofNullable(previouslyHostedDataSets)
+ .orElse(new HashSet<>(0));
+
+ // update data set -> task executor mapping and find datasets for which lost a partition
+ reportEntries.forEach(hostedPartition -> {
+ final Map<ResourceID, Set<ResultPartitionID>> taskExecutorHosts = dataSetToTaskExecutors.computeIfAbsent(hostedPartition.getDataSetId(), ignored -> new HashMap<>());
+ final Set<ResultPartitionID> previouslyHostedPartitions = taskExecutorHosts.put(taskExecutorId, hostedPartition.getHostedPartitions());
+
+ final boolean noPartitionLost = previouslyHostedPartitions == null || hostedPartition.getHostedPartitions().containsAll(previouslyHostedPartitions);
+ if (noPartitionLost) {
+ potentiallyCorruptedDataSets.remove(hostedPartition.getDataSetId());
+ }
+ });
+
+ // now only contains data sets for which a partition is no longer tracked
+ return potentiallyCorruptedDataSets;
+ }
+
+ private void updateDataSetMetaData(ClusterPartitionReport clusterPartitionReport) {
+ // add meta info for new data sets
+ clusterPartitionReport.getEntries().forEach(entry ->
+ dataSetMetaInfo.compute(entry.getDataSetId(), (dataSetID, dataSetMetaInfo) -> {
+ if (dataSetMetaInfo == null) {
+ return new DataSetMetaInfo(entry.getNumTotalPartitions());
+ } else {
+ // double check that the meta data is consistent
+ Preconditions.checkState(dataSetMetaInfo.getNumTotalPartitions() == entry.getNumTotalPartitions());
+ return dataSetMetaInfo;
+ }
+ }));
+ }
+
+ private void checkForFullyLostDatasets(Set<IntermediateDataSetID> dataSetsWithLostPartitions) {
+ dataSetsWithLostPartitions.forEach(dataSetId -> {
+ if (getHostingTaskExecutors(dataSetId).isEmpty()) {
+ LOG.debug("There are no longer partitions being tracked for dataset {}.", dataSetId);
+ dataSetMetaInfo.remove(dataSetId);
+ Optional.ofNullable(partitionReleaseCompletionFutures.get(dataSetId)).map(future -> future.complete(null));
+ }
+ });
+ }
+
+ private Map<ResourceID, Set<IntermediateDataSetID>> prepareReleaseCalls(Set<IntermediateDataSetID> dataSetsToRelease) {
+ final Map<ResourceID, Set<IntermediateDataSetID>> releaseCalls = new HashMap<>();
+ dataSetsToRelease.forEach(dataSetToRelease -> {
+ final Set<ResourceID> hostingTaskExecutors = getHostingTaskExecutors(dataSetToRelease);
+ hostingTaskExecutors.forEach(hostingTaskExecutor -> insert(hostingTaskExecutor, dataSetToRelease, releaseCalls));
+ });
+ return releaseCalls;
+ }
+
+ private Set<ResourceID> getHostingTaskExecutors(IntermediateDataSetID dataSetId) {
+ Preconditions.checkNotNull(dataSetId);
+
+ Map<ResourceID, Set<ResultPartitionID>> trackedPartitions = dataSetToTaskExecutors.get(dataSetId);
+ if (trackedPartitions == null) {
+ return Collections.emptySet();
+ } else {
+ return trackedPartitions.keySet();
+ }
+ }
+
+ @Override
+ public Map<IntermediateDataSetID, DataSetMetaInfo> listDataSets() {
+ return dataSetMetaInfo.entrySet().stream()
+ .filter(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());
+
+ int numTrackedPartitions = 0;
+ for (Set<ResultPartitionID> hostedPartitions : taskExecutorToPartitions.values()) {
+ numTrackedPartitions += hostedPartitions.size();
+ }
Review comment:
```suggestion
final int numTrackedPartitions = taskExecutorToPartitions.values().stream().mapToInt(Set::size).sum();
```
nit idea, as you seem to like streams :)
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156483467",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6821",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156536980",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "14b3f8a73814fa648afc42be527a565821a262a8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "14b3f8a73814fa648afc42be527a565821a262a8",
"triggerType" : "PUSH"
}, {
"hash" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"triggerType" : "PUSH"
}, {
"hash" : "8a86aa89a221085749362a508718bd4959ac8190",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6830",
"triggerID" : "8a86aa89a221085749362a508718bd4959ac8190",
"triggerType" : "PUSH"
}, {
"hash" : "8a86aa89a221085749362a508718bd4959ac8190",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156550433",
"triggerID" : "8a86aa89a221085749362a508718bd4959ac8190",
"triggerType" : "PUSH"
}, {
"hash" : "998a1c6d2107addf7bfafd0114b2f696837134aa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "998a1c6d2107addf7bfafd0114b2f696837134aa",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 14b3f8a73814fa648afc42be527a565821a262a8 UNKNOWN
* 38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72 UNKNOWN
* 8a86aa89a221085749362a508718bd4959ac8190 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/156550433) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6830)
* 998a1c6d2107addf7bfafd0114b2f696837134aa UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156483467",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6821",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* df635bda88e745a98993533481f9a343603bf919 Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/156483467) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6821)
* 3e149786a3b54c141ecd14b99d98062d31f56bb4 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r397769498
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ClusterPartitionReleaser.java
##########
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+
+import java.util.Set;
+
+/**
+ * Interface for releasing cluster partitions on a task executor.
+ */
+public interface ClusterPartitionReleaser {
Review comment:
```suggestion
@FunctionalInterface
public interface ClusterPartitionReleaser {
```
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r398381553
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImplTest.java
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.collection.IsEmptyCollection.empty;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for the {@link ResourceManagerPartitionTrackerImpl}.
+ */
+public class ResourceManagerPartitionTrackerImplTest extends TestLogger {
Review comment:
For this particular component, I would also suggest to test that the freeing of internal maps is checked because their sizes are somewhat side effect of the component in the system. They could be injected in constructor for testing or checked over some other test API. As mentioned in another comment, not sure that e.g. partitionReleaseCompletionFutures is freed properly and we do not accumulate data there over time.
I was also thinking to use listDataSets for this check but it seems to be tricky.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r397868553
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTracker.java
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
Review comment:
a bit strange that this all resides in network package
I would expect it to be somewhere in `org.apache.flink.runtime.resourcemanager. partition`.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r401424021
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImplTest.java
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.collection.IsEmptyCollection.empty;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for the {@link ResourceManagerPartitionTrackerImpl}.
+ */
+public class ResourceManagerPartitionTrackerImplTest extends TestLogger {
+
+ private static final ClusterPartitionReport EMPTY_PARTITION_REPORT = new ClusterPartitionReport(Collections.emptySet());
+
+ private static final ResourceID TASK_EXECUTOR_ID_1 = ResourceID.generate();
+ private static final ResourceID TASK_EXECUTOR_ID_2 = ResourceID.generate();
+ private static final IntermediateDataSetID DATA_SET_ID = new IntermediateDataSetID();
+ private static final ResultPartitionID PARTITION_ID_1 = new ResultPartitionID();
+ private static final ResultPartitionID PARTITION_ID_2 = new ResultPartitionID();
+
+ @Test
+ public void testProcessEmptyClusterPartitionReport() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(TASK_EXECUTOR_ID_1, EMPTY_PARTITION_REPORT);
+ assertThat(partitionReleaser.releaseCalls, empty());
+ }
+
+ /**
+ * Verifies that a task executor hosting multiple partitions of a data set receives a release call if a subset of
+ * its partitions is lost.
+ */
+ @Test
+ public void testReportProcessingWithPartitionLossOnSameTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_1, Collections.singleton(DATA_SET_ID))));
+ }
+
+ /**
+ * Verifies that a task executor hosting partitions of a data set receives a release call if a partition of the
+ * data set is lost on another task executor.
+ */
+ @Test
+ public void testReportProcessingWithPartitionLossOnOtherTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_2,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(TASK_EXECUTOR_ID_1, EMPTY_PARTITION_REPORT);
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_2, Collections.singleton(DATA_SET_ID))));
+ }
+
+ @Test
+ public void testListDataSetsBasics() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ assertEquals(0, tracker.listDataSets().size());
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(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);
+ assertEquals(1, metaInfo.getNumTotalPartitions());
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ EMPTY_PARTITION_REPORT);
+ assertEquals(0, tracker.listDataSets().size());
+ }
+
+ @Test
+ public void testListDataSetsMultiplePartitionsOnSingleTaskExecutor() {
Review comment:
Indeed, there is going to be some code duplication. The idea was to reduce its size with the `report` method (which happened) in favour of isolated checks at the end. I am ok to keep it like this.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r398028126
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImpl.java
##########
@@ -0,0 +1,232 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+/**
+ * Default {@link ResourceManagerPartitionTracker} implementation.
+ *
+ * <p>Internal tracking info must only be updated upon reception of a {@link ClusterPartitionReport}, as the task
+ * executor state is the source of truth.
+ */
+public class ResourceManagerPartitionTrackerImpl implements ResourceManagerPartitionTracker {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ResourceManagerPartitionTrackerImpl.class);
+
+ private final Map<ResourceID, Set<IntermediateDataSetID>> taskExecutorToDataSets = new HashMap<>();
+ private final Map<IntermediateDataSetID, Map<ResourceID, Set<ResultPartitionID>>> dataSetToTaskExecutors = new HashMap<>();
+ private final Map<IntermediateDataSetID, DataSetMetaInfo> dataSetMetaInfo = new HashMap<>();
+ private final Map<IntermediateDataSetID, CompletableFuture<Void>> partitionReleaseCompletionFutures = new HashMap<>();
+
+ private final ClusterPartitionReleaser clusterPartitionReleaser;
+
+ public ResourceManagerPartitionTrackerImpl(ClusterPartitionReleaser clusterPartitionReleaser) {
+ this.clusterPartitionReleaser = clusterPartitionReleaser;
+ }
+
+ @Override
+ public void processTaskExecutorClusterPartitionReport(ResourceID taskExecutorId, ClusterPartitionReport clusterPartitionReport) {
+ Preconditions.checkNotNull(taskExecutorId);
+ Preconditions.checkNotNull(clusterPartitionReport);
+ LOG.debug("Processing cluster partition report from task executor {}: {}.", taskExecutorId, clusterPartitionReport);
+
+ internalProcessClusterPartitionReport(taskExecutorId, clusterPartitionReport);
+ }
+
+ @Override
+ public void processTaskExecutorShutdown(ResourceID taskExecutorId) {
+ Preconditions.checkNotNull(taskExecutorId);
+ LOG.debug("Processing shutdown of task executor {}.", taskExecutorId);
+
+ internalProcessClusterPartitionReport(taskExecutorId, new ClusterPartitionReport(Collections.emptyList()));
+ }
+
+ @Override
+ public CompletableFuture<Void> releaseClusterPartitions(IntermediateDataSetID dataSetId) {
+ Preconditions.checkNotNull(dataSetId);
+ LOG.debug("Releasing cluster partitions for data set {}.", dataSetId);
+
+ CompletableFuture<Void> partitionReleaseCompletionFuture = partitionReleaseCompletionFutures.computeIfAbsent(dataSetId, ignored -> new CompletableFuture<>());
+ internalReleasePartitions(Collections.singleton(dataSetId));
+ return partitionReleaseCompletionFuture;
+ }
+
+ private void internalProcessClusterPartitionReport(ResourceID taskExecutorId, ClusterPartitionReport clusterPartitionReport) {
+ final Set<IntermediateDataSetID> dataSetsWithLostPartitions = clusterPartitionReport.getEntries().isEmpty()
+ ? processEmptyReport(taskExecutorId)
+ : setHostedDataSetsAndCheckCorruption(taskExecutorId, clusterPartitionReport.getEntries());
+
+ updateDataSetMetaData(clusterPartitionReport);
+
+ checkForFullyLostDatasets(dataSetsWithLostPartitions);
+
+ internalReleasePartitions(dataSetsWithLostPartitions);
+ }
+
+ private void internalReleasePartitions(Set<IntermediateDataSetID> dataSetsToRelease) {
+ Map<ResourceID, Set<IntermediateDataSetID>> releaseCalls = prepareReleaseCalls(dataSetsToRelease);
+ releaseCalls.forEach(clusterPartitionReleaser::releaseClusterPartitions);
+ }
+
+ private Set<IntermediateDataSetID> processEmptyReport(ResourceID taskExecutorId) {
+ Set<IntermediateDataSetID> previouslyHostedDatasets = taskExecutorToDataSets.remove(taskExecutorId);
+ if (previouslyHostedDatasets == null) {
+ // default path for task executors that never have any cluster partitions
+ previouslyHostedDatasets = Collections.emptySet();
+ } else {
+ previouslyHostedDatasets.forEach(dataSetId -> removeInnerKey(dataSetId, taskExecutorId, dataSetToTaskExecutors));
+ }
+ return previouslyHostedDatasets;
+ }
+
+ /**
+ * Updates the data sets for which the given task executor is hosting partitions and returns data sets that were
+ * corrupted due to a loss of partitions.
+ *
+ * @param taskExecutorId ID of the hosting TaskExecutor
+ * @param reportEntries IDs of data sets for which partitions are hosted
+ * @return corrupted data sets
+ */
+ private Set<IntermediateDataSetID> setHostedDataSetsAndCheckCorruption(ResourceID taskExecutorId, Collection<ClusterPartitionReport.ClusterPartitionReportEntry> reportEntries) {
+ final Set<IntermediateDataSetID> currentlyHostedDatasets = reportEntries
+ .stream()
+ .map(ClusterPartitionReport.ClusterPartitionReportEntry::getDataSetId)
+ .collect(Collectors.toSet());
+
+ final Set<IntermediateDataSetID> previouslyHostedDataSets = taskExecutorToDataSets.put(
+ taskExecutorId,
+ currentlyHostedDatasets);
+
+ // previously tracked data sets may be corrupted since we may be tracking less partitions than before
+ final Set<IntermediateDataSetID> potentiallyCorruptedDataSets = Optional
+ .ofNullable(previouslyHostedDataSets)
+ .orElse(new HashSet<>(0));
+
+ // update data set -> task executor mapping and find datasets for which lost a partition
+ reportEntries.forEach(hostedPartition -> {
+ final Map<ResourceID, Set<ResultPartitionID>> taskExecutorHosts = dataSetToTaskExecutors.computeIfAbsent(hostedPartition.getDataSetId(), ignored -> new HashMap<>());
+ final Set<ResultPartitionID> previouslyHostedPartitions = taskExecutorHosts.put(taskExecutorId, hostedPartition.getHostedPartitions());
+
+ final boolean noPartitionLost = previouslyHostedPartitions == null || hostedPartition.getHostedPartitions().containsAll(previouslyHostedPartitions);
+ if (noPartitionLost) {
+ potentiallyCorruptedDataSets.remove(hostedPartition.getDataSetId());
+ }
+ });
+
+ // now only contains data sets for which a partition is no longer tracked
+ return potentiallyCorruptedDataSets;
+ }
+
+ private void updateDataSetMetaData(ClusterPartitionReport clusterPartitionReport) {
+ // add meta info for new data sets
+ clusterPartitionReport.getEntries().forEach(entry ->
+ dataSetMetaInfo.compute(entry.getDataSetId(), (dataSetID, dataSetMetaInfo) -> {
+ if (dataSetMetaInfo == null) {
+ return new DataSetMetaInfo(entry.getNumTotalPartitions());
+ } else {
+ // double check that the meta data is consistent
+ Preconditions.checkState(dataSetMetaInfo.getNumTotalPartitions() == entry.getNumTotalPartitions());
+ return dataSetMetaInfo;
+ }
+ }));
+ }
+
+ private void checkForFullyLostDatasets(Set<IntermediateDataSetID> dataSetsWithLostPartitions) {
+ dataSetsWithLostPartitions.forEach(dataSetId -> {
+ if (getHostingTaskExecutors(dataSetId).isEmpty()) {
+ LOG.debug("There are no longer partitions being tracked for dataset {}.", dataSetId);
+ dataSetMetaInfo.remove(dataSetId);
+ Optional.ofNullable(partitionReleaseCompletionFutures.get(dataSetId)).map(future -> future.complete(null));
Review comment:
do we also want to GC the completed future from `partitionReleaseCompletionFutures`?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156483467",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6821",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156536980",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "14b3f8a73814fa648afc42be527a565821a262a8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "14b3f8a73814fa648afc42be527a565821a262a8",
"triggerType" : "PUSH"
}, {
"hash" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"triggerType" : "PUSH"
}, {
"hash" : "8a86aa89a221085749362a508718bd4959ac8190",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6830",
"triggerID" : "8a86aa89a221085749362a508718bd4959ac8190",
"triggerType" : "PUSH"
}, {
"hash" : "8a86aa89a221085749362a508718bd4959ac8190",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156550433",
"triggerID" : "8a86aa89a221085749362a508718bd4959ac8190",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 14b3f8a73814fa648afc42be527a565821a262a8 UNKNOWN
* 38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72 UNKNOWN
* 8a86aa89a221085749362a508718bd4959ac8190 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/156550433) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6830)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zentol commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r399074456
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerPartitionLifecycleTest.java
##########
@@ -0,0 +1,247 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.resourcemanager;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.heartbeat.HeartbeatServices;
+import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
+import org.apache.flink.runtime.instance.HardwareDescription;
+import org.apache.flink.runtime.io.network.partition.ResourceManagerPartitionTrackerImpl;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.leaderelection.TestingLeaderElectionService;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.registration.RegistrationResponse;
+import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager;
+import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManagerBuilder;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorHeartbeatPayload;
+import org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGatewayBuilder;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.runtime.util.TestingFatalErrorHandler;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for the partition-lifecycle logic in the {@link ResourceManager}.
+ */
+public class ResourceManagerPartitionLifecycleTest extends TestLogger {
Review comment:
Possibly, most of the code was in fact copied over. I'm not a huge fan of making methods from other tests accessible just because they _happen to now_ match the same use-case.
What we really want are some flexible utility methods/resources, but this would be quite an effor.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0240c441a2ea285e49af3172c2e5829207be011b UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zentol commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r400045114
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImplTest.java
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.collection.IsEmptyCollection.empty;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for the {@link ResourceManagerPartitionTrackerImpl}.
+ */
+public class ResourceManagerPartitionTrackerImplTest extends TestLogger {
Review comment:
I will add a `VisibleForTesting` method for checking whether all maps are empty.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156483467",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6821",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "CANCELED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156536980",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "14b3f8a73814fa648afc42be527a565821a262a8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "14b3f8a73814fa648afc42be527a565821a262a8",
"triggerType" : "PUSH"
}, {
"hash" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3e149786a3b54c141ecd14b99d98062d31f56bb4 Travis: [CANCELED](https://travis-ci.com/github/flink-ci/flink/builds/156536980) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827)
* 14b3f8a73814fa648afc42be527a565821a262a8 UNKNOWN
* 38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0240c441a2ea285e49af3172c2e5829207be011b Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153925949) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r398397956
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerPartitionLifecycleTest.java
##########
@@ -0,0 +1,247 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.resourcemanager;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.heartbeat.HeartbeatServices;
+import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
+import org.apache.flink.runtime.instance.HardwareDescription;
+import org.apache.flink.runtime.io.network.partition.ResourceManagerPartitionTrackerImpl;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.leaderelection.TestingLeaderElectionService;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.registration.RegistrationResponse;
+import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager;
+import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManagerBuilder;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorHeartbeatPayload;
+import org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGatewayBuilder;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.runtime.util.TestingFatalErrorHandler;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for the partition-lifecycle logic in the {@link ResourceManager}.
+ */
+public class ResourceManagerPartitionLifecycleTest extends TestLogger {
Review comment:
Could we reuse some code from `ResourceManagerTest`?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600653689
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 0240c441a2ea285e49af3172c2e5829207be011b (Wed Mar 18 14:26:12 UTC 2020)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156483467",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6821",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156536980",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "14b3f8a73814fa648afc42be527a565821a262a8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "14b3f8a73814fa648afc42be527a565821a262a8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3e149786a3b54c141ecd14b99d98062d31f56bb4 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/156536980) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827)
* 14b3f8a73814fa648afc42be527a565821a262a8 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r398362604
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImplTest.java
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.collection.IsEmptyCollection.empty;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for the {@link ResourceManagerPartitionTrackerImpl}.
+ */
+public class ResourceManagerPartitionTrackerImplTest extends TestLogger {
+
+ private static final ClusterPartitionReport EMPTY_PARTITION_REPORT = new ClusterPartitionReport(Collections.emptySet());
+
+ private static final ResourceID TASK_EXECUTOR_ID_1 = ResourceID.generate();
+ private static final ResourceID TASK_EXECUTOR_ID_2 = ResourceID.generate();
+ private static final IntermediateDataSetID DATA_SET_ID = new IntermediateDataSetID();
+ private static final ResultPartitionID PARTITION_ID_1 = new ResultPartitionID();
+ private static final ResultPartitionID PARTITION_ID_2 = new ResultPartitionID();
+
+ @Test
+ public void testProcessEmptyClusterPartitionReport() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(TASK_EXECUTOR_ID_1, EMPTY_PARTITION_REPORT);
+ assertThat(partitionReleaser.releaseCalls, empty());
+ }
+
+ /**
+ * Verifies that a task executor hosting multiple partitions of a data set receives a release call if a subset of
+ * its partitions is lost.
+ */
+ @Test
+ public void testReportProcessingWithPartitionLossOnSameTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_1, Collections.singleton(DATA_SET_ID))));
+ }
+
+ /**
+ * Verifies that a task executor hosting partitions of a data set receives a release call if a partition of the
+ * data set is lost on another task executor.
+ */
+ @Test
+ public void testReportProcessingWithPartitionLossOnOtherTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_2,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(TASK_EXECUTOR_ID_1, EMPTY_PARTITION_REPORT);
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_2, Collections.singleton(DATA_SET_ID))));
+ }
+
+ @Test
+ public void testListDataSetsBasics() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ assertEquals(0, tracker.listDataSets().size());
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(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);
+ assertEquals(1, metaInfo.getNumTotalPartitions());
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ EMPTY_PARTITION_REPORT);
+ assertEquals(0, tracker.listDataSets().size());
+ }
+
+ @Test
+ public void testListDataSetsMultiplePartitionsOnSingleTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ // data set consists of 2 partitions but only 1 is being tracked -> incomplete and should not be listed (yet)
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1));
+ assertEquals(0, tracker.listDataSets().size());
+
+ // start tracking another partitions, but we lost partition 1 so the data set is still incomplete
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+ assertEquals(0, tracker.listDataSets().size());
+
+ // dataset is considered complete since all partitions are being tracked
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1, PARTITION_ID_2));
+ final Map<IntermediateDataSetID, DataSetMetaInfo> listing = tracker.listDataSets();
+ assertThat(listing, hasKey(DATA_SET_ID));
+
+ // dataset is no longer considered complete since partition 2 was lost
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1));
+ assertEquals(0, tracker.listDataSets().size());
+ }
+
+ @Test
+ public void testListDataSetsMultiplePartitionsAcrossTaskExecutors() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1));
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_2,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+ final Map<IntermediateDataSetID, DataSetMetaInfo> listing = tracker.listDataSets();
+ assertThat(listing, hasKey(DATA_SET_ID));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ EMPTY_PARTITION_REPORT);
+ assertEquals(0, tracker.listDataSets().size());
+ }
+
+ @Test
+ public void testReleasePartition() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_2,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ tracker.releaseClusterPartitions(DATA_SET_ID);
+
+ assertThat(partitionReleaser.releaseCalls, containsInAnyOrder(
+ Tuple2.of(TASK_EXECUTOR_ID_1, Collections.singleton(DATA_SET_ID)),
+ Tuple2.of(TASK_EXECUTOR_ID_2, Collections.singleton(DATA_SET_ID))));
+
+ // the data set should still be tracked, since the partition release was not confirmed yet by the task executors
+ assertThat(tracker.listDataSets().keySet(), contains(DATA_SET_ID));
+ }
+
+ @Test
+ public void testShutdownProcessing() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorShutdown(TASK_EXECUTOR_ID_1);
+ assertThat(partitionReleaser.releaseCalls, empty());
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 3, PARTITION_ID_1, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_2,
+ createClusterPartitionReport(DATA_SET_ID, 3, new ResultPartitionID()));
+
+ tracker.processTaskExecutorShutdown(TASK_EXECUTOR_ID_1);
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_2, Collections.singleton(DATA_SET_ID))));
+ }
+
+ private static ClusterPartitionReport createClusterPartitionReport(IntermediateDataSetID dataSetId, int numTotalPartitions, ResultPartitionID... partitionId) {
+ return new ClusterPartitionReport(Collections.singletonList(
+ new ClusterPartitionReport.ClusterPartitionReportEntry(
+ dataSetId,
+ new HashSet<>(Arrays.asList(partitionId)),
+ numTotalPartitions)));
+ }
+
+ private static class TestClusterPartitionReleaser implements ClusterPartitionReleaser {
+
+ final List<Tuple2<ResourceID, Set<IntermediateDataSetID>>> releaseCalls = new ArrayList<>();
Review comment:
nit: can be private
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r397768198
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerFactory.java
##########
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+/**
+ * Factory for {@link ResourceManagerPartitionTracker}.
+ */
+public interface ResourceManagerPartitionTrackerFactory {
Review comment:
```suggestion
@FunctionalInterface
public interface ResourceManagerPartitionTrackerFactory
```
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156483467",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6821",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156536980",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "14b3f8a73814fa648afc42be527a565821a262a8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "14b3f8a73814fa648afc42be527a565821a262a8",
"triggerType" : "PUSH"
}, {
"hash" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"triggerType" : "PUSH"
}, {
"hash" : "8a86aa89a221085749362a508718bd4959ac8190",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6830",
"triggerID" : "8a86aa89a221085749362a508718bd4959ac8190",
"triggerType" : "PUSH"
}, {
"hash" : "8a86aa89a221085749362a508718bd4959ac8190",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156550433",
"triggerID" : "8a86aa89a221085749362a508718bd4959ac8190",
"triggerType" : "PUSH"
}, {
"hash" : "998a1c6d2107addf7bfafd0114b2f696837134aa",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6981",
"triggerID" : "998a1c6d2107addf7bfafd0114b2f696837134aa",
"triggerType" : "PUSH"
}, {
"hash" : "998a1c6d2107addf7bfafd0114b2f696837134aa",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/157731336",
"triggerID" : "998a1c6d2107addf7bfafd0114b2f696837134aa",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 14b3f8a73814fa648afc42be527a565821a262a8 UNKNOWN
* 38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72 UNKNOWN
* 8a86aa89a221085749362a508718bd4959ac8190 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/156550433) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6830)
* 998a1c6d2107addf7bfafd0114b2f696837134aa Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/157731336) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6981)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zentol commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r400035559
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImplTest.java
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.collection.IsEmptyCollection.empty;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for the {@link ResourceManagerPartitionTrackerImpl}.
+ */
+public class ResourceManagerPartitionTrackerImplTest extends TestLogger {
+
+ private static final ClusterPartitionReport EMPTY_PARTITION_REPORT = new ClusterPartitionReport(Collections.emptySet());
+
+ private static final ResourceID TASK_EXECUTOR_ID_1 = ResourceID.generate();
+ private static final ResourceID TASK_EXECUTOR_ID_2 = ResourceID.generate();
+ private static final IntermediateDataSetID DATA_SET_ID = new IntermediateDataSetID();
+ private static final ResultPartitionID PARTITION_ID_1 = new ResultPartitionID();
+ private static final ResultPartitionID PARTITION_ID_2 = new ResultPartitionID();
+
+ @Test
+ public void testProcessEmptyClusterPartitionReport() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(TASK_EXECUTOR_ID_1, EMPTY_PARTITION_REPORT);
+ assertThat(partitionReleaser.releaseCalls, empty());
+ }
+
+ /**
+ * Verifies that a task executor hosting multiple partitions of a data set receives a release call if a subset of
+ * its partitions is lost.
+ */
+ @Test
+ public void testReportProcessingWithPartitionLossOnSameTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_1, Collections.singleton(DATA_SET_ID))));
+ }
+
+ /**
+ * Verifies that a task executor hosting partitions of a data set receives a release call if a partition of the
+ * data set is lost on another task executor.
+ */
+ @Test
+ public void testReportProcessingWithPartitionLossOnOtherTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_2,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(TASK_EXECUTOR_ID_1, EMPTY_PARTITION_REPORT);
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_2, Collections.singleton(DATA_SET_ID))));
+ }
+
+ @Test
+ public void testListDataSetsBasics() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ assertEquals(0, tracker.listDataSets().size());
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(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);
+ assertEquals(1, metaInfo.getNumTotalPartitions());
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ EMPTY_PARTITION_REPORT);
+ assertEquals(0, tracker.listDataSets().size());
+ }
+
+ @Test
+ public void testListDataSetsMultiplePartitionsOnSingleTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ // data set consists of 2 partitions but only 1 is being tracked -> incomplete and should not be listed (yet)
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1));
+ assertEquals(0, tracker.listDataSets().size());
+
+ // start tracking another partitions, but we lost partition 1 so the data set is still incomplete
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+ assertEquals(0, tracker.listDataSets().size());
+
+ // dataset is considered complete since all partitions are being tracked
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1, PARTITION_ID_2));
+ final Map<IntermediateDataSetID, DataSetMetaInfo> listing = tracker.listDataSets();
+ assertThat(listing, hasKey(DATA_SET_ID));
+
+ // dataset is no longer considered complete since partition 2 was lost
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1));
+ assertEquals(0, tracker.listDataSets().size());
+ }
+
+ @Test
+ public void testListDataSetsMultiplePartitionsAcrossTaskExecutors() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1));
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_2,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+ final Map<IntermediateDataSetID, DataSetMetaInfo> listing = tracker.listDataSets();
+ assertThat(listing, hasKey(DATA_SET_ID));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ EMPTY_PARTITION_REPORT);
+ assertEquals(0, tracker.listDataSets().size());
+ }
+
+ @Test
+ public void testReleasePartition() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_2,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ tracker.releaseClusterPartitions(DATA_SET_ID);
+
+ assertThat(partitionReleaser.releaseCalls, containsInAnyOrder(
+ Tuple2.of(TASK_EXECUTOR_ID_1, Collections.singleton(DATA_SET_ID)),
+ Tuple2.of(TASK_EXECUTOR_ID_2, Collections.singleton(DATA_SET_ID))));
+
+ // the data set should still be tracked, since the partition release was not confirmed yet by the task executors
+ assertThat(tracker.listDataSets().keySet(), contains(DATA_SET_ID));
+ }
+
+ @Test
+ public void testShutdownProcessing() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorShutdown(TASK_EXECUTOR_ID_1);
+ assertThat(partitionReleaser.releaseCalls, empty());
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 3, PARTITION_ID_1, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_2,
+ createClusterPartitionReport(DATA_SET_ID, 3, new ResultPartitionID()));
+
+ tracker.processTaskExecutorShutdown(TASK_EXECUTOR_ID_1);
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_2, Collections.singleton(DATA_SET_ID))));
+ }
+
+ private static ClusterPartitionReport createClusterPartitionReport(IntermediateDataSetID dataSetId, int numTotalPartitions, ResultPartitionID... partitionId) {
+ return new ClusterPartitionReport(Collections.singletonList(
+ new ClusterPartitionReport.ClusterPartitionReportEntry(
+ dataSetId,
+ new HashSet<>(Arrays.asList(partitionId)),
+ numTotalPartitions)));
+ }
+
+ private static class TestClusterPartitionReleaser implements ClusterPartitionReleaser {
+
+ final List<Tuple2<ResourceID, Set<IntermediateDataSetID>>> releaseCalls = new ArrayList<>();
Review comment:
it is a field that is purposefully accessed from the outside and hence shouldn't be private for clarity reasons.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156483467",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6821",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "CANCELED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156536980",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "14b3f8a73814fa648afc42be527a565821a262a8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "14b3f8a73814fa648afc42be527a565821a262a8",
"triggerType" : "PUSH"
}, {
"hash" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"triggerType" : "PUSH"
}, {
"hash" : "8a86aa89a221085749362a508718bd4959ac8190",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6830",
"triggerID" : "8a86aa89a221085749362a508718bd4959ac8190",
"triggerType" : "PUSH"
}, {
"hash" : "8a86aa89a221085749362a508718bd4959ac8190",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156550433",
"triggerID" : "8a86aa89a221085749362a508718bd4959ac8190",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3e149786a3b54c141ecd14b99d98062d31f56bb4 Travis: [CANCELED](https://travis-ci.com/github/flink-ci/flink/builds/156536980) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827)
* 14b3f8a73814fa648afc42be527a565821a262a8 UNKNOWN
* 38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72 UNKNOWN
* 8a86aa89a221085749362a508718bd4959ac8190 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/156550433) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6830)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156483467",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6821",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156536980",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3e149786a3b54c141ecd14b99d98062d31f56bb4 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/156536980) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zentol commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r400727826
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImplTest.java
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.collection.IsEmptyCollection.empty;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for the {@link ResourceManagerPartitionTrackerImpl}.
+ */
+public class ResourceManagerPartitionTrackerImplTest extends TestLogger {
+
+ private static final ClusterPartitionReport EMPTY_PARTITION_REPORT = new ClusterPartitionReport(Collections.emptySet());
+
+ private static final ResourceID TASK_EXECUTOR_ID_1 = ResourceID.generate();
+ private static final ResourceID TASK_EXECUTOR_ID_2 = ResourceID.generate();
+ private static final IntermediateDataSetID DATA_SET_ID = new IntermediateDataSetID();
+ private static final ResultPartitionID PARTITION_ID_1 = new ResultPartitionID();
+ private static final ResultPartitionID PARTITION_ID_2 = new ResultPartitionID();
+
+ @Test
+ public void testProcessEmptyClusterPartitionReport() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(TASK_EXECUTOR_ID_1, EMPTY_PARTITION_REPORT);
+ assertThat(partitionReleaser.releaseCalls, empty());
+ }
+
+ /**
+ * Verifies that a task executor hosting multiple partitions of a data set receives a release call if a subset of
+ * its partitions is lost.
+ */
+ @Test
+ public void testReportProcessingWithPartitionLossOnSameTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_1, Collections.singleton(DATA_SET_ID))));
+ }
+
+ /**
+ * Verifies that a task executor hosting partitions of a data set receives a release call if a partition of the
+ * data set is lost on another task executor.
+ */
+ @Test
+ public void testReportProcessingWithPartitionLossOnOtherTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_2,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(TASK_EXECUTOR_ID_1, EMPTY_PARTITION_REPORT);
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_2, Collections.singleton(DATA_SET_ID))));
+ }
+
+ @Test
+ public void testListDataSetsBasics() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ assertEquals(0, tracker.listDataSets().size());
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(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);
+ assertEquals(1, metaInfo.getNumTotalPartitions());
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ EMPTY_PARTITION_REPORT);
+ assertEquals(0, tracker.listDataSets().size());
+ }
+
+ @Test
+ public void testListDataSetsMultiplePartitionsOnSingleTaskExecutor() {
Review comment:
I've addressed your second comment (I think).
As for the other one, your state transitions aren't quite correct, or rather overly simplified.
It goes from `unknown -> partially-complete -> complete -> partially-complete`.
Trying to split these up just means we duplicate code we already have. You can't go `partially-complete -> complete` without first having to go `unknown -> partially-complete`, so if you were to write separate tests you're running the same code `unknown -> partially-complete` twice.
And this is true for all state transitions.
Assertions are usable enough for us to differentiate between failures at different stages, so I don't buy in to the whole "test exactly one thing" mentality.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156483467",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6821",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* df635bda88e745a98993533481f9a343603bf919 Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/156483467) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6821)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r398389868
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImplTest.java
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.collection.IsEmptyCollection.empty;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for the {@link ResourceManagerPartitionTrackerImpl}.
+ */
+public class ResourceManagerPartitionTrackerImplTest extends TestLogger {
+
+ private static final ClusterPartitionReport EMPTY_PARTITION_REPORT = new ClusterPartitionReport(Collections.emptySet());
+
+ private static final ResourceID TASK_EXECUTOR_ID_1 = ResourceID.generate();
+ private static final ResourceID TASK_EXECUTOR_ID_2 = ResourceID.generate();
+ private static final IntermediateDataSetID DATA_SET_ID = new IntermediateDataSetID();
+ private static final ResultPartitionID PARTITION_ID_1 = new ResultPartitionID();
+ private static final ResultPartitionID PARTITION_ID_2 = new ResultPartitionID();
+
+ @Test
+ public void testProcessEmptyClusterPartitionReport() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(TASK_EXECUTOR_ID_1, EMPTY_PARTITION_REPORT);
+ assertThat(partitionReleaser.releaseCalls, empty());
+ }
+
+ /**
+ * Verifies that a task executor hosting multiple partitions of a data set receives a release call if a subset of
+ * its partitions is lost.
+ */
+ @Test
+ public void testReportProcessingWithPartitionLossOnSameTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_1, Collections.singleton(DATA_SET_ID))));
+ }
+
+ /**
+ * Verifies that a task executor hosting partitions of a data set receives a release call if a partition of the
+ * data set is lost on another task executor.
+ */
+ @Test
+ public void testReportProcessingWithPartitionLossOnOtherTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_2,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(TASK_EXECUTOR_ID_1, EMPTY_PARTITION_REPORT);
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_2, Collections.singleton(DATA_SET_ID))));
+ }
+
+ @Test
+ public void testListDataSetsBasics() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ assertEquals(0, tracker.listDataSets().size());
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(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);
+ assertEquals(1, metaInfo.getNumTotalPartitions());
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ EMPTY_PARTITION_REPORT);
+ assertEquals(0, tracker.listDataSets().size());
+ }
+
+ @Test
+ public void testListDataSetsMultiplePartitionsOnSingleTaskExecutor() {
Review comment:
Just an idea, if we init `partitionReleaser/tracker` in test setup and factor out
`tracker.processTaskExecutorClusterPartitionReport(TASK_EXECUTOR_ID_1, createClusterPartitionReport(..`
into e.g. `report(TASK_EXECUTOR_ID_1, DATA_SET_ID, 2, PARTITION_ID_1, ..)`
this would reduce code size and simplify writing tests with duplicated setup but testing one thing.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r398027534
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImpl.java
##########
@@ -0,0 +1,232 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+/**
+ * Default {@link ResourceManagerPartitionTracker} implementation.
+ *
+ * <p>Internal tracking info must only be updated upon reception of a {@link ClusterPartitionReport}, as the task
+ * executor state is the source of truth.
+ */
+public class ResourceManagerPartitionTrackerImpl implements ResourceManagerPartitionTracker {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ResourceManagerPartitionTrackerImpl.class);
+
+ private final Map<ResourceID, Set<IntermediateDataSetID>> taskExecutorToDataSets = new HashMap<>();
+ private final Map<IntermediateDataSetID, Map<ResourceID, Set<ResultPartitionID>>> dataSetToTaskExecutors = new HashMap<>();
+ private final Map<IntermediateDataSetID, DataSetMetaInfo> dataSetMetaInfo = new HashMap<>();
+ private final Map<IntermediateDataSetID, CompletableFuture<Void>> partitionReleaseCompletionFutures = new HashMap<>();
+
+ private final ClusterPartitionReleaser clusterPartitionReleaser;
+
+ public ResourceManagerPartitionTrackerImpl(ClusterPartitionReleaser clusterPartitionReleaser) {
+ this.clusterPartitionReleaser = clusterPartitionReleaser;
+ }
+
+ @Override
+ public void processTaskExecutorClusterPartitionReport(ResourceID taskExecutorId, ClusterPartitionReport clusterPartitionReport) {
+ Preconditions.checkNotNull(taskExecutorId);
+ Preconditions.checkNotNull(clusterPartitionReport);
+ LOG.debug("Processing cluster partition report from task executor {}: {}.", taskExecutorId, clusterPartitionReport);
+
+ internalProcessClusterPartitionReport(taskExecutorId, clusterPartitionReport);
+ }
+
+ @Override
+ public void processTaskExecutorShutdown(ResourceID taskExecutorId) {
+ Preconditions.checkNotNull(taskExecutorId);
+ LOG.debug("Processing shutdown of task executor {}.", taskExecutorId);
+
+ internalProcessClusterPartitionReport(taskExecutorId, new ClusterPartitionReport(Collections.emptyList()));
+ }
+
+ @Override
+ public CompletableFuture<Void> releaseClusterPartitions(IntermediateDataSetID dataSetId) {
+ Preconditions.checkNotNull(dataSetId);
+ LOG.debug("Releasing cluster partitions for data set {}.", dataSetId);
+
+ CompletableFuture<Void> partitionReleaseCompletionFuture = partitionReleaseCompletionFutures.computeIfAbsent(dataSetId, ignored -> new CompletableFuture<>());
+ internalReleasePartitions(Collections.singleton(dataSetId));
Review comment:
do we want to check whether `internalReleasePartitions` actually issued any releases?
to avoid dangling futures in `partitionReleaseCompletionFutures` if `dataSetId` is not actually tracked at all
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zentol commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r400029895
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTracker.java
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
Review comment:
All partition tracking code is currently in this package. I agree that we may want to move all them elsewhere at some point.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156483467",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6821",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "CANCELED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156536980",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "14b3f8a73814fa648afc42be527a565821a262a8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "14b3f8a73814fa648afc42be527a565821a262a8",
"triggerType" : "PUSH"
}, {
"hash" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"triggerType" : "PUSH"
}, {
"hash" : "8a86aa89a221085749362a508718bd4959ac8190",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8a86aa89a221085749362a508718bd4959ac8190",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3e149786a3b54c141ecd14b99d98062d31f56bb4 Travis: [CANCELED](https://travis-ci.com/github/flink-ci/flink/builds/156536980) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827)
* 14b3f8a73814fa648afc42be527a565821a262a8 UNKNOWN
* 38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72 UNKNOWN
* 8a86aa89a221085749362a508718bd4959ac8190 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-603773092
The commits are in wrong order:
```
0240c441a2 [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
c5d34a3124 [FLINK-14791][coordination] Constructor changes
```
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156483467",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6821",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156536980",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "14b3f8a73814fa648afc42be527a565821a262a8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "14b3f8a73814fa648afc42be527a565821a262a8",
"triggerType" : "PUSH"
}, {
"hash" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"triggerType" : "PUSH"
}, {
"hash" : "8a86aa89a221085749362a508718bd4959ac8190",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6830",
"triggerID" : "8a86aa89a221085749362a508718bd4959ac8190",
"triggerType" : "PUSH"
}, {
"hash" : "8a86aa89a221085749362a508718bd4959ac8190",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156550433",
"triggerID" : "8a86aa89a221085749362a508718bd4959ac8190",
"triggerType" : "PUSH"
}, {
"hash" : "998a1c6d2107addf7bfafd0114b2f696837134aa",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6981",
"triggerID" : "998a1c6d2107addf7bfafd0114b2f696837134aa",
"triggerType" : "PUSH"
}, {
"hash" : "998a1c6d2107addf7bfafd0114b2f696837134aa",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/157731336",
"triggerID" : "998a1c6d2107addf7bfafd0114b2f696837134aa",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 14b3f8a73814fa648afc42be527a565821a262a8 UNKNOWN
* 38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72 UNKNOWN
* 998a1c6d2107addf7bfafd0114b2f696837134aa Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/157731336) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6981)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r397865079
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ClusterPartitionReleaser.java
##########
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+
+import java.util.Set;
+
+/**
+ * Interface for releasing cluster partitions on a task executor.
+ */
+public interface ClusterPartitionReleaser {
Review comment:
```suggestion
public interface TaskExecutorClusterPartitionReleaser {
```
Just if we have also something to release external partitions over shuffle master/ClusterPartitionShuffleClient.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zentol commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r400034659
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImpl.java
##########
@@ -0,0 +1,232 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+/**
+ * Default {@link ResourceManagerPartitionTracker} implementation.
+ *
+ * <p>Internal tracking info must only be updated upon reception of a {@link ClusterPartitionReport}, as the task
+ * executor state is the source of truth.
+ */
+public class ResourceManagerPartitionTrackerImpl implements ResourceManagerPartitionTracker {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ResourceManagerPartitionTrackerImpl.class);
+
+ private final Map<ResourceID, Set<IntermediateDataSetID>> taskExecutorToDataSets = new HashMap<>();
+ private final Map<IntermediateDataSetID, Map<ResourceID, Set<ResultPartitionID>>> dataSetToTaskExecutors = new HashMap<>();
+ private final Map<IntermediateDataSetID, DataSetMetaInfo> dataSetMetaInfo = new HashMap<>();
+ private final Map<IntermediateDataSetID, CompletableFuture<Void>> partitionReleaseCompletionFutures = new HashMap<>();
+
+ private final ClusterPartitionReleaser clusterPartitionReleaser;
+
+ public ResourceManagerPartitionTrackerImpl(ClusterPartitionReleaser clusterPartitionReleaser) {
+ this.clusterPartitionReleaser = clusterPartitionReleaser;
+ }
+
+ @Override
+ public void processTaskExecutorClusterPartitionReport(ResourceID taskExecutorId, ClusterPartitionReport clusterPartitionReport) {
+ Preconditions.checkNotNull(taskExecutorId);
+ Preconditions.checkNotNull(clusterPartitionReport);
+ LOG.debug("Processing cluster partition report from task executor {}: {}.", taskExecutorId, clusterPartitionReport);
+
+ internalProcessClusterPartitionReport(taskExecutorId, clusterPartitionReport);
+ }
+
+ @Override
+ public void processTaskExecutorShutdown(ResourceID taskExecutorId) {
+ Preconditions.checkNotNull(taskExecutorId);
+ LOG.debug("Processing shutdown of task executor {}.", taskExecutorId);
+
+ internalProcessClusterPartitionReport(taskExecutorId, new ClusterPartitionReport(Collections.emptyList()));
+ }
+
+ @Override
+ public CompletableFuture<Void> releaseClusterPartitions(IntermediateDataSetID dataSetId) {
+ Preconditions.checkNotNull(dataSetId);
+ LOG.debug("Releasing cluster partitions for data set {}.", dataSetId);
+
+ CompletableFuture<Void> partitionReleaseCompletionFuture = partitionReleaseCompletionFutures.computeIfAbsent(dataSetId, ignored -> new CompletableFuture<>());
+ internalReleasePartitions(Collections.singleton(dataSetId));
+ return partitionReleaseCompletionFuture;
+ }
+
+ private void internalProcessClusterPartitionReport(ResourceID taskExecutorId, ClusterPartitionReport clusterPartitionReport) {
+ final Set<IntermediateDataSetID> dataSetsWithLostPartitions = clusterPartitionReport.getEntries().isEmpty()
+ ? processEmptyReport(taskExecutorId)
+ : setHostedDataSetsAndCheckCorruption(taskExecutorId, clusterPartitionReport.getEntries());
+
+ updateDataSetMetaData(clusterPartitionReport);
+
+ checkForFullyLostDatasets(dataSetsWithLostPartitions);
+
+ internalReleasePartitions(dataSetsWithLostPartitions);
+ }
+
+ private void internalReleasePartitions(Set<IntermediateDataSetID> dataSetsToRelease) {
+ Map<ResourceID, Set<IntermediateDataSetID>> releaseCalls = prepareReleaseCalls(dataSetsToRelease);
+ releaseCalls.forEach(clusterPartitionReleaser::releaseClusterPartitions);
+ }
+
+ private Set<IntermediateDataSetID> processEmptyReport(ResourceID taskExecutorId) {
+ Set<IntermediateDataSetID> previouslyHostedDatasets = taskExecutorToDataSets.remove(taskExecutorId);
+ if (previouslyHostedDatasets == null) {
+ // default path for task executors that never have any cluster partitions
+ previouslyHostedDatasets = Collections.emptySet();
+ } else {
+ previouslyHostedDatasets.forEach(dataSetId -> removeInnerKey(dataSetId, taskExecutorId, dataSetToTaskExecutors));
+ }
+ return previouslyHostedDatasets;
+ }
+
+ /**
+ * Updates the data sets for which the given task executor is hosting partitions and returns data sets that were
+ * corrupted due to a loss of partitions.
+ *
+ * @param taskExecutorId ID of the hosting TaskExecutor
+ * @param reportEntries IDs of data sets for which partitions are hosted
+ * @return corrupted data sets
+ */
+ private Set<IntermediateDataSetID> setHostedDataSetsAndCheckCorruption(ResourceID taskExecutorId, Collection<ClusterPartitionReport.ClusterPartitionReportEntry> reportEntries) {
+ final Set<IntermediateDataSetID> currentlyHostedDatasets = reportEntries
+ .stream()
+ .map(ClusterPartitionReport.ClusterPartitionReportEntry::getDataSetId)
+ .collect(Collectors.toSet());
+
+ final Set<IntermediateDataSetID> previouslyHostedDataSets = taskExecutorToDataSets.put(
+ taskExecutorId,
+ currentlyHostedDatasets);
+
+ // previously tracked data sets may be corrupted since we may be tracking less partitions than before
+ final Set<IntermediateDataSetID> potentiallyCorruptedDataSets = Optional
+ .ofNullable(previouslyHostedDataSets)
+ .orElse(new HashSet<>(0));
+
+ // update data set -> task executor mapping and find datasets for which lost a partition
+ reportEntries.forEach(hostedPartition -> {
+ final Map<ResourceID, Set<ResultPartitionID>> taskExecutorHosts = dataSetToTaskExecutors.computeIfAbsent(hostedPartition.getDataSetId(), ignored -> new HashMap<>());
+ final Set<ResultPartitionID> previouslyHostedPartitions = taskExecutorHosts.put(taskExecutorId, hostedPartition.getHostedPartitions());
+
+ final boolean noPartitionLost = previouslyHostedPartitions == null || hostedPartition.getHostedPartitions().containsAll(previouslyHostedPartitions);
+ if (noPartitionLost) {
+ potentiallyCorruptedDataSets.remove(hostedPartition.getDataSetId());
+ }
+ });
+
+ // now only contains data sets for which a partition is no longer tracked
+ return potentiallyCorruptedDataSets;
+ }
+
+ private void updateDataSetMetaData(ClusterPartitionReport clusterPartitionReport) {
+ // add meta info for new data sets
+ clusterPartitionReport.getEntries().forEach(entry ->
+ dataSetMetaInfo.compute(entry.getDataSetId(), (dataSetID, dataSetMetaInfo) -> {
+ if (dataSetMetaInfo == null) {
+ return new DataSetMetaInfo(entry.getNumTotalPartitions());
+ } else {
+ // double check that the meta data is consistent
+ Preconditions.checkState(dataSetMetaInfo.getNumTotalPartitions() == entry.getNumTotalPartitions());
+ return dataSetMetaInfo;
+ }
+ }));
+ }
+
+ private void checkForFullyLostDatasets(Set<IntermediateDataSetID> dataSetsWithLostPartitions) {
+ dataSetsWithLostPartitions.forEach(dataSetId -> {
+ if (getHostingTaskExecutors(dataSetId).isEmpty()) {
+ LOG.debug("There are no longer partitions being tracked for dataset {}.", dataSetId);
+ dataSetMetaInfo.remove(dataSetId);
+ Optional.ofNullable(partitionReleaseCompletionFutures.get(dataSetId)).map(future -> future.complete(null));
+ }
+ });
+ }
+
+ private Map<ResourceID, Set<IntermediateDataSetID>> prepareReleaseCalls(Set<IntermediateDataSetID> dataSetsToRelease) {
+ final Map<ResourceID, Set<IntermediateDataSetID>> releaseCalls = new HashMap<>();
+ dataSetsToRelease.forEach(dataSetToRelease -> {
+ final Set<ResourceID> hostingTaskExecutors = getHostingTaskExecutors(dataSetToRelease);
+ hostingTaskExecutors.forEach(hostingTaskExecutor -> insert(hostingTaskExecutor, dataSetToRelease, releaseCalls));
+ });
+ return releaseCalls;
+ }
+
+ private Set<ResourceID> getHostingTaskExecutors(IntermediateDataSetID dataSetId) {
+ Preconditions.checkNotNull(dataSetId);
+
+ Map<ResourceID, Set<ResultPartitionID>> trackedPartitions = dataSetToTaskExecutors.get(dataSetId);
+ if (trackedPartitions == null) {
+ return Collections.emptySet();
+ } else {
+ return trackedPartitions.keySet();
+ }
+ }
+
+ @Override
+ public Map<IntermediateDataSetID, DataSetMetaInfo> listDataSets() {
+ return dataSetMetaInfo.entrySet().stream()
+ .filter(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());
+
+ int numTrackedPartitions = 0;
+ for (Set<ResultPartitionID> hostedPartitions : taskExecutorToPartitions.values()) {
+ numTrackedPartitions += hostedPartitions.size();
+ }
Review comment:
I had the streams solution before but it was such an eye-sore that i opted for the simple loop approach instead.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156483467",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "df635bda88e745a98993533481f9a343603bf919",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6821",
"triggerID" : "df635bda88e745a98993533481f9a343603bf919",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6827",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156536980",
"triggerID" : "3e149786a3b54c141ecd14b99d98062d31f56bb4",
"triggerType" : "PUSH"
}, {
"hash" : "14b3f8a73814fa648afc42be527a565821a262a8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "14b3f8a73814fa648afc42be527a565821a262a8",
"triggerType" : "PUSH"
}, {
"hash" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72",
"triggerType" : "PUSH"
}, {
"hash" : "8a86aa89a221085749362a508718bd4959ac8190",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6830",
"triggerID" : "8a86aa89a221085749362a508718bd4959ac8190",
"triggerType" : "PUSH"
}, {
"hash" : "8a86aa89a221085749362a508718bd4959ac8190",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/156550433",
"triggerID" : "8a86aa89a221085749362a508718bd4959ac8190",
"triggerType" : "PUSH"
}, {
"hash" : "998a1c6d2107addf7bfafd0114b2f696837134aa",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6981",
"triggerID" : "998a1c6d2107addf7bfafd0114b2f696837134aa",
"triggerType" : "PUSH"
}, {
"hash" : "998a1c6d2107addf7bfafd0114b2f696837134aa",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/157731336",
"triggerID" : "998a1c6d2107addf7bfafd0114b2f696837134aa",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 14b3f8a73814fa648afc42be527a565821a262a8 UNKNOWN
* 38a8e1762445f4a2e521a0e2bc5459bb9dc3dc72 UNKNOWN
* 998a1c6d2107addf7bfafd0114b2f696837134aa Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/157731336) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6981)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zentol merged pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
zentol merged pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#issuecomment-600657884
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
}, {
"hash" : "0240c441a2ea285e49af3172c2e5829207be011b",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153925949",
"triggerID" : "0240c441a2ea285e49af3172c2e5829207be011b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0240c441a2ea285e49af3172c2e5829207be011b Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153925949) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6365)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zentol commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r400034100
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImpl.java
##########
@@ -0,0 +1,232 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+/**
+ * Default {@link ResourceManagerPartitionTracker} implementation.
+ *
+ * <p>Internal tracking info must only be updated upon reception of a {@link ClusterPartitionReport}, as the task
+ * executor state is the source of truth.
+ */
+public class ResourceManagerPartitionTrackerImpl implements ResourceManagerPartitionTracker {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ResourceManagerPartitionTrackerImpl.class);
+
+ private final Map<ResourceID, Set<IntermediateDataSetID>> taskExecutorToDataSets = new HashMap<>();
+ private final Map<IntermediateDataSetID, Map<ResourceID, Set<ResultPartitionID>>> dataSetToTaskExecutors = new HashMap<>();
+ private final Map<IntermediateDataSetID, DataSetMetaInfo> dataSetMetaInfo = new HashMap<>();
+ private final Map<IntermediateDataSetID, CompletableFuture<Void>> partitionReleaseCompletionFutures = new HashMap<>();
+
+ private final ClusterPartitionReleaser clusterPartitionReleaser;
+
+ public ResourceManagerPartitionTrackerImpl(ClusterPartitionReleaser clusterPartitionReleaser) {
+ this.clusterPartitionReleaser = clusterPartitionReleaser;
+ }
+
+ @Override
+ public void processTaskExecutorClusterPartitionReport(ResourceID taskExecutorId, ClusterPartitionReport clusterPartitionReport) {
+ Preconditions.checkNotNull(taskExecutorId);
+ Preconditions.checkNotNull(clusterPartitionReport);
+ LOG.debug("Processing cluster partition report from task executor {}: {}.", taskExecutorId, clusterPartitionReport);
+
+ internalProcessClusterPartitionReport(taskExecutorId, clusterPartitionReport);
+ }
+
+ @Override
+ public void processTaskExecutorShutdown(ResourceID taskExecutorId) {
+ Preconditions.checkNotNull(taskExecutorId);
+ LOG.debug("Processing shutdown of task executor {}.", taskExecutorId);
+
+ internalProcessClusterPartitionReport(taskExecutorId, new ClusterPartitionReport(Collections.emptyList()));
+ }
+
+ @Override
+ public CompletableFuture<Void> releaseClusterPartitions(IntermediateDataSetID dataSetId) {
+ Preconditions.checkNotNull(dataSetId);
+ LOG.debug("Releasing cluster partitions for data set {}.", dataSetId);
+
+ CompletableFuture<Void> partitionReleaseCompletionFuture = partitionReleaseCompletionFutures.computeIfAbsent(dataSetId, ignored -> new CompletableFuture<>());
+ internalReleasePartitions(Collections.singleton(dataSetId));
+ return partitionReleaseCompletionFuture;
+ }
+
+ private void internalProcessClusterPartitionReport(ResourceID taskExecutorId, ClusterPartitionReport clusterPartitionReport) {
+ final Set<IntermediateDataSetID> dataSetsWithLostPartitions = clusterPartitionReport.getEntries().isEmpty()
+ ? processEmptyReport(taskExecutorId)
+ : setHostedDataSetsAndCheckCorruption(taskExecutorId, clusterPartitionReport.getEntries());
+
+ updateDataSetMetaData(clusterPartitionReport);
+
+ checkForFullyLostDatasets(dataSetsWithLostPartitions);
+
+ internalReleasePartitions(dataSetsWithLostPartitions);
+ }
+
+ private void internalReleasePartitions(Set<IntermediateDataSetID> dataSetsToRelease) {
+ Map<ResourceID, Set<IntermediateDataSetID>> releaseCalls = prepareReleaseCalls(dataSetsToRelease);
+ releaseCalls.forEach(clusterPartitionReleaser::releaseClusterPartitions);
+ }
+
+ private Set<IntermediateDataSetID> processEmptyReport(ResourceID taskExecutorId) {
+ Set<IntermediateDataSetID> previouslyHostedDatasets = taskExecutorToDataSets.remove(taskExecutorId);
+ if (previouslyHostedDatasets == null) {
+ // default path for task executors that never have any cluster partitions
+ previouslyHostedDatasets = Collections.emptySet();
+ } else {
+ previouslyHostedDatasets.forEach(dataSetId -> removeInnerKey(dataSetId, taskExecutorId, dataSetToTaskExecutors));
+ }
+ return previouslyHostedDatasets;
+ }
+
+ /**
+ * Updates the data sets for which the given task executor is hosting partitions and returns data sets that were
+ * corrupted due to a loss of partitions.
+ *
+ * @param taskExecutorId ID of the hosting TaskExecutor
+ * @param reportEntries IDs of data sets for which partitions are hosted
+ * @return corrupted data sets
+ */
+ private Set<IntermediateDataSetID> setHostedDataSetsAndCheckCorruption(ResourceID taskExecutorId, Collection<ClusterPartitionReport.ClusterPartitionReportEntry> reportEntries) {
+ final Set<IntermediateDataSetID> currentlyHostedDatasets = reportEntries
+ .stream()
+ .map(ClusterPartitionReport.ClusterPartitionReportEntry::getDataSetId)
+ .collect(Collectors.toSet());
+
+ final Set<IntermediateDataSetID> previouslyHostedDataSets = taskExecutorToDataSets.put(
+ taskExecutorId,
+ currentlyHostedDatasets);
+
+ // previously tracked data sets may be corrupted since we may be tracking less partitions than before
+ final Set<IntermediateDataSetID> potentiallyCorruptedDataSets = Optional
+ .ofNullable(previouslyHostedDataSets)
+ .orElse(new HashSet<>(0));
+
+ // update data set -> task executor mapping and find datasets for which lost a partition
+ reportEntries.forEach(hostedPartition -> {
+ final Map<ResourceID, Set<ResultPartitionID>> taskExecutorHosts = dataSetToTaskExecutors.computeIfAbsent(hostedPartition.getDataSetId(), ignored -> new HashMap<>());
+ final Set<ResultPartitionID> previouslyHostedPartitions = taskExecutorHosts.put(taskExecutorId, hostedPartition.getHostedPartitions());
+
+ final boolean noPartitionLost = previouslyHostedPartitions == null || hostedPartition.getHostedPartitions().containsAll(previouslyHostedPartitions);
+ if (noPartitionLost) {
+ potentiallyCorruptedDataSets.remove(hostedPartition.getDataSetId());
+ }
+ });
+
+ // now only contains data sets for which a partition is no longer tracked
+ return potentiallyCorruptedDataSets;
+ }
+
+ private void updateDataSetMetaData(ClusterPartitionReport clusterPartitionReport) {
+ // add meta info for new data sets
+ clusterPartitionReport.getEntries().forEach(entry ->
+ dataSetMetaInfo.compute(entry.getDataSetId(), (dataSetID, dataSetMetaInfo) -> {
+ if (dataSetMetaInfo == null) {
+ return new DataSetMetaInfo(entry.getNumTotalPartitions());
+ } else {
+ // double check that the meta data is consistent
+ Preconditions.checkState(dataSetMetaInfo.getNumTotalPartitions() == entry.getNumTotalPartitions());
+ return dataSetMetaInfo;
+ }
+ }));
+ }
+
+ private void checkForFullyLostDatasets(Set<IntermediateDataSetID> dataSetsWithLostPartitions) {
+ dataSetsWithLostPartitions.forEach(dataSetId -> {
+ if (getHostingTaskExecutors(dataSetId).isEmpty()) {
+ LOG.debug("There are no longer partitions being tracked for dataset {}.", dataSetId);
+ dataSetMetaInfo.remove(dataSetId);
+ Optional.ofNullable(partitionReleaseCompletionFutures.get(dataSetId)).map(future -> future.complete(null));
Review comment:
yes, will change it to remove() the future instead of get().
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on a change in pull request #11443:
[FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
Posted by GitBox <gi...@apache.org>.
azagrebin commented on a change in pull request #11443: [FLINK-14791][coordination] ResourceManager tracks ClusterPartitions
URL: https://github.com/apache/flink/pull/11443#discussion_r398371898
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResourceManagerPartitionTrackerImplTest.java
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.collection.IsEmptyCollection.empty;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for the {@link ResourceManagerPartitionTrackerImpl}.
+ */
+public class ResourceManagerPartitionTrackerImplTest extends TestLogger {
+
+ private static final ClusterPartitionReport EMPTY_PARTITION_REPORT = new ClusterPartitionReport(Collections.emptySet());
+
+ private static final ResourceID TASK_EXECUTOR_ID_1 = ResourceID.generate();
+ private static final ResourceID TASK_EXECUTOR_ID_2 = ResourceID.generate();
+ private static final IntermediateDataSetID DATA_SET_ID = new IntermediateDataSetID();
+ private static final ResultPartitionID PARTITION_ID_1 = new ResultPartitionID();
+ private static final ResultPartitionID PARTITION_ID_2 = new ResultPartitionID();
+
+ @Test
+ public void testProcessEmptyClusterPartitionReport() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(TASK_EXECUTOR_ID_1, EMPTY_PARTITION_REPORT);
+ assertThat(partitionReleaser.releaseCalls, empty());
+ }
+
+ /**
+ * Verifies that a task executor hosting multiple partitions of a data set receives a release call if a subset of
+ * its partitions is lost.
+ */
+ @Test
+ public void testReportProcessingWithPartitionLossOnSameTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_1, Collections.singleton(DATA_SET_ID))));
+ }
+
+ /**
+ * Verifies that a task executor hosting partitions of a data set receives a release call if a partition of the
+ * data set is lost on another task executor.
+ */
+ @Test
+ public void testReportProcessingWithPartitionLossOnOtherTaskExecutor() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_1,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_1));
+
+ tracker.processTaskExecutorClusterPartitionReport(
+ TASK_EXECUTOR_ID_2,
+ createClusterPartitionReport(DATA_SET_ID, 2, PARTITION_ID_2));
+
+ tracker.processTaskExecutorClusterPartitionReport(TASK_EXECUTOR_ID_1, EMPTY_PARTITION_REPORT);
+
+ assertThat(partitionReleaser.releaseCalls, contains(Tuple2.of(TASK_EXECUTOR_ID_2, Collections.singleton(DATA_SET_ID))));
+ }
+
+ @Test
+ public void testListDataSetsBasics() {
+ TestClusterPartitionReleaser partitionReleaser = new TestClusterPartitionReleaser();
+ final ResourceManagerPartitionTracker tracker = new ResourceManagerPartitionTrackerImpl(partitionReleaser);
+
+ assertEquals(0, tracker.listDataSets().size());
Review comment:
```suggestion
assertThat(tracker.listDataSets().size(), is(0));
```
minor: looks we usually do it like this for readability
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services