@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))); }
@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; }); }
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; }); } }
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); }
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(); }
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();  }