flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tillrohrmann <...@git.apache.org>
Subject [GitHub] flink pull request #5091: [FLINK-7956] [flip6] Add support for queued schedu...
Date Thu, 14 Dec 2017 10:20:30 GMT
Github user tillrohrmann commented on a diff in the pull request:

    https://github.com/apache/flink/pull/5091#discussion_r156903740
  
    --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduledUnit.java
---
    @@ -19,68 +19,104 @@
     package org.apache.flink.runtime.jobmanager.scheduler;
     
     import org.apache.flink.runtime.executiongraph.Execution;
    +import org.apache.flink.runtime.instance.SlotSharingGroupId;
     import org.apache.flink.runtime.jobgraph.JobVertexID;
     import org.apache.flink.util.Preconditions;
     
    +import javax.annotation.Nullable;
    +
     public class ScheduledUnit {
    -	
    +
    +	@Nullable
     	private final Execution vertexExecution;
    -	
    -	private final SlotSharingGroup sharingGroup;
    -	
    -	private final CoLocationConstraint locationConstraint;
    +
    +	private final JobVertexID jobVertexId;
    +
    +	@Nullable
    +	private final SlotSharingGroupId slotSharingGroupId;
    +
    +	@Nullable
    +	private final CoLocationConstraint coLocationConstraint;
     	
     	// --------------------------------------------------------------------------------------------
     	
     	public ScheduledUnit(Execution task) {
    -		Preconditions.checkNotNull(task);
    -		
    -		this.vertexExecution = task;
    -		this.sharingGroup = null;
    -		this.locationConstraint = null;
    +		this(
    +			Preconditions.checkNotNull(task),
    +			task.getVertex().getJobvertexId(),
    +			null,
    +			null);
     	}
     	
    -	public ScheduledUnit(Execution task, SlotSharingGroup sharingUnit) {
    -		Preconditions.checkNotNull(task);
    -		
    -		this.vertexExecution = task;
    -		this.sharingGroup = sharingUnit;
    -		this.locationConstraint = null;
    +	public ScheduledUnit(Execution task, SlotSharingGroupId slotSharingGroupId) {
    +		this(
    +			Preconditions.checkNotNull(task),
    +			task.getVertex().getJobvertexId(),
    +			slotSharingGroupId,
    +			null);
     	}
     	
    -	public ScheduledUnit(Execution task, SlotSharingGroup sharingUnit, CoLocationConstraint
locationConstraint) {
    -		Preconditions.checkNotNull(task);
    -		Preconditions.checkNotNull(sharingUnit);
    -		Preconditions.checkNotNull(locationConstraint);
    -		
    +	public ScheduledUnit(
    +			Execution task,
    +			SlotSharingGroupId slotSharingGroupId,
    +			CoLocationConstraint coLocationConstraint) {
    +		this(
    +			Preconditions.checkNotNull(task),
    +			task.getVertex().getJobvertexId(),
    +			slotSharingGroupId,
    +			coLocationConstraint);
    +	}
    +
    +	public ScheduledUnit(
    +			JobVertexID jobVertexId,
    +			SlotSharingGroupId slotSharingGroupId,
    +			CoLocationConstraint coLocationConstraint) {
    +		this(
    +			null,
    +			jobVertexId,
    +			slotSharingGroupId,
    +			coLocationConstraint);
    +	}
    +
    +	public ScheduledUnit(
    +		Execution task,
    +		JobVertexID jobVertexId,
    --- End diff --
    
    Yes because `task` can be null. Will add the missing `@Nullable` annotation.


---

Mime
View raw message