Repository: aurora Updated Branches: refs/heads/master ef0975655 -> 95f0d376f
Removing taskId from ResourceRequest. Reviewed at https://reviews.apache.org/r/33191/ Project: http://git-wip-us.apache.org/repos/asf/aurora/repo Commit: http://git-wip-us.apache.org/repos/asf/aurora/commit/95f0d376 Tree: http://git-wip-us.apache.org/repos/asf/aurora/tree/95f0d376 Diff: http://git-wip-us.apache.org/repos/asf/aurora/diff/95f0d376 Branch: refs/heads/master Commit: 95f0d376fad138f515b38a34e559ca91a7bc8869 Parents: ef09756 Author: Maxim Khutornenko <[email protected]> Authored: Tue Apr 14 15:17:34 2015 -0700 Committer: Maxim Khutornenko <[email protected]> Committed: Tue Apr 14 15:17:34 2015 -0700 ---------------------------------------------------------------------- .../aurora/scheduler/async/TaskScheduler.java | 9 +++++---- .../async/preemptor/PreemptionSlotFinder.java | 2 +- .../scheduler/filter/SchedulingFilter.java | 14 +++---------- .../aurora/scheduler/state/TaskAssigner.java | 11 ++++++---- .../scheduler/async/TaskSchedulerImplTest.java | 8 ++++---- .../scheduler/async/TaskSchedulerTest.java | 21 ++++++++++---------- .../events/NotifyingSchedulingFilterTest.java | 2 +- .../filter/SchedulingFilterImplTest.java | 16 +++++++-------- .../scheduler/state/TaskAssignerImplTest.java | 10 ++++++---- 9 files changed, 45 insertions(+), 48 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/aurora/blob/95f0d376/src/main/java/org/apache/aurora/scheduler/async/TaskScheduler.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/aurora/scheduler/async/TaskScheduler.java b/src/main/java/org/apache/aurora/scheduler/async/TaskScheduler.java index 6f169e8..a500e55 100644 --- a/src/main/java/org/apache/aurora/scheduler/async/TaskScheduler.java +++ b/src/main/java/org/apache/aurora/scheduler/async/TaskScheduler.java @@ -119,7 +119,8 @@ public interface TaskScheduler extends EventSubscriber { private Function<HostOffer, Assignment> getAssignerFunction( final MutableStoreProvider storeProvider, - final ResourceRequest resourceRequest) { + final ResourceRequest resourceRequest, + final String taskId) { // TODO(wfarner): Turn this into Predicate<Offer>, and in the caller, find the first match // and perform the assignment at the very end. This will allow us to use optimistic locking @@ -133,14 +134,14 @@ public interface TaskScheduler extends EventSubscriber { if (reservation.isPresent()) { if (TaskGroupKey.from(resourceRequest.getTask()).equals(reservation.get())) { // Slave is reserved to satisfy this task group. - return assigner.maybeAssign(storeProvider, offer, resourceRequest); + return assigner.maybeAssign(storeProvider, offer, resourceRequest, taskId); } else { // Slave is reserved for another task. return Assignment.failure(); } } else { // Slave is not reserved. - return assigner.maybeAssign(storeProvider, offer, resourceRequest); + return assigner.maybeAssign(storeProvider, offer, resourceRequest, taskId); } } }; @@ -186,7 +187,7 @@ public interface TaskScheduler extends EventSubscriber { AttributeAggregate aggregate = AttributeAggregate.getJobActiveState(store, task.getJob()); try { boolean launched = offerManager.launchFirst( - getAssignerFunction(store, new ResourceRequest(task, taskId, aggregate)), + getAssignerFunction(store, new ResourceRequest(task, aggregate), taskId), TaskGroupKey.from(task)); if (!launched) { http://git-wip-us.apache.org/repos/asf/aurora/blob/95f0d376/src/main/java/org/apache/aurora/scheduler/async/preemptor/PreemptionSlotFinder.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/aurora/scheduler/async/preemptor/PreemptionSlotFinder.java b/src/main/java/org/apache/aurora/scheduler/async/preemptor/PreemptionSlotFinder.java index 427c0de..f16f964 100644 --- a/src/main/java/org/apache/aurora/scheduler/async/preemptor/PreemptionSlotFinder.java +++ b/src/main/java/org/apache/aurora/scheduler/async/preemptor/PreemptionSlotFinder.java @@ -309,7 +309,7 @@ public interface PreemptionSlotFinder { Set<Veto> vetoes = schedulingFilter.filter( new UnusedResource(totalResource, attributes.get()), - new ResourceRequest(pendingTask.getTask(), pendingTask.getTaskId(), jobState)); + new ResourceRequest(pendingTask.getTask(), jobState)); if (vetoes.isEmpty()) { return Optional.of(ImmutableSet.copyOf(toPreemptTasks)); http://git-wip-us.apache.org/repos/asf/aurora/blob/95f0d376/src/main/java/org/apache/aurora/scheduler/filter/SchedulingFilter.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/aurora/scheduler/filter/SchedulingFilter.java b/src/main/java/org/apache/aurora/scheduler/filter/SchedulingFilter.java index dc1ad6e..260e11e 100644 --- a/src/main/java/org/apache/aurora/scheduler/filter/SchedulingFilter.java +++ b/src/main/java/org/apache/aurora/scheduler/filter/SchedulingFilter.java @@ -282,12 +282,10 @@ public interface SchedulingFilter { */ class ResourceRequest { private final ITaskConfig task; - private final String taskId; private final AttributeAggregate jobState; - public ResourceRequest(ITaskConfig task, String taskId, AttributeAggregate jobState) { + public ResourceRequest(ITaskConfig task, AttributeAggregate jobState) { this.task = task; - this.taskId = taskId; this.jobState = jobState; } @@ -303,10 +301,6 @@ public interface SchedulingFilter { return jobState; } - public String getTaskId() { - return taskId; - } - public Set<String> getRequestedPorts() { return task.getRequestedPorts(); } @@ -318,14 +312,12 @@ public interface SchedulingFilter { } ResourceRequest other = (ResourceRequest) o; - return Objects.equals(task, other.task) - && Objects.equals(getTaskId(), other.getTaskId()) - && Objects.equals(getJobState(), other.getJobState()); + return Objects.equals(task, other.task) && Objects.equals(getJobState(), other.getJobState()); } @Override public int hashCode() { - return Objects.hash(task, taskId, jobState); + return Objects.hash(task, jobState); } } http://git-wip-us.apache.org/repos/asf/aurora/blob/95f0d376/src/main/java/org/apache/aurora/scheduler/state/TaskAssigner.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/aurora/scheduler/state/TaskAssigner.java b/src/main/java/org/apache/aurora/scheduler/state/TaskAssigner.java index 1cef4e1..3acb45a 100644 --- a/src/main/java/org/apache/aurora/scheduler/state/TaskAssigner.java +++ b/src/main/java/org/apache/aurora/scheduler/state/TaskAssigner.java @@ -167,12 +167,14 @@ public interface TaskAssigner { * @param storeProvider Storage provider. * @param offer The resource offer. * @param resourceRequest The request for resources being scheduled. + * @param taskId Task id to assign. * @return {@link Assignment} with assignment result. */ Assignment maybeAssign( MutableStoreProvider storeProvider, HostOffer offer, - ResourceRequest resourceRequest); + ResourceRequest resourceRequest, + String taskId); class TaskAssignerImpl implements TaskAssigner { private static final Logger LOG = Logger.getLogger(TaskAssignerImpl.class.getName()); @@ -226,7 +228,8 @@ public interface TaskAssigner { public Assignment maybeAssign( MutableStoreProvider storeProvider, HostOffer offer, - ResourceRequest resourceRequest) { + ResourceRequest resourceRequest, + String taskId) { Set<Veto> vetoes = filter.filter( new UnusedResource(ResourceSlot.from(offer.getOffer()), offer.getAttributes()), @@ -236,10 +239,10 @@ public interface TaskAssigner { storeProvider, offer.getOffer(), resourceRequest.getRequestedPorts(), - resourceRequest.getTaskId())); + taskId)); } else { LOG.fine("Slave " + offer.getOffer().getHostname() - + " vetoed task " + resourceRequest.getTaskId() + ": " + vetoes); + + " vetoed task " + taskId + ": " + vetoes); return Assignment.failure(vetoes); } } http://git-wip-us.apache.org/repos/asf/aurora/blob/95f0d376/src/test/java/org/apache/aurora/scheduler/async/TaskSchedulerImplTest.java ---------------------------------------------------------------------- diff --git a/src/test/java/org/apache/aurora/scheduler/async/TaskSchedulerImplTest.java b/src/test/java/org/apache/aurora/scheduler/async/TaskSchedulerImplTest.java index b0cced7..d879d33 100644 --- a/src/test/java/org/apache/aurora/scheduler/async/TaskSchedulerImplTest.java +++ b/src/test/java/org/apache/aurora/scheduler/async/TaskSchedulerImplTest.java @@ -139,8 +139,8 @@ public class TaskSchedulerImplTest extends EasyMockTest { expect(assigner.maybeAssign( storageUtil.mutableStoreProvider, OFFER, - new ResourceRequest(task.getAssignedTask().getTask(), Tasks.id(task), EMPTY))) - .andReturn(Assignment.success(TaskInfo.getDefaultInstance())); + new ResourceRequest(task.getAssignedTask().getTask(), EMPTY), + Tasks.id(task))).andReturn(Assignment.success(TaskInfo.getDefaultInstance())); } @Test @@ -267,8 +267,8 @@ public class TaskSchedulerImplTest extends EasyMockTest { expect(assigner.maybeAssign( EasyMock.<MutableStoreProvider>anyObject(), eq(OFFER), - eq(new ResourceRequest(taskA.getAssignedTask().getTask(), Tasks.id(taskA), EMPTY)))) - .andReturn(Assignment.success(TaskInfo.getDefaultInstance())); + eq(new ResourceRequest(taskA.getAssignedTask().getTask(), EMPTY)), + eq(Tasks.id(taskA)))).andReturn(Assignment.success(TaskInfo.getDefaultInstance())); control.replay(); http://git-wip-us.apache.org/repos/asf/aurora/blob/95f0d376/src/test/java/org/apache/aurora/scheduler/async/TaskSchedulerTest.java ---------------------------------------------------------------------- diff --git a/src/test/java/org/apache/aurora/scheduler/async/TaskSchedulerTest.java b/src/test/java/org/apache/aurora/scheduler/async/TaskSchedulerTest.java index 858069e..798ff6e 100644 --- a/src/test/java/org/apache/aurora/scheduler/async/TaskSchedulerTest.java +++ b/src/test/java/org/apache/aurora/scheduler/async/TaskSchedulerTest.java @@ -288,7 +288,8 @@ public class TaskSchedulerTest extends EasyMockTest { return expect(assigner.maybeAssign( EasyMock.<MutableStoreProvider>anyObject(), eq(offer), - eq(new ResourceRequest(task.getAssignedTask().getTask(), Tasks.id(task), jobAggregate)))); + eq(new ResourceRequest(task.getAssignedTask().getTask(), jobAggregate)), + eq(Tasks.id(task)))); } private IExpectationSetters<?> expectNoReservation(String slaveId) { @@ -534,16 +535,16 @@ public class TaskSchedulerTest extends EasyMockTest { captureB.getValue().run(); } - private Capture<ResourceRequest> expectTaskScheduled(IScheduledTask task) { + private Capture<String> expectTaskScheduled(IScheduledTask task) { TaskInfo mesosTask = makeTaskInfo(task); - Capture<ResourceRequest> request = createCapture(); + Capture<String> taskId = createCapture(); expect(assigner.maybeAssign( EasyMock.<MutableStoreProvider>anyObject(), EasyMock.<HostOffer>anyObject(), - capture(request))) - .andReturn(Assignment.success(mesosTask)); + EasyMock.<ResourceRequest>anyObject(), + capture(taskId))).andReturn(Assignment.success(mesosTask)); driver.launchTask(EasyMock.<OfferID>anyObject(), eq(mesosTask)); - return request; + return taskId; } @Test @@ -578,8 +579,8 @@ public class TaskSchedulerTest extends EasyMockTest { Capture<Runnable> timeoutA = expectTaskRetryIn(FIRST_SCHEDULE_DELAY_MS); Capture<Runnable> timeoutB = expectTaskRetryIn(FIRST_SCHEDULE_DELAY_MS); - Capture<ResourceRequest> firstScheduled = expectTaskScheduled(jobA0); - Capture<ResourceRequest> secondScheduled = expectTaskScheduled(jobB0); + Capture<String> firstScheduled = expectTaskScheduled(jobA0); + Capture<String> secondScheduled = expectTaskScheduled(jobB0); // Expect another watch of the task group for job A. expectTaskRetryIn(FIRST_SCHEDULE_DELAY_MS); @@ -598,9 +599,7 @@ public class TaskSchedulerTest extends EasyMockTest { timeoutB.getValue().run(); assertEquals( ImmutableSet.of(Tasks.id(jobA0), Tasks.id(jobB0)), - ImmutableSet.of( - firstScheduled.getValue().getTaskId(), - secondScheduled.getValue().getTaskId())); + ImmutableSet.of(firstScheduled.getValue(), secondScheduled.getValue())); } @Test http://git-wip-us.apache.org/repos/asf/aurora/blob/95f0d376/src/test/java/org/apache/aurora/scheduler/events/NotifyingSchedulingFilterTest.java ---------------------------------------------------------------------- diff --git a/src/test/java/org/apache/aurora/scheduler/events/NotifyingSchedulingFilterTest.java b/src/test/java/org/apache/aurora/scheduler/events/NotifyingSchedulingFilterTest.java index 2b71043..abbc8e3 100644 --- a/src/test/java/org/apache/aurora/scheduler/events/NotifyingSchedulingFilterTest.java +++ b/src/test/java/org/apache/aurora/scheduler/events/NotifyingSchedulingFilterTest.java @@ -49,7 +49,7 @@ public class NotifyingSchedulingFilterTest extends EasyMockTest { ResourceSlot.from(TASK, TaskExecutors.NO_OVERHEAD_EXECUTOR), IHostAttributes.build(new HostAttributes().setHost("host").setMode(MaintenanceMode.NONE))); private static final ResourceRequest REQUEST = - new ResourceRequest(TASK, "taskId", AttributeAggregate.EMPTY); + new ResourceRequest(TASK, AttributeAggregate.EMPTY); private static final Veto VETO_1 = Veto.insufficientResources("ram", 1); private static final Veto VETO_2 = Veto.insufficientResources("ram", 2); http://git-wip-us.apache.org/repos/asf/aurora/blob/95f0d376/src/test/java/org/apache/aurora/scheduler/filter/SchedulingFilterImplTest.java ---------------------------------------------------------------------- diff --git a/src/test/java/org/apache/aurora/scheduler/filter/SchedulingFilterImplTest.java b/src/test/java/org/apache/aurora/scheduler/filter/SchedulingFilterImplTest.java index 26bad99..15bb13b 100644 --- a/src/test/java/org/apache/aurora/scheduler/filter/SchedulingFilterImplTest.java +++ b/src/test/java/org/apache/aurora/scheduler/filter/SchedulingFilterImplTest.java @@ -131,22 +131,22 @@ public class SchedulingFilterImplTest extends EasyMockTest { none, defaultFilter.filter( new UnusedResource(twoPorts, hostA), - new ResourceRequest(noPortTask, TASK_ID, EMPTY))); + new ResourceRequest(noPortTask, EMPTY))); assertEquals( none, defaultFilter.filter( new UnusedResource(twoPorts, hostA), - new ResourceRequest(onePortTask, TASK_ID, EMPTY))); + new ResourceRequest(onePortTask, EMPTY))); assertEquals( none, defaultFilter.filter( new UnusedResource(twoPorts, hostA), - new ResourceRequest(twoPortTask, TASK_ID, EMPTY))); + new ResourceRequest(twoPortTask, EMPTY))); assertEquals( ImmutableSet.of(PORTS.veto(1)), defaultFilter.filter( new UnusedResource(twoPorts, hostA), - new ResourceRequest(threePortTask, TASK_ID, EMPTY))); + new ResourceRequest(threePortTask, EMPTY))); } @Test @@ -410,7 +410,7 @@ public class SchedulingFilterImplTest extends EasyMockTest { ImmutableSet.<Veto>of(), defaultFilter.filter( new UnusedResource(DEFAULT_OFFER, hostA), - new ResourceRequest(task, TASK_ID, EMPTY))); + new ResourceRequest(task, EMPTY))); Constraint jvmNegated = jvmConstraint.deepCopy(); jvmNegated.getConstraint().getValue().setNegated(true); @@ -524,7 +524,7 @@ public class SchedulingFilterImplTest extends EasyMockTest { expected, defaultFilter.filter( new UnusedResource(DEFAULT_OFFER, hostAttributes), - new ResourceRequest(task, TASK_ID, aggregate)) + new ResourceRequest(task, aggregate)) .isEmpty()); Constraint negated = constraint.deepCopy(); @@ -534,7 +534,7 @@ public class SchedulingFilterImplTest extends EasyMockTest { !expected, defaultFilter.filter( new UnusedResource(DEFAULT_OFFER, hostAttributes), - new ResourceRequest(negatedTask, TASK_ID, aggregate)) + new ResourceRequest(negatedTask, aggregate)) .isEmpty()); return task; } @@ -565,7 +565,7 @@ public class SchedulingFilterImplTest extends EasyMockTest { ImmutableSet.copyOf(vetoes), defaultFilter.filter( new UnusedResource(DEFAULT_OFFER, hostAttributes), - new ResourceRequest(task, TASK_ID, jobState))); + new ResourceRequest(task, jobState))); } private static IHostAttributes hostAttributes( http://git-wip-us.apache.org/repos/asf/aurora/blob/95f0d376/src/test/java/org/apache/aurora/scheduler/state/TaskAssignerImplTest.java ---------------------------------------------------------------------- diff --git a/src/test/java/org/apache/aurora/scheduler/state/TaskAssignerImplTest.java b/src/test/java/org/apache/aurora/scheduler/state/TaskAssignerImplTest.java index ff95c36..702a01a 100644 --- a/src/test/java/org/apache/aurora/scheduler/state/TaskAssignerImplTest.java +++ b/src/test/java/org/apache/aurora/scheduler/state/TaskAssignerImplTest.java @@ -101,7 +101,7 @@ public class TaskAssignerImplTest extends EasyMockTest { public void testAssignNoVetoes() { expect(filter.filter( new UnusedResource(ResourceSlot.from(MESOS_OFFER), OFFER.getAttributes()), - new ResourceRequest(TASK.getAssignedTask().getTask(), Tasks.id(TASK), EMPTY))) + new ResourceRequest(TASK.getAssignedTask().getTask(), EMPTY))) .andReturn(ImmutableSet.<Veto>of()); expect(stateManager.assignTask( storeProvider, @@ -120,14 +120,15 @@ public class TaskAssignerImplTest extends EasyMockTest { assigner.maybeAssign( storeProvider, OFFER, - new ResourceRequest(TASK.getAssignedTask().getTask(), Tasks.id(TASK), EMPTY))); + new ResourceRequest(TASK.getAssignedTask().getTask(), EMPTY), + Tasks.id(TASK))); } @Test public void testAssignVetoes() { expect(filter.filter( new UnusedResource(ResourceSlot.from(MESOS_OFFER), OFFER.getAttributes()), - new ResourceRequest(TASK.getAssignedTask().getTask(), Tasks.id(TASK), EMPTY))) + new ResourceRequest(TASK.getAssignedTask().getTask(), EMPTY))) .andReturn(ImmutableSet.of(Veto.constraintMismatch("denied"))); control.replay(); @@ -137,6 +138,7 @@ public class TaskAssignerImplTest extends EasyMockTest { assigner.maybeAssign( storeProvider, OFFER, - new ResourceRequest(TASK.getAssignedTask().getTask(), Tasks.id(TASK), EMPTY))); + new ResourceRequest(TASK.getAssignedTask().getTask(), EMPTY), + Tasks.id(TASK))); } }
