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 2021/07/13 00:23:28 UTC
[GitHub] [flink] zhuzhurk commented on a change in pull request #16436: [FLINK-22017][coordination] Allow BLOCKING result partition to be individually consumable
zhuzhurk commented on a change in pull request #16436:
URL: https://github.com/apache/flink/pull/16436#discussion_r668312184
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/IntermediateResultPartition.java
##########
@@ -72,6 +75,13 @@ public ResultPartitionType getResultType() {
return getEdgeManager().getConsumerVertexGroupsForPartition(partitionId);
}
+ public List<ConsumedPartitionGroup> getConsumedPartitionGroups() {
+ if (consumedPartitionGroups == null) {
+ consumedPartitionGroups = getEdgeManager().getConsumedPartitionGroupsById(partitionId);
Review comment:
I think we do not need this field `consumedPartitionGroups`.
Using it without initialization may even cause problems.
In the next commit I can see that `getConsumedPartitionGroups()` is used instead of `consumedPartitionGroups`.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/EdgeManagerBuildUtil.java
##########
@@ -88,6 +88,7 @@ private static void connectAllToAll(
Arrays.stream(intermediateResult.getPartitions())
.map(IntermediateResultPartition::getPartitionId)
.collect(Collectors.toList()));
+ registerConsumedPartitionGroupToEdgeManager(intermediateResult, consumedPartitions);
Review comment:
maybe add a method `createAndRegisterConsumedPartitionGroup(IntermediateResultPartition... partitions)`, in case some `ConsumedPartitionGroup` is created but not registered?
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphConstructionTest.java
##########
@@ -485,4 +491,39 @@ public void testMoreThanOneConsumerForIntermediateResult() {
fail(e.getMessage());
}
}
+
+ @Test
+ public void testRegisterConsumedPartitionGroupToEdgeManager() throws Exception {
+ JobVertex v1 = new JobVertex("source");
+ JobVertex v2 = new JobVertex("sink");
+
+ v1.setParallelism(2);
+ v2.setParallelism(2);
+
+ v2.connectNewDataSetAsInput(
+ v1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.BLOCKING);
+
+ List<JobVertex> ordered = new ArrayList<>(Arrays.asList(v1, v2));
+ ExecutionGraph eg = createDefaultExecutionGraph(ordered);
+ eg.attachJobGraph(ordered);
+
+ IntermediateResult result =
+ Objects.requireNonNull(eg.getJobVertex(v1.getID())).getProducedDataSets()[0];
+
+ IntermediateResultPartition partition1 = result.getPartitions()[0];
+ IntermediateResultPartition partition2 = result.getPartitions()[1];
+
+ assertEquals(
+ partition1.getConsumedPartitionGroups().get(0),
+ partition2.getConsumedPartitionGroups().get(0));
+
+ ConsumedPartitionGroup consumedPartitionGroup =
+ partition1.getConsumedPartitionGroups().get(0);
+ Set<IntermediateResultPartitionID> partitionIds = new HashSet<>();
+ for (IntermediateResultPartitionID partitionId : consumedPartitionGroup) {
+ partitionIds.add(partitionId);
+ }
+ assertTrue(partitionIds.contains(partition1.getPartitionId()));
Review comment:
can be `assertThat(partitionIds, containsInAnyOrder(partition1.getPartitionId(), partition2.getPartitionId()))`
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/EdgeManager.java
##########
@@ -89,4 +92,17 @@ public void connectVertexWithConsumedPartitionGroup(
return Collections.unmodifiableList(
getConsumedPartitionGroupsForVertexInternal(executionVertexId));
}
+
+ public void registerConsumedPartitionGroup(ConsumedPartitionGroup group) {
+ for (IntermediateResultPartitionID partitionId : group) {
+ consumedPartitionsById
+ .computeIfAbsent(partitionId, ignore -> new ArrayList<>())
+ .add(group);
+ }
+ }
+
+ public List<ConsumedPartitionGroup> getConsumedPartitionGroupsById(
+ IntermediateResultPartitionID id) {
+ return Collections.unmodifiableList(consumedPartitionsById.get(id));
Review comment:
maybe `consumedPartitionsById.get(id)` -> `consumedPartitionsById.computeIfAbsent(id, id -> new ArrayList<>())`?
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adapter/DefaultResultPartitionTest.java
##########
@@ -34,30 +46,71 @@
/** Unit tests for {@link DefaultResultPartition}. */
public class DefaultResultPartitionTest extends TestLogger {
- private static final TestResultPartitionStateSupplier resultPartitionState =
- new TestResultPartitionStateSupplier();
-
- private final IntermediateResultPartitionID resultPartitionId =
- new IntermediateResultPartitionID();
- private final IntermediateDataSetID intermediateResultId = new IntermediateDataSetID();
+ @Test
+ public void testGetPartitionState() {
+ final TestResultPartitionStateSupplier resultPartitionState =
+ new TestResultPartitionStateSupplier();
- private DefaultResultPartition resultPartition;
+ final IntermediateResultPartitionID resultPartitionId = new IntermediateResultPartitionID();
+ final IntermediateDataSetID intermediateResultId = new IntermediateDataSetID();
- @Before
- public void setUp() {
- resultPartition =
+ final DefaultResultPartition resultPartition =
new DefaultResultPartition(
resultPartitionId, intermediateResultId, BLOCKING, resultPartitionState);
- }
- @Test
- public void testGetPartitionState() {
for (ResultPartitionState state : ResultPartitionState.values()) {
resultPartitionState.setResultPartitionState(state);
assertEquals(state, resultPartition.getState());
}
}
+ @Test
+ public void testGetConsumedPartitionGroup() throws Exception {
Review comment:
I think this case is better to be placed in `EdgeManagerTest`.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/ConsumedPartitionGroup.java
##########
@@ -58,4 +64,21 @@ public boolean isEmpty() {
public IntermediateResultPartitionID getFirst() {
return iterator().next();
}
+
+ public int partitionUnfinished() {
+ return unfinishedPartitions.incrementAndGet();
+ }
+
+ public int partitionFinished() {
+ return unfinishedPartitions.decrementAndGet();
+ }
+
+ @VisibleForTesting
+ public int getUnfinishedPartition() {
Review comment:
`getUnfinishedPartition` -> `getNumberOfUnfinishedPartitions`
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/PipelinedRegionSchedulingStrategy.java
##########
@@ -201,13 +182,11 @@ public void onExecutionStateChange(
.filter(
partition ->
partition.getState() == ResultPartitionState.CONSUMABLE)
- .flatMap(
- partition ->
- correlatedResultPartitionGroups
- .getOrDefault(
- partition.getResultId(),
- Collections.emptySet())
- .stream())
+ .flatMap(partition -> partition.getConsumedPartitionGroups().stream())
+ .filter(
+ group ->
+ crossRegionConsumedPartitionGroups.contains(group)
+ || group.isConsumable())
.collect(Collectors.toSet());
// for POINTWISE consumers of a BLOCKING partition, it's possible that some of the
Review comment:
The following filter `areRegionVerticesAllInCreatedState` can be removed now
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/TestingSchedulingExecutionVertex.java
##########
@@ -94,6 +94,9 @@ void addConsumedPartition(TestingSchedulingResultPartition consumedPartition) {
ConsumedPartitionGroup.fromSinglePartition(consumedPartition.getId());
consumedPartition.registerConsumedPartitionGroup(consumedPartitionGroup);
+ if (consumedPartition.getState() == ResultPartitionState.CONSUMABLE) {
Review comment:
can we put this condition block into `consumedPartition.registerConsumedPartitionGroup`?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/IntermediateResultPartition.java
##########
@@ -120,17 +118,8 @@ boolean markFinished() {
hasDataProduced = true;
Review comment:
it's better to check that this partition was not finished yet.
this can expose problems to ease trouble shooting.
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/PipelinedRegionSchedulingStrategyTest.java
##########
@@ -174,10 +174,12 @@ public void testFinishedBlockingResultPartitionProducerDoNotScheduleNonCreatedRe
final PipelinedRegionSchedulingStrategy schedulingStrategy = startScheduling(topology);
+ // non-CREATED regions should not be re-scheduled
consumer.get(0).setState(ExecutionState.SCHEDULED);
Review comment:
now we do not expect this case to happen.
exception should be thrown in this case.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/ConsumedPartitionGroup.java
##########
@@ -58,4 +64,21 @@ public boolean isEmpty() {
public IntermediateResultPartitionID getFirst() {
return iterator().next();
}
+
+ public int partitionUnfinished() {
+ return unfinishedPartitions.incrementAndGet();
+ }
+
+ public int partitionFinished() {
+ return unfinishedPartitions.decrementAndGet();
+ }
+
+ @VisibleForTesting
+ public int getUnfinishedPartition() {
+ return unfinishedPartitions.get();
+ }
+
+ public boolean isConsumable() {
Review comment:
This is not right for pipelined partitions.
So I think a name `areAllPartitionsFinished()` is more accurate.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org