[FLINK-12883][WIP][runtime] Add elaborated partition release logic #8804

GJL · 2019-06-20T09:34:31Z

What is the purpose of the change

WIP

Brief change log

See commits

Verifying this change

This change added tests and can be verified as follows:

Added unit tests, such as RegionPartitionReleaseStrategyTest

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2019-06-20T09:36:28Z

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.

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 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

Bot commands

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

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java

...rc/test/java/org/apache/flink/runtime/executiongraph/RegionPartitionReleaseStrategyTest.java

...runtime/executiongraph/failover/flip1/partitionrelease/PipelinedRegionExecutionViewTest.java

zentol

I think we're on the right track.

zentol · 2019-06-20T10:59:47Z

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java

 			// by filtering out late failure calls, we can save some work in
 			// avoiding redundant local failover
 			if (execution.getGlobalModVersion() == globalModVersion) {
+				partitionReleaseStrategy.vertexUnfinished(execution.getVertex().getID());


I don't believe this to be necessary. This branch is only executed if the new state is FAILED, but this operation here only has any effect if the execution in question previously arrived in a FINISHED state. Since we don't allow state transitions from FINISHED to FAILED; this operation should always be a no-op.

I think you are right.

zentol · 2019-06-20T11:01:59Z

...e/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategy.java

+
+	/**
+	 * Calling this method informs the strategy that a vertex is no longer in finished state, e.g.,
+	 * when an execution is restarted.


the use of "execution" makes this javadoc a bit ambiguous, as it could either mean that we re-execute a vertex and restart an Execution.

I can change it to "[...]e.g., when a vertex is re-executed"

zentol · 2019-06-20T11:07:25Z

...a/org/apache/flink/runtime/executiongraph/failover/flip1/RestartPipelinedRegionStrategy.java

-		for (HashSet<FailoverVertex> regionVertices : vertexToRegion.values()) {
-			distinctRegions.put(regionVertices, null);
-		}
+		final Set<Set<FailoverVertex>> distinctRegions = PipelinedRegionComputeUtil.computePipelinedRegions(topology);


we should modify the constructor to accept the pipelined regions instead; similarly to the RegionPartitionReleaseStrategy

That's not so obvious for me. RestartPipelinedRegionStrategy still needs FailoverVertex at the end of the day. Mapping from PipelinedRegion to Set<FailoverVertex> is tedious.

To me It's also not clear whether it is the right abstraction that PipelinedRegionComputeUtil#computePipelinedRegions() receives FailoverTopology as input.

Lastly, I think that this comment competes with that one.

zentol · 2019-06-20T11:18:24Z

...k/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java

+		return new PipelinedRegionConsumedBlockingPartitions(pipelinedRegion, resultPartitionsOutsideOfRegion);
+	}
+
+	private Set<IntermediateResultPartitionID> findResultPartitionsOutsideOfRegion(final PipelinedRegion pipelinedRegion) {


findResultPartitionsInsideOfRegion; could be static

This method calls .map(schedulingTopology::getVertexOrThrow) so it cannot be static.

zentol · 2019-06-20T11:21:45Z

...k/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java

+		return filterResultPartitionsOutsideOfRegion(allConsumedPartitionsInRegion, pipelinedRegion);
+	}
+
+	private Set<IntermediateResultPartitionID> filterResultPartitionsOutsideOfRegion(


could be static

zentol · 2019-06-20T11:23:37Z

...k/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java

+		}
+	}
+
+	private void initRegionExecutionViewByVertex(final Set<PipelinedRegion> pipelinedRegions) {


I'd modify these to return a collection instead and make them static.

zentol · 2019-06-20T11:50:15Z

...k/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java

+		this.schedulingTopology = checkNotNull(schedulingTopology);
+
+		checkNotNull(pipelinedRegions);
+		initConsumedBlockingPartitionsByRegion(pipelinedRegions);


i'm wondering whether we shouldn't use the Failover classes here, purely because similar code (restart strategy, pipelined region computation) also works against that. I'm just worried that we end up re-inventing the wheel; everything we do here the RestartPipelinedRegionStrategy also has to do at some point.

AN example is determining the set of input vertices. If you'd rely on FailoverRegion instead of your new PipelinedRegion, you could extend the FailoverRegion with the below method (just maybe without streams), allowing us to share the logic between the release and failover strategies.

public Set<FailoverVertex> getInputVertices() { return getAllExecutionVertices().stream() .map(FailoverVertex::getInputEdges) .flatMap(edges -> StreamSupport.stream(edges.spliterator(), false)) .map(FailoverEdge::getSourceVertex) .filter(vertex -> !executionVertices.contains(vertex)) .collect(Collectors.toSet()); }

zentol · 2019-06-20T12:10:49Z

...rc/test/java/org/apache/flink/runtime/executiongraph/RegionPartitionReleaseStrategyTest.java

+		final RegionPartitionReleaseStrategy regionPartitionReleaseStrategy = new RegionPartitionReleaseStrategy(testingSchedulingTopology, pipelinedRegions);
+
+		regionPartitionReleaseStrategy.vertexFinished(onlyConsumerVertexId);
+		regionPartitionReleaseStrategy.vertexUnfinished(onlyConsumerVertexId);


This test would pass even if vertexUnfinished is a no-op.

Also, shouldn't we at some point mark the producer vertex as finished? A scenario where a consumer is finished without the producer being finished sounds a bit like undefined behavior.

This test would pass even if vertexUnfinished is a no-op.

Good catch.

Also, shouldn't we at some point mark the producer vertex as finished? A scenario where a consumer is finished without the producer being finished sounds a bit like undefined behavior.

We can, but:

calls to ExecutionGraph#updateState(TaskExecutionState state) can overtake each other. Therefore it is to be expected that a consumer can be finished (in our bookkeeping) before the producer.

I don't think it adds to the readability of the test

zentol · 2019-06-20T12:13:54Z

...ink/runtime/executiongraph/failover/flip1/partitionrelease/PipelinedRegionExecutionView.java

+ */
+class PipelinedRegionExecutionView {
+
+	private final PipelinedRegion pipelinedRegion;


Do we only have this field to ensure we don't unfinish add arbitrary vertices? You could have the same behavior by maintaining a Map<EVID, Boolean>. You would then only pass in a set of vertices through the constructor.

There is a getter for the pipelinedRegion which is actually used. Also, I don't see anything wrong with object composition here. Using Map would require additional processing of the input.

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java

GJL · 2019-06-20T14:04:22Z

.../java/org/apache/flink/runtime/executiongraph/failover/flip1/PipelinedRegionComputeUtil.java

+
+	private static Set<Set<FailoverVertex>> uniqueRegions(final Map<FailoverVertex, Set<FailoverVertex>> vertexToRegion) {
+		// find out all the distinct regions
+		final IdentityHashMap<Set<FailoverVertex>, Object> distinctRegions = new IdentityHashMap<>();


Consider Collections.newSetFromMap(new IdentityHashMap<>())

GJL · 2019-06-21T06:59:43Z

...ink/runtime/executiongraph/failover/flip1/partitionrelease/PipelinedRegionExecutionView.java

+ * Provides a virtual execution state of a {@link PipelinedRegion}.
+ *
+ * <p>A pipelined region can be either finished or unfinished. It is finished iff. all its
+ * execution have reached the finished state.


executions

GJL · 2019-06-21T08:54:13Z

Rebased to master

zentol · 2019-06-21T11:41:40Z

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java


 		if (attempt != null) {
 			try {
+				maybeReleasePartitions(attempt, state);


this must be done after the attempt is marked as finished. If the finishing fails we might still need these partitions.

zentol

The current implementation may leak partitions if a vertex that produces a blocking partition was finished and is being reset.
The reliance on the current execution attempt may make it difficult to handle this case, as this imposes a temporal constraint: the release strategy must be informed about this event, and handle it, before the vertex is reset and a new Execution is created.

GJL · 2019-06-24T08:06:06Z

@zentol If a blocking partition is finished and its producer being reset, are there cases where we do not want to release that partition? I wonder if the release strategy should be responsible to release partitions in cases of errors.

zentol · 2019-06-24T09:13:51Z

If a blocking partition is finished and its producer being reset, are there cases where we do not want to release that partition?

I don't think so.

I wonder if the release strategy should be responsible to release partitions in cases of errors.

I'm not sure either. We'd have to handle it in the ExecutionVertex otherwise, which would be unfortunate as it adds yet another place from where release calls can originate.

zhijiangW · 2019-06-24T09:38:16Z

...e/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategy.java

+	/**
+	 * Calling this method informs the strategy that a vertex is no longer in finished state, e.g.,
+	 * when a vertex is re-executed.
+	 * @param executionVertexID Id of the vertex that is no longer in finished state.


add an empty line before @param to keep consistency with above vertexFinished

zhijiangW · 2019-06-24T11:06:11Z

...e/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategy.java

+	 * Calling this method informs the strategy that a vertex finished.
+	 *
+	 * @param finishedVertex Id of the vertex that finished the execution
+	 * @return A list of result partition that can be released


partition->partitions

good catch, fixed

zhijiangW · 2019-06-24T11:12:45Z

...ink/runtime/executiongraph/failover/flip1/partitionrelease/PipelinedRegionExecutionView.java

+		unfinishedVertices.remove(executionVertexId);
+	}
+
+	public void unfinish(final ExecutionVertexID executionVertexId) {


could be package private

zhijiangW · 2019-06-24T11:12:56Z

...ink/runtime/executiongraph/failover/flip1/partitionrelease/PipelinedRegionExecutionView.java

+		unfinishedVertices.add(executionVertexId);
+	}
+
+	public PipelinedRegion getPipelinedRegion() {


ditto: package private

zhijiangW · 2019-06-24T11:13:52Z

...ink/runtime/executiongraph/failover/flip1/partitionrelease/PipelinedRegionExecutionView.java

+ * <p>A pipelined region can be either finished or unfinished. It is finished iff. all its
+ * executions have reached the finished state.
+ */
+class PipelinedRegionExecutionView {


PipelinedRegionExecutionView -> PipelinedRegionStateView? because we want to expose the region state here.

PipelinedRegionExecutionStateView would be most exact. I'll think about it.

zhijiangW · 2019-06-24T11:16:36Z

...ink/runtime/executiongraph/failover/flip1/partitionrelease/PipelinedRegionExecutionView.java

+		return unfinishedVertices.isEmpty();
+	}
+
+	public void finish(final ExecutionVertexID executionVertexId) {


better to call vertexFinished as PartitionReleaseStrategy#vertexFinished, also for the following vertexUnfinished

good point, changed it

zhijiangW · 2019-06-25T04:21:45Z

...org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PipelinedRegion.java

+		return new PipelinedRegion(new HashSet<>(Arrays.asList(executionVertexIds)));
+	}
+
+	public Set<ExecutionVertexID> getExecutionVertexIds() {


could be package private

zhijiangW · 2019-06-25T04:23:53Z

...org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PipelinedRegion.java

+	}
+
+	public static PipelinedRegion from(final Set<ExecutionVertexID> executionVertexIds) {
+		return new PipelinedRegion(executionVertexIds);


checkNotNull(executionVertexIds)

check added

zhijiangW · 2019-06-25T04:24:09Z

...org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/PipelinedRegion.java

+	}
+
+	public static PipelinedRegion from(final ExecutionVertexID... executionVertexIds) {
+		return new PipelinedRegion(new HashSet<>(Arrays.asList(executionVertexIds)));


ditto: checkNotNull(executionVertexIds)

I think here it is ok to omit checkNotNull.

PipelinedRegion.from(null) is an ambiguous call, and does not compile currently

PipelinedRegion.from(null, null) cannot be caught by checkNotNull.

Arrays.asList() already eagerly checks for a null argument. From the stacktrace it should be obvious where the NPE originates from.

zhijiangW · 2019-06-25T04:27:00Z

...xecutiongraph/failover/flip1/partitionrelease/PipelinedRegionConsumedBlockingPartitions.java

+		this.consumedBlockingPartitions = checkNotNull(consumedBlockingPartitions);
+	}
+
+	public Set<IntermediateResultPartitionID> getConsumedBlockingPartitions() {


package private method

The class already has default visibilty. We could reduce visibility for all it's methods but that makes harder to move the class to a different package if it is needed in the future. Moreover, if a class can be extended,public makes it clear what the interface of this class is:

https://stackoverflow.com/questions/5260467/public-methods-in-package-private-classes

zhijiangW · 2019-06-25T04:27:10Z

...xecutiongraph/failover/flip1/partitionrelease/PipelinedRegionConsumedBlockingPartitions.java

+		return consumedBlockingPartitions;
+	}
+
+	public PipelinedRegion getPipelinedRegion() {


ditto: package private

zhijiangW · 2019-06-25T06:10:21Z

...k/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java

+
+		checkNotNull(pipelinedRegions);
+		initConsumedBlockingPartitionsByRegion(pipelinedRegions);
+		initRegionExecutionViewByVertex(pipelinedRegions);


Could loop pipelinedRegions only once for both inits.

True but I think a method should only do one thing. For performance reasons only, it's not worth it imo.

We could still have two individual methods via adjusting the parameter as PipelinedRegion instead of Set in initConsumedBlockingPartitionsByRegion, and make the loop in constructor.
I believe it has no performance issue here, so you could keep the way as now. :)

zhijiangW · 2019-06-25T06:13:58Z

...k/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java

+
+	private final Map<PipelinedRegion, PipelinedRegionConsumedBlockingPartitions> consumedBlockingPartitionsByRegion = new IdentityHashMap<>();
+
+	private final Map<ExecutionVertexID, PipelinedRegionExecutionView> regionExecutionViewByVertex = new HashMap<>();


might give an initial size for above maps

Size needs to be numberOfExecutionVertexIds / loadFactor to avoid rehashing. I don't think it's worth it here.

zhijiangW

Thanks for opening this PR and I like the overall implementations @GJL !

I have not finished the whole review and left some minor comments atm. I would continue on it later.

gaoyunhaii · 2019-06-25T07:30:15Z

...e/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategy.java

+public interface PartitionReleaseStrategy {
+
+	/**
+	 * Calling this method informs the strategy that a vertex finished.


I think that -> when may be better.

Current Javadoc is comprehensible imo.

gaoyunhaii · 2019-06-25T07:30:53Z

...e/flink/runtime/executiongraph/failover/flip1/partitionrelease/PartitionReleaseStrategy.java

+	List<IntermediateResultPartitionID> vertexFinished(ExecutionVertexID finishedVertex);
+
+	/**
+	 * Calling this method informs the strategy that a vertex is no longer in finished state, e.g.,


Similarly, I think that -> when may be better.

zhijiangW · 2019-06-25T08:15:12Z

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java

+		// TODO: finish implementation
+	}
+
+	private ResultPartitionID createResultPartitionId(final IntermediateResultPartitionID resultPartition) {


add Id suffix for the parameter naming

zhijiangW · 2019-06-25T08:48:46Z

...runtime/executiongraph/failover/flip1/partitionrelease/PipelinedRegionExecutionViewTest.java

+		final PipelinedRegionExecutionView pipelinedRegionExecutionView = new PipelinedRegionExecutionView(pipelinedRegion);
+
+		pipelinedRegionExecutionView.vertexFinished(TEST_EXECUTION_VERTEX_ID);
+		pipelinedRegionExecutionView.vertexUnfinished(TEST_EXECUTION_VERTEX_ID);


Only confirm does it exist in real process that one finished vertex would be unfinished again?
If not, I wonder whether we need some addition checks in detail implementations of vertexUnfinished/vertexFinished.

yes, it can happen in production that a finished vertex is reset. The simplest example is a global failover where some vertex was already finished.

There are also other cases connected to the new failover strategy, which may reset finished vertices if an upstream partition is required for a failover but has disappeared. Then this partitions producer is restarted, along with all downstream consumers, finished or not.

zhijiangW · 2019-06-25T09:34:35Z

...rc/test/java/org/apache/flink/runtime/executiongraph/RegionPartitionReleaseStrategyTest.java

+		assertThat(partitionsToRelease, is(empty()));
+	}
+
+	private static Set<PipelinedRegion> pipelinedRegionsSet(final PipelinedRegion... pipelinedRegions) {


might be further deduplication via
pipelinedRegionsSet(ExecutionVertexID producerIds, ExecutionVertexID... consumerIds)

I prefer the current abstraction.

In your signature it is unclear where the pipelined region boundaries are:

pipelinedRegionsSet(ExecutionVertexID producerIds, ExecutionVertexID... consumerIds)

pipelinedRegionsSet(onlyProducerVertexId, consumerVertex1, consumerVertex2);

zhijiangW · 2019-06-25T09:35:20Z

I have finished the whole review @GJL !

gaoyunhaii · 2019-06-26T08:53:04Z

.../java/org/apache/flink/runtime/executiongraph/failover/flip1/PipelinedRegionComputeUtil.java

+		}
+
+		// we use the map (list -> null) to imitate an IdentityHashSet (which does not exist)
+		// this helps to optimize the building performance as it uses reference equality


This comment seems to be outdated since the result is now simply saved in a set backend by a IdentityHashMap.

good catch, fixed

GJL · 2019-06-27T15:16:42Z

rebased to master

GJL · 2019-06-28T07:15:32Z

Squashed commits

zentol

As a reminder, once #8928 is merged we have to extend it in this PR to also inform the release strategy of reset vertices, otherwise the scenario below could result in wrong behaviors:

Assume a failover region consisting of 3 operators (o1,o2,o3), where o1 consumes some blocking partition, o2/o3 consume o1 (pipelined) and each have a blocking output.

The job executes; o1 and o2 are finished, o3 fails.

At this point, the strategy knows that o1 and o2 is finished, and that o3 is not finished.

The job is restarted, and the entire region is re-deployed. The strategy is not informed any of these state changes, and still believes o1 and o2 to be finished.

On this run, the deployment of o2 takes a long at time, so much that o1 and o3 are finishing before o2 was ever deployed (and thus, no external state update has been received).

At this point the strategy would release the partition that o1 is consuming, because it knows that o1 and o3 just finished, and o2 never was unfinished.

Replace instances where we use an IdentityHashMap as a set with Collections.newSetFromMap() in RestartPipelinedRegionStrategy and PipelineRegionComputeUtil.

…Util#uniqueRegions()

zentol

there's an edge case that wasn't considered

zentol · 2019-07-01T12:15:59Z

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java

-						return false;
-				}
+				final boolean stateUpdated = updateStateInternal(state, attempt);
+				maybeReleasePartitions(state, attempt);


We should either take stateUpdated into account here, or check the actual execution state opposed to what is in the task state update.

If you have a vertex in a CANCELING state, which then receives a FINISHED state updated, the vertex completes the canceling and switches to CANCELED. In this situation we will already issue release calls for all partitions this vertex has produced.
The strategy however would still consider this vertex as finished, and potentially release preceding partitions.

pushed a fix or this

makes sense to me

...rc/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java

GJL · 2019-07-02T13:02:33Z

LGTM

- Introduce interface PartitionReleaseStrategy. - Introduce RegionPartitionReleaseStrategy and NotReleasingPartitionReleaseStrategy implementations, which can be configured via a new config option. - Add unit tests for new classes. - Increase visibility of methods in TestingSchedulingTopology for unit tests outside of its package.

GJL requested a review from tillrohrmann June 20, 2019 09:34

rmetzger added the review=description? label Jun 20, 2019

zentol reviewed Jun 20, 2019

View reviewed changes

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java Outdated Show resolved Hide resolved

rmetzger added the component=Runtime/Coordination label Jun 20, 2019

GJL commented Jun 20, 2019

View reviewed changes

...rc/test/java/org/apache/flink/runtime/executiongraph/RegionPartitionReleaseStrategyTest.java Outdated Show resolved Hide resolved

GJL commented Jun 20, 2019

View reviewed changes

...runtime/executiongraph/failover/flip1/partitionrelease/PipelinedRegionExecutionViewTest.java Outdated Show resolved Hide resolved

zentol reviewed Jun 20, 2019

View reviewed changes

zentol self-assigned this Jun 20, 2019

GJL commented Jun 20, 2019

View reviewed changes

GJL commented Jun 21, 2019

View reviewed changes

GJL force-pushed the FLINK-12883 branch from 5ebcaa6 to db49e3b Compare June 21, 2019 08:54

zentol reviewed Jun 21, 2019

View reviewed changes

zentol reviewed Jun 22, 2019

View reviewed changes

zhijiangW reviewed Jun 24, 2019

View reviewed changes

zhijiangW reviewed Jun 25, 2019

View reviewed changes

gaoyunhaii reviewed Jun 25, 2019

View reviewed changes

zhijiangW reviewed Jun 25, 2019

View reviewed changes

zhijiangW mentioned this pull request Jun 25, 2019

[FLINK-12960] Introduce ShuffleDescriptor#ReleaseType and ShuffleDescriptor#getSupportedReleaseTypes #8857

Merged

gaoyunhaii reviewed Jun 26, 2019

View reviewed changes

GJL force-pushed the FLINK-12883 branch from aa0d0b8 to 9851cbc Compare June 27, 2019 15:16

GJL force-pushed the FLINK-12883 branch 2 times, most recently from 066a269 to 8f954d0 Compare June 28, 2019 07:14

zentol force-pushed the FLINK-12883 branch from 8f954d0 to 8fd78ca Compare June 29, 2019 16:34

GJL mentioned this pull request Jul 1, 2019

[FLINK-12876][runtime] Add an adapter of region failover NG for legacy scheduler #8922

Closed

zentol reviewed Jul 1, 2019

View reviewed changes

GJL added 3 commits July 1, 2019 13:33

[FLINK-12883][runtime] Extract computation of pipelined regions.

571ce50

[hotfix][runtime] Use Set instead of IdentityHashMap where possible

8a3f540

Replace instances where we use an IdentityHashMap as a set with Collections.newSetFromMap() in RestartPipelinedRegionStrategy and PipelineRegionComputeUtil.

[hotfix][runtime] Remove obsolete comment from PipelinedRegionCompute…

86268c6

…Util#uniqueRegions()

zentol requested changes Jul 1, 2019

View reviewed changes

zentol force-pushed the FLINK-12883 branch from 8fd78ca to fa153fa Compare July 1, 2019 12:18

GJL commented Jul 2, 2019

View reviewed changes

...rc/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java Outdated Show resolved Hide resolved

GJL commented Jul 2, 2019

View reviewed changes

...rc/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java Outdated Show resolved Hide resolved

GJL added 2 commits July 2, 2019 17:07

[FLINK-12883][runtime] Add getID() to ExecutionVertex

16868f0

zentol force-pushed the FLINK-12883 branch from 363434a to 669d1de Compare July 2, 2019 15:07

zentol merged commit c9aa9a1 into apache:master Jul 2, 2019

GJL deleted the FLINK-12883 branch September 3, 2019 09:55


		private final Map<PipelinedRegion, PipelinedRegionConsumedBlockingPartitions> consumedBlockingPartitionsByRegion = new IdentityHashMap<>();

		private final Map<ExecutionVertexID, PipelinedRegionExecutionView> regionExecutionViewByVertex = new HashMap<>();

[FLINK-12883][WIP][runtime] Add elaborated partition release logic #8804

[FLINK-12883][WIP][runtime] Add elaborated partition release logic #8804

Uh oh!

Conversation

GJL commented Jun 20, 2019

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Jun 20, 2019

Review Progress

Uh oh!

Uh oh!

Uh oh!

Uh oh!

zentol left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

GJL Jun 20, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

GJL commented Jun 21, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zentol left a comment

Choose a reason for hiding this comment

Uh oh!

GJL commented Jun 24, 2019

Uh oh!

zentol commented Jun 24, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

GJL Jun 20, 2019 •

edited

Loading

zhijiangW Jun 24, 2019 •

edited

Loading

GJL Jun 25, 2019 •

edited

Loading