azagrebin commented on a change in pull request #8362: [FLINK-11391] Introduce
shuffle master interface
URL: https://github.com/apache/flink/pull/8362#discussion_r289810369
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/PartitionInfo.java
##########
@@ -18,48 +18,56 @@
package org.apache.flink.runtime.executiongraph;
-import org.apache.flink.runtime.deployment.InputChannelDeploymentDescriptor;
+import org.apache.flink.runtime.deployment.ResultPartitionDeploymentDescriptor;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
-import org.apache.flink.util.Preconditions;
+import org.apache.flink.runtime.shuffle.ShuffleDescriptor;
import java.io.Serializable;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkState;
/**
* Contains information where to find a partition. The partition is defined by
the
- * {@link IntermediateDataSetID} and the partition location is specified by
- * {@link InputChannelDeploymentDescriptor}.
+ * {@link IntermediateDataSetID} and the partition is specified by
+ * {@link ShuffleDescriptor}.
*/
public class PartitionInfo implements Serializable {
private static final long serialVersionUID = 1724490660830968430L;
private final IntermediateDataSetID intermediateDataSetID;
- private final InputChannelDeploymentDescriptor
inputChannelDeploymentDescriptor;
- public PartitionInfo(IntermediateDataSetID
intermediateResultPartitionID, InputChannelDeploymentDescriptor
inputChannelDeploymentDescriptor) {
- this.intermediateDataSetID =
Preconditions.checkNotNull(intermediateResultPartitionID);
- this.inputChannelDeploymentDescriptor =
Preconditions.checkNotNull(inputChannelDeploymentDescriptor);
+ private final ShuffleDescriptor shuffleDescriptor;
+
+ public PartitionInfo(
+ IntermediateDataSetID intermediateResultPartitionID,
+ ShuffleDescriptor shuffleDescriptor) {
+ this.intermediateDataSetID = intermediateResultPartitionID;
+ this.shuffleDescriptor = shuffleDescriptor;
}
public IntermediateDataSetID getIntermediateDataSetID() {
return intermediateDataSetID;
}
- public InputChannelDeploymentDescriptor
getInputChannelDeploymentDescriptor() {
- return inputChannelDeploymentDescriptor;
+ public ShuffleDescriptor getShuffleDescriptor() {
+ return shuffleDescriptor;
}
- //
------------------------------------------------------------------------
-
static PartitionInfo fromEdge(ExecutionEdge executionEdge) {
- final InputChannelDeploymentDescriptor
inputChannelDeploymentDescriptor =
InputChannelDeploymentDescriptor.fromEdge(executionEdge);
-
- Preconditions.checkState(
-
!inputChannelDeploymentDescriptor.getConsumedPartitionLocation().isUnknown(),
- "PartitionInfo contains an unknown partition
location.");
+ IntermediateDataSetID intermediateDataSetID =
executionEdge.getSource().getIntermediateResult().getId();
+ ShuffleDescriptor shuffleDescriptor =
getKnownConsumedPartitionShuffleDescriptor(executionEdge);
+ return new PartitionInfo(intermediateDataSetID,
shuffleDescriptor);
+ }
- return new PartitionInfo(
-
executionEdge.getSource().getIntermediateResult().getId(),
- inputChannelDeploymentDescriptor);
+ private static ShuffleDescriptor
getKnownConsumedPartitionShuffleDescriptor(ExecutionEdge edge) {
Review comment:
I think we can reuse it just by passing `allowLazyDeployment=true` to
`getConsumedPartitionShuffleDescriptor`.
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
