PendingTaskProcessor implements Runnable { @Override public void run() { metrics.recordTaskProcessorRun(); storage.read(store -> { Multimap<String, PreemptionVictim> slavesToActiveTasks = clusterState.getSlavesToActiveTasks(); if (slavesToActiveTasks.isEmpty()) { return null; } Map<String, HostOffer> slavesToOffers = Maps.uniqueIndex(offerManager.getOffers(), OFFER_TO_SLAVE_ID); Set<String> allSlaves = Sets.newHashSet(Iterables.concat( slavesToOffers.keySet(), slavesToActiveTasks.keySet())); LoadingCache<IJobKey, AttributeAggregate> jobStates = attributeCache(store); List<TaskGroupKey> pendingGroups = fetchIdlePendingGroups(store); Iterator<TaskGroupKey> groups = Iterators.consumingIterator(pendingGroups.iterator()); while (!pendingGroups.isEmpty()) { boolean matched = false; TaskGroupKey group = groups.next(); ITaskConfig task = group.getTask(); metrics.recordPreemptionAttemptFor(task); Iterator<String> slaveIterator = allSlaves.iterator(); while (slaveIterator.hasNext()) { String slaveId = slaveIterator.next(); Optional<ImmutableSet<PreemptionVictim>> candidates = preemptionVictimFilter.filterPreemptionVictims( task, slavesToActiveTasks.get(slaveId), jobStates.getUnchecked(task.getJob()), Optional.fromNullable(slavesToOffers.get(slaveId)), store); metrics.recordSlotSearchResult(candidates, task); if (candidates.isPresent()) { slaveIterator.remove(); slotCache.put(new PreemptionProposal(candidates.get(), slaveId), group); matched = true; break; } } if (!matched) { pendingGroups.removeAll(ImmutableSet.of(group)); groups = Iterators.consumingIterator(pendingGroups.iterator()); } } return null; }); } @Inject  PendingTaskProcessor(
      Storage storage,
      OfferManager offerManager,
      PreemptionVictimFilter preemptionVictimFilter,
      PreemptorMetrics metrics,
      @PreemptionDelay Amount<Long, Time> preemptionCandidacyDelay,
      BiCache<PreemptionProposal, TaskGroupKey> slotCache,
      ClusterState clusterState,
      Clock clock); @Override void run();  }
@Test public void testSearchSlotSuccessful() throws Exception { expectGetPendingTasks(TASK_A, TASK_B); expectGetClusterState(TASK_A, TASK_B); HostOffer offer1 = makeOffer(SLAVE_ID_1); HostOffer offer2 = makeOffer(SLAVE_ID_2); expectOffers(offer1, offer2); expectSlotSearch(TASK_A.getAssignedTask().getTask(), TASK_A); expectSlotSearch(TASK_B.getAssignedTask().getTask(), TASK_B); control.replay(); clock.advance(PREEMPTION_DELAY); slotFinder.run(); assertEquals(1L, statsProvider.getLongValue(TASK_PROCESSOR_RUN_NAME)); assertEquals(2L, statsProvider.getLongValue(attemptsStatName(true))); assertEquals(2L, statsProvider.getLongValue(slotSearchStatName(true, true))); assertEquals(0L, statsProvider.getLongValue(slotSearchStatName(false, true))); assertEquals(2L, statsProvider.getLongValue(CACHE_STAT)); }
@Test public void testSearchSlotFailed() throws Exception { expectGetPendingTasks(TASK_A); expectGetClusterState(TASK_A); HostOffer offer1 = makeOffer(SLAVE_ID_1); expectOffers(offer1); expectSlotSearch(TASK_A.getAssignedTask().getTask()); control.replay(); clock.advance(PREEMPTION_DELAY); slotFinder.run(); assertEquals(1L, statsProvider.getLongValue(TASK_PROCESSOR_RUN_NAME)); assertEquals(1L, statsProvider.getLongValue(attemptsStatName(true))); assertEquals(0L, statsProvider.getLongValue(slotSearchStatName(true, true))); assertEquals(1L, statsProvider.getLongValue(slotSearchStatName(false, true))); }
@Test public void testHasCachedSlots() throws Exception { slotCache.put(SLOT_A, group(TASK_A)); expectGetPendingTasks(TASK_A); expectGetClusterState(TASK_A); HostOffer offer1 = makeOffer(SLAVE_ID_1); expectOffers(offer1); control.replay(); clock.advance(PREEMPTION_DELAY); slotFinder.run(); assertEquals(1L, statsProvider.getLongValue(TASK_PROCESSOR_RUN_NAME)); assertEquals(0L, statsProvider.getLongValue(attemptsStatName(true))); assertEquals(0L, statsProvider.getLongValue(slotSearchStatName(true, true))); assertEquals(0L, statsProvider.getLongValue(slotSearchStatName(false, true))); }
@Test public void testMultipleTaskGroups() throws Exception { IScheduledTask task1 = makeTask(JOB_A, "1"); IScheduledTask task2 = makeTask(JOB_A, "2"); IScheduledTask task3 = makeTask(JOB_A, "3"); IScheduledTask task4 = makeTask(JOB_B, "4"); IScheduledTask task5 = makeTask(JOB_B, "5"); expectGetPendingTasks(task1, task4, task2, task5, task3); expectGetClusterState(TASK_A, TASK_B); HostOffer offer1 = makeOffer(SLAVE_ID_1); HostOffer offer2 = makeOffer(SLAVE_ID_2); expectOffers(offer1, offer2); expectSlotSearch(task1.getAssignedTask().getTask()); expectSlotSearch(task4.getAssignedTask().getTask(), TASK_B); PreemptionProposal proposal1 = createPreemptionProposal(TASK_B, SLAVE_ID_1); PreemptionProposal proposal2 = createPreemptionProposal(TASK_B, SLAVE_ID_2); control.replay(); clock.advance(PREEMPTION_DELAY); slotFinder.run(); assertEquals(slotCache.get(proposal1), Optional.of(group(task4))); assertEquals(slotCache.get(proposal2), Optional.of(group(task5))); assertEquals(1L, statsProvider.getLongValue(TASK_PROCESSOR_RUN_NAME)); assertEquals(3L, statsProvider.getLongValue(attemptsStatName(true))); assertEquals(2L, statsProvider.getLongValue(slotSearchStatName(true, true))); assertTrue(ImmutableSet.of(1L, 2L).contains( statsProvider.getLongValue(slotSearchStatName(false, true)))); assertEquals(2L, statsProvider.getLongValue(CACHE_STAT)); }
@Test public void testNoVictims() throws Exception { expectGetClusterState(); control.replay(); clock.advance(PREEMPTION_DELAY); slotFinder.run(); assertEquals(1L, statsProvider.getLongValue(TASK_PROCESSOR_RUN_NAME)); assertEquals(0L, statsProvider.getLongValue(attemptsStatName(true))); assertEquals(0L, statsProvider.getLongValue(slotSearchStatName(true, true))); assertEquals(0L, statsProvider.getLongValue(slotSearchStatName(false, true))); }