zhijiangW commented on a change in pull request #8362: [FLINK-11391] Introduce
shuffle master interface
URL: https://github.com/apache/flink/pull/8362#discussion_r289692010
##########
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);
Review comment:
It might be no need to make `getKnownConsumedPartitionShuffleDescriptor` as
a separate method, because this method only has three lines.
----------------------------------------------------------------
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
