Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobEdge.java (Java)

Code before:
        if (source == null || target == null || distributionPattern == null) {
            throw new NullPointerException();
        }
        this.target = target;
        this.distributionPattern = distributionPattern;
        this.source = source;
        this.isBroadcast = isBroadcast;
        this.isForward = isForward;
        this.typeNumber = typeNumber;
        this.interInputsKeysCorrelated = interInputsKeysCorrelated;
        this.intraInputKeyCorrelated = intraInputKeyCorrelated;
    }

    public void forceUpdate(
            boolean isForward,
            DistributionPattern distributionPattern,
            SubtaskStateMapper upstreamSubtaskStateMapper,
            SubtaskStateMapper downstreamSubtaskStateMapper,
            String shipStrategyName) {
        this.isForward = isForward;
        this.distributionPattern = distributionPattern;
        this.upstreamSubtaskStateMapper = checkNotNull(upstreamSubtaskStateMapper);
        this.downstreamSubtaskStateMapper = checkNotNull(downstreamSubtaskStateMapper);
        this.shipStrategyName = shipStrategyName;
        this.source.forceUpdate(isForward, distributionPattern);
    }

    /**
     * Returns the data set at the source of the edge. May be null, if the edge refers to the source
     * via an ID and has not been connected.
     *
     * @return The data set at the source of the edge
     */
    public IntermediateDataSet getSource() {
        return source;
    }

    /**
     * Returns the vertex connected to this edge.
     *
     * @return The vertex connected to this edge.
     */
    public JobVertex getTarget() {
        return target;
    }

    /**
     * Returns the distribution pattern used for this edge.
     *
     * @return The distribution pattern used for this edge.
     */

Review comment (suggestion):
I know this might be tricky, but could we prevent adding methods which mutate JobGraph? Any methods on JobGraph which mutate it, should yield a new JobGraph and not touch the existing JobGraph instance. 

Code after:
        if (source == null || target == null || distributionPattern == null) {
            throw new NullPointerException();
        }
        this.target = target;
        this.distributionPattern = distributionPattern;
        this.source = source;
        this.isBroadcast = isBroadcast;
        this.isForward = isForward;
        this.typeNumber = typeNumber;
        this.interInputsKeysCorrelated = interInputsKeysCorrelated;
        this.intraInputKeyCorrelated = intraInputKeyCorrelated;
    }

    public void updateDistributionPattern(
            boolean isForward,
            DistributionPattern distributionPattern,
            SubtaskStateMapper upstreamSubtaskStateMapper,
            SubtaskStateMapper downstreamSubtaskStateMapper,
            String shipStrategyName) {
        this.isForward = isForward;
        this.distributionPattern = distributionPattern;
        this.upstreamSubtaskStateMapper = checkNotNull(upstreamSubtaskStateMapper);
        this.downstreamSubtaskStateMapper = checkNotNull(downstreamSubtaskStateMapper);
        this.shipStrategyName = shipStrategyName;
        this.source.updateDistributionPattern(isForward, distributionPattern);
    }

    /**
     * Returns the data set at the source of the edge. May be null, if the edge refers to the source
     * via an ID and has not been connected.
     *
     * @return The data set at the source of the edge
     */
    public IntermediateDataSet getSource() {
        return source;
    }

    /**
     * Returns the vertex connected to this edge.
     *
     * @return The vertex connected to this edge.
     */
    public JobVertex getTarget() {
        return target;
    }

    /**
     * Returns the distribution pattern used for this edge.
     *
     * @return The distribution pattern used for this edge.
     */