@Test public void testPurgeMisplacedTasks() throws Exception { HostState host = installHostStateWithUUID("host", spawn, true, new HostLocation("", "", "")); HostState otherHost = installHostStateWithUUID("otherHost", spawn, true, new HostLocation("", "", "")); waitForAllUpHosts(); Job job = createSpawnJob(spawn, 1, Arrays.asList("host"), now, 80_000_000_000L, 0); host.setStopped(simulateJobKeys(job)); host.setMax(new HostCapacity(10, 10, 10, 100_000_000_000L)); host.setUsed(new HostCapacity(10, 10, 10, 90_900_000_000L)); spawn.getJobCommandManager().putEntity("foo", new JobCommand(), false); hostManager.updateHostState(host); bal.updateAggregateStatistics(hostManager.listHostStatus(null)); spawn.deleteJob(job.getId()); int optimizedDirsIndex = spawn.rebalanceHost("host").toString().indexOf("JobTaskMoveAssignment"); assertTrue("should not fail to delete orphan task", optimizedDirsIndex != -1); }
private List<JobTaskMoveAssignment> purgeMisplacedTasks(HostState host, int deleteLimit) { List<JobTaskMoveAssignment> rv = new ArrayList<>(); for (JobKey key : host.allJobKeys()) { if (spawn.getJob(key) == null) { rv.add(new JobTaskMoveAssignment(key, host.getHostUuid(), null, false, true)); } else { JobTask task = spawn.getTask(key); if (!host.getHostUuid().equals(task.getHostUUID()) && !task.hasReplicaOnHost(host.getHostUuid())) { spawn.fixTaskDir(key.getJobUuid(), key.getNodeNumber(), false, false); deleteLimit -= 1; } } if (rv.size() >= deleteLimit) { break; } } return rv; }
SpawnBalancer implements Codable, AutoCloseable { private List<JobTaskMoveAssignment> purgeMisplacedTasks(HostState host, int deleteLimit) { List<JobTaskMoveAssignment> rv = new ArrayList<>(); for (JobKey key : host.allJobKeys()) { if (spawn.getJob(key) == null) { rv.add(new JobTaskMoveAssignment(key, host.getHostUuid(), null, false, true)); } else { JobTask task = spawn.getTask(key); if (!host.getHostUuid().equals(task.getHostUUID()) && !task.hasReplicaOnHost(host.getHostUuid())) { spawn.fixTaskDir(key.getJobUuid(), key.getNodeNumber(), false, false); deleteLimit -= 1; } } if (rv.size() >= deleteLimit) { break; } } return rv; } }
SpawnBalancer implements Codable, AutoCloseable { private List<JobTaskMoveAssignment> purgeMisplacedTasks(HostState host, int deleteLimit) { List<JobTaskMoveAssignment> rv = new ArrayList<>(); for (JobKey key : host.allJobKeys()) { if (spawn.getJob(key) == null) { rv.add(new JobTaskMoveAssignment(key, host.getHostUuid(), null, false, true)); } else { JobTask task = spawn.getTask(key); if (!host.getHostUuid().equals(task.getHostUUID()) && !task.hasReplicaOnHost(host.getHostUuid())) { spawn.fixTaskDir(key.getJobUuid(), key.getNodeNumber(), false, false); deleteLimit -= 1; } } if (rv.size() >= deleteLimit) { break; } } return rv; } SpawnBalancer(Spawn spawn, HostManager hostManager); }
SpawnBalancer implements Codable, AutoCloseable { private List<JobTaskMoveAssignment> purgeMisplacedTasks(HostState host, int deleteLimit) { List<JobTaskMoveAssignment> rv = new ArrayList<>(); for (JobKey key : host.allJobKeys()) { if (spawn.getJob(key) == null) { rv.add(new JobTaskMoveAssignment(key, host.getHostUuid(), null, false, true)); } else { JobTask task = spawn.getTask(key); if (!host.getHostUuid().equals(task.getHostUUID()) && !task.hasReplicaOnHost(host.getHostUuid())) { spawn.fixTaskDir(key.getJobUuid(), key.getNodeNumber(), false, false); deleteLimit -= 1; } } if (rv.size() >= deleteLimit) { break; } } return rv; } SpawnBalancer(Spawn spawn, HostManager hostManager); double getHostScoreCached(String hostId); synchronized void startAutobalanceTask(); @Override void close(); void saveConfigToDataStore(); void startTaskSizePolling(); Map<JobTask, String> assignTasksFromMultipleJobsToHosts(Collection<JobTask> tasks,
                                                                   Collection<HostState> hosts); Map<HostState, Double> generateHostStateScoreMap(Collection<HostState> hosts); boolean canReceiveNewTasks(HostState host); List<JobTaskMoveAssignment> pushTasksOffHostForFilesystemOkayFailure(HostState host, int moveLimit); long getTaskTrueSize(JobTask task); static boolean isInMovableState(JobTask task); boolean okayToAutobalance(); @Nullable List<JobTaskMoveAssignment> getAssignmentsForAutoBalance(RebalanceType type,
                                                                              RebalanceWeight weight); List<Job> getJobsToAutobalance(List<HostState> hosts); Map<Integer, List<String>> getAssignmentsForNewReplicas(IJob job); Map<Integer, List<String>> getAssignmentsForNewReplicas(IJob job, int taskID); void requestJobSizeUpdate(String jobId, int taskId); SpawnBalancerConfig getConfig(); void setConfig(SpawnBalancerConfig config); void clearRecentlyRebalancedHosts(); void fixTasksForFailedHost(List<HostState> hosts, String failedHost); List<JobTask> generateAssignedTasksForNewJob(String jobID, int taskCount, Collection<HostState> hosts); List<JobTaskMoveAssignment> getAssignmentsForJobReallocation(Job job,
                                                                        int tasksToMove,
                                                                        List<HostState> hosts); List<JobTaskMoveAssignment> getAssignmentsToBalanceHost(HostState host, List<HostState> hosts); List<HostState> sortHostsByActiveTasks(Collection<HostState> hosts); void removeInvalidReplicas(IJob job); boolean isDiskFull(@Nullable HostState host); @SuppressWarnings("FloatingPointEquality") @Nonnull Collection<JobTask> getTasksToMove(String fromHostUuid, String toHostUuid,
                                                       int numTasks, boolean obeyDontAutobalanceMe); boolean isTaskSpreadOutAcrossAd(@Nullable HostLocation fromHostLocation,
                                           @Nullable HostLocation toHostLocation, JobTask task); }
SpawnBalancer implements Codable, AutoCloseable { private List<JobTaskMoveAssignment> purgeMisplacedTasks(HostState host, int deleteLimit) { List<JobTaskMoveAssignment> rv = new ArrayList<>(); for (JobKey key : host.allJobKeys()) { if (spawn.getJob(key) == null) { rv.add(new JobTaskMoveAssignment(key, host.getHostUuid(), null, false, true)); } else { JobTask task = spawn.getTask(key); if (!host.getHostUuid().equals(task.getHostUUID()) && !task.hasReplicaOnHost(host.getHostUuid())) { spawn.fixTaskDir(key.getJobUuid(), key.getNodeNumber(), false, false); deleteLimit -= 1; } } if (rv.size() >= deleteLimit) { break; } } return rv; } SpawnBalancer(Spawn spawn, HostManager hostManager); double getHostScoreCached(String hostId); synchronized void startAutobalanceTask(); @Override void close(); void saveConfigToDataStore(); void startTaskSizePolling(); Map<JobTask, String> assignTasksFromMultipleJobsToHosts(Collection<JobTask> tasks,
                                                                   Collection<HostState> hosts); Map<HostState, Double> generateHostStateScoreMap(Collection<HostState> hosts); boolean canReceiveNewTasks(HostState host); List<JobTaskMoveAssignment> pushTasksOffHostForFilesystemOkayFailure(HostState host, int moveLimit); long getTaskTrueSize(JobTask task); static boolean isInMovableState(JobTask task); boolean okayToAutobalance(); @Nullable List<JobTaskMoveAssignment> getAssignmentsForAutoBalance(RebalanceType type,
                                                                              RebalanceWeight weight); List<Job> getJobsToAutobalance(List<HostState> hosts); Map<Integer, List<String>> getAssignmentsForNewReplicas(IJob job); Map<Integer, List<String>> getAssignmentsForNewReplicas(IJob job, int taskID); void requestJobSizeUpdate(String jobId, int taskId); SpawnBalancerConfig getConfig(); void setConfig(SpawnBalancerConfig config); void clearRecentlyRebalancedHosts(); void fixTasksForFailedHost(List<HostState> hosts, String failedHost); List<JobTask> generateAssignedTasksForNewJob(String jobID, int taskCount, Collection<HostState> hosts); List<JobTaskMoveAssignment> getAssignmentsForJobReallocation(Job job,
                                                                        int tasksToMove,
                                                                        List<HostState> hosts); List<JobTaskMoveAssignment> getAssignmentsToBalanceHost(HostState host, List<HostState> hosts); List<HostState> sortHostsByActiveTasks(Collection<HostState> hosts); void removeInvalidReplicas(IJob job); boolean isDiskFull(@Nullable HostState host); @SuppressWarnings("FloatingPointEquality") @Nonnull Collection<JobTask> getTasksToMove(String fromHostUuid, String toHostUuid,
                                                       int numTasks, boolean obeyDontAutobalanceMe); boolean isTaskSpreadOutAcrossAd(@Nullable HostLocation fromHostLocation,
                                           @Nullable HostLocation toHostLocation, JobTask task);  }