@Test public void testGetTimeInStateForFailedTask() { TaskStatus checked = TaskStatus.newBuilder().withState(TaskState.KillInitiated).withTimestamp(1000).build(); Task task = REFERENCE_TASK.toBuilder() .withStatus(checked) .withStatusHistory( TaskStatus.newBuilder().withState(TaskState.Accepted).withTimestamp(0).build(), TaskStatus.newBuilder().withState(TaskState.Launched).withTimestamp(100).build() ) .build(); assertThat(JobFunctions.getTimeInState(task, TaskState.Accepted, clock)).contains(100L); assertThat(JobFunctions.getTimeInState(task, TaskState.Launched, clock)).contains(900L); assertThat(JobFunctions.getTimeInState(task, TaskState.StartInitiated, clock)).isEmpty(); assertThat(JobFunctions.getTimeInState(task, TaskState.Started, clock)).isEmpty(); assertThat(JobFunctions.getTimeInState(task, TaskState.KillInitiated, clock).get()).isGreaterThan(0); assertThat(JobFunctions.getTimeInState(task, TaskState.Finished, clock)).isEmpty(); }
public static Optional<Long> getTimeInState(Task task, TaskState checkedState, Clock clock) { return findTaskStatus(task, checkedState).map(checkedStatus -> { TaskState currentState = task.getStatus().getState(); if (currentState == checkedState) { return clock.wallTime() - task.getStatus().getTimestamp(); } if (TaskState.isAfter(checkedState, currentState)) { return 0L; } return findStatusAfter(task, checkedState).map(after -> after.getTimestamp() - checkedStatus.getTimestamp()).orElse(0L); }); }
JobFunctions { public static Optional<Long> getTimeInState(Task task, TaskState checkedState, Clock clock) { return findTaskStatus(task, checkedState).map(checkedStatus -> { TaskState currentState = task.getStatus().getState(); if (currentState == checkedState) { return clock.wallTime() - task.getStatus().getTimestamp(); } if (TaskState.isAfter(checkedState, currentState)) { return 0L; } return findStatusAfter(task, checkedState).map(after -> after.getTimestamp() - checkedStatus.getTimestamp()).orElse(0L); }); } }
JobFunctions { public static Optional<Long> getTimeInState(Task task, TaskState checkedState, Clock clock) { return findTaskStatus(task, checkedState).map(checkedStatus -> { TaskState currentState = task.getStatus().getState(); if (currentState == checkedState) { return clock.wallTime() - task.getStatus().getTimestamp(); } if (TaskState.isAfter(checkedState, currentState)) { return 0L; } return findStatusAfter(task, checkedState).map(after -> after.getTimestamp() - checkedStatus.getTimestamp()).orElse(0L); }); } private  JobFunctions(); }
JobFunctions { public static Optional<Long> getTimeInState(Task task, TaskState checkedState, Clock clock) { return findTaskStatus(task, checkedState).map(checkedStatus -> { TaskState currentState = task.getStatus().getState(); if (currentState == checkedState) { return clock.wallTime() - task.getStatus().getTimestamp(); } if (TaskState.isAfter(checkedState, currentState)) { return 0L; } return findStatusAfter(task, checkedState).map(after -> after.getTimestamp() - checkedStatus.getTimestamp()).orElse(0L); }); } private  JobFunctions(); static DisruptionBudget getNoDisruptionBudgetMarker(); static JobType getJobType(Job<?> job); static JobType getJobType(JobDescriptor<?> jobDescriptor); static boolean isBatchJob(Job<?> job); static boolean isBatchJob(JobDescriptor<?> jobDescriptor); static Job<BatchJobExt> asBatchJob(Job<?> job); static JobDescriptor<BatchJobExt> asBatchJob(JobDescriptor<?> jobDescriptor); static boolean isServiceJob(Job<?> job); static boolean isServiceJob(JobDescriptor<?> jobDescriptor); static boolean isDisabled(Job<?> job); static Job<ServiceJobExt> asServiceJob(Job<?> job); static JobDescriptor<ServiceJobExt> asServiceJob(JobDescriptor<?> jobDescriptor); static int getJobDesiredSize(Job<?> job); static boolean isBatchTask(Task task); static boolean isServiceTask(Task task); static boolean hasDisruptionBudget(JobDescriptor<?> jobDescriptor); static boolean hasDisruptionBudget(Job<?> job); static boolean everStarted(Task task); static Job changeJobStatus(Job job, JobState jobState, String reasonCode); static Job changeJobStatus(Job job, JobStatus status); static JobDescriptor<BatchJobExt> changeBatchJobSize(JobDescriptor<BatchJobExt> jobDescriptor, int size); static JobDescriptor<?> filterOutGeneratedAttributes(JobDescriptor<?> jobDescriptor); static JobDescriptor<E> appendJobDescriptorAttributes(JobDescriptor<E> jobDescriptor,
                                                                                              Map<String, ?> attributes); static JobDescriptor<E> appendJobDescriptorAttribute(JobDescriptor<E> jobDescriptor,
                                                                                             String attributeName,
                                                                                             Object attributeValue); static JobDescriptor<E> appendContainerAttribute(JobDescriptor<E> jobDescriptor,
                                                                                         String attributeName,
                                                                                         Object attributeValue); static Job<E> appendContainerAttribute(Job<E> job,
                                                                               String attributeName,
                                                                               Object attributeValue); static JobDescriptor<E> appendHardConstraint(JobDescriptor<E> jobDescriptor,
                                                                                     String name,
                                                                                     String value); static JobDescriptor<E> appendSoftConstraint(JobDescriptor<E> jobDescriptor,
                                                                                     String name,
                                                                                     String value); static Job<E> appendHardConstraint(Job<E> job, String name, String value); static Job<E> appendSoftConstraint(Job<E> job, String name, String value); static Job<E> appendJobDescriptorAttribute(Job<E> job,
                                                                                   String attributeName,
                                                                                   Object attributeValue); static Optional<String> findHardConstraint(Job<E> job, String name); static Optional<String> findSoftConstraint(Job<E> job, String name); static Task appendTaskAttribute(Task task, String attributeName, Object attributeValue); static Job<BatchJobExt> changeBatchJobSize(Job<BatchJobExt> job, int size); static JobDescriptor<ServiceJobExt> changeServiceJobCapacity(JobDescriptor<ServiceJobExt> jobDescriptor, int size); static JobDescriptor<ServiceJobExt> changeServiceJobCapacity(JobDescriptor<ServiceJobExt> jobDescriptor, Capacity capacity); static Job<ServiceJobExt> changeServiceJobCapacity(Job<ServiceJobExt> job, Capacity capacity); static Job<ServiceJobExt> changeServiceJobCapacity(Job<ServiceJobExt> job, CapacityAttributes capacityAttributes); static JobDescriptor<E> changeSecurityGroups(JobDescriptor<E> jobDescriptor, List<String> securityGroups); static JobDescriptor<E> incrementJobDescriptorSize(JobDescriptor<E> jobDescriptor, int delta); static Job<E> incrementJobSize(Job<E> job, int delta); static Job<ServiceJobExt> changeJobEnabledStatus(Job<ServiceJobExt> job, boolean enabled); static Job<ServiceJobExt> changeServiceJobProcesses(Job<ServiceJobExt> job, ServiceJobProcesses processes); static Task changeTaskStatus(Task task, TaskStatus status); static Task changeTaskStatus(Task task, TaskState taskState, String reasonCode, String reasonMessage, Clock clock); static Task fixArchivedTaskStatus(Task task, Clock clock); static Task addAllocatedResourcesToTask(Task task, TaskStatus status, TwoLevelResource twoLevelResource, Map<String, String> taskContext); static Function<Job<E>, Job<E>> withJobId(String jobId); static Function<JobDescriptor<BatchJobExt>, JobDescriptor<BatchJobExt>> ofBatchSize(int size); static Function<JobDescriptor<ServiceJobExt>, JobDescriptor<ServiceJobExt>> ofServiceSize(int size); static Function<JobDescriptor<E>, JobDescriptor<E>> havingProvider(String name, String... attributes); static Function<JobDescriptor<E>, JobDescriptor<E>> withDisruptionBudget(DisruptionBudget disruptionBudget); static Function<Job<E>, Job<E>> withJobDisruptionBudget(DisruptionBudget disruptionBudget); static Retryer retryerFrom(RetryPolicy retryPolicy); static Retryer retryer(Job<?> job); static RetryPolicy getRetryPolicy(Job<?> job); static JobDescriptor<E> changeRetryPolicy(JobDescriptor<E> input, RetryPolicy retryPolicy); static JobDescriptor<BatchJobExt> changeRetryLimit(JobDescriptor<BatchJobExt> input, int retryLimit); static JobDescriptor<E> changeDisruptionBudget(JobDescriptor<E> input, DisruptionBudget disruptionBudget); static Job<E> changeDisruptionBudget(Job<E> input, DisruptionBudget disruptionBudget); static Job<E> updateJobAttributes(Job<E> input, Map<String, String> attributes); static Job<E> deleteJobAttributes(Job<E> input, Set<String> keys); static Optional<Long> getTimeInState(Task task, TaskState checkedState, Clock clock); static Task moveTask(String jobIdFrom, String jobIdTo, Task taskBefore); static boolean containsExactlyTaskStates(Task task, TaskState... expectedStates); static Optional<JobStatus> findJobStatus(Job<?> job, JobState checkedState); static Optional<TaskStatus> findTaskStatus(Task task, TaskState checkedState); static Optional<TaskStatus> findStatusAfter(Task task, TaskState before); static Optional<Integer> getOpportunisticCpuCount(Task task); static Optional<Duration> getJobRuntimePrediction(Job job); @SuppressWarnings("unchecked") static Optional<Duration> getJobRuntimePrediction(JobDescriptor jobDescriptor); static boolean isOwnedByKubeScheduler(Task task); static String getEffectiveCapacityGroup(Job job); }
JobFunctions { public static Optional<Long> getTimeInState(Task task, TaskState checkedState, Clock clock) { return findTaskStatus(task, checkedState).map(checkedStatus -> { TaskState currentState = task.getStatus().getState(); if (currentState == checkedState) { return clock.wallTime() - task.getStatus().getTimestamp(); } if (TaskState.isAfter(checkedState, currentState)) { return 0L; } return findStatusAfter(task, checkedState).map(after -> after.getTimestamp() - checkedStatus.getTimestamp()).orElse(0L); }); } private  JobFunctions(); static DisruptionBudget getNoDisruptionBudgetMarker(); static JobType getJobType(Job<?> job); static JobType getJobType(JobDescriptor<?> jobDescriptor); static boolean isBatchJob(Job<?> job); static boolean isBatchJob(JobDescriptor<?> jobDescriptor); static Job<BatchJobExt> asBatchJob(Job<?> job); static JobDescriptor<BatchJobExt> asBatchJob(JobDescriptor<?> jobDescriptor); static boolean isServiceJob(Job<?> job); static boolean isServiceJob(JobDescriptor<?> jobDescriptor); static boolean isDisabled(Job<?> job); static Job<ServiceJobExt> asServiceJob(Job<?> job); static JobDescriptor<ServiceJobExt> asServiceJob(JobDescriptor<?> jobDescriptor); static int getJobDesiredSize(Job<?> job); static boolean isBatchTask(Task task); static boolean isServiceTask(Task task); static boolean hasDisruptionBudget(JobDescriptor<?> jobDescriptor); static boolean hasDisruptionBudget(Job<?> job); static boolean everStarted(Task task); static Job changeJobStatus(Job job, JobState jobState, String reasonCode); static Job changeJobStatus(Job job, JobStatus status); static JobDescriptor<BatchJobExt> changeBatchJobSize(JobDescriptor<BatchJobExt> jobDescriptor, int size); static JobDescriptor<?> filterOutGeneratedAttributes(JobDescriptor<?> jobDescriptor); static JobDescriptor<E> appendJobDescriptorAttributes(JobDescriptor<E> jobDescriptor,
                                                                                              Map<String, ?> attributes); static JobDescriptor<E> appendJobDescriptorAttribute(JobDescriptor<E> jobDescriptor,
                                                                                             String attributeName,
                                                                                             Object attributeValue); static JobDescriptor<E> appendContainerAttribute(JobDescriptor<E> jobDescriptor,
                                                                                         String attributeName,
                                                                                         Object attributeValue); static Job<E> appendContainerAttribute(Job<E> job,
                                                                               String attributeName,
                                                                               Object attributeValue); static JobDescriptor<E> appendHardConstraint(JobDescriptor<E> jobDescriptor,
                                                                                     String name,
                                                                                     String value); static JobDescriptor<E> appendSoftConstraint(JobDescriptor<E> jobDescriptor,
                                                                                     String name,
                                                                                     String value); static Job<E> appendHardConstraint(Job<E> job, String name, String value); static Job<E> appendSoftConstraint(Job<E> job, String name, String value); static Job<E> appendJobDescriptorAttribute(Job<E> job,
                                                                                   String attributeName,
                                                                                   Object attributeValue); static Optional<String> findHardConstraint(Job<E> job, String name); static Optional<String> findSoftConstraint(Job<E> job, String name); static Task appendTaskAttribute(Task task, String attributeName, Object attributeValue); static Job<BatchJobExt> changeBatchJobSize(Job<BatchJobExt> job, int size); static JobDescriptor<ServiceJobExt> changeServiceJobCapacity(JobDescriptor<ServiceJobExt> jobDescriptor, int size); static JobDescriptor<ServiceJobExt> changeServiceJobCapacity(JobDescriptor<ServiceJobExt> jobDescriptor, Capacity capacity); static Job<ServiceJobExt> changeServiceJobCapacity(Job<ServiceJobExt> job, Capacity capacity); static Job<ServiceJobExt> changeServiceJobCapacity(Job<ServiceJobExt> job, CapacityAttributes capacityAttributes); static JobDescriptor<E> changeSecurityGroups(JobDescriptor<E> jobDescriptor, List<String> securityGroups); static JobDescriptor<E> incrementJobDescriptorSize(JobDescriptor<E> jobDescriptor, int delta); static Job<E> incrementJobSize(Job<E> job, int delta); static Job<ServiceJobExt> changeJobEnabledStatus(Job<ServiceJobExt> job, boolean enabled); static Job<ServiceJobExt> changeServiceJobProcesses(Job<ServiceJobExt> job, ServiceJobProcesses processes); static Task changeTaskStatus(Task task, TaskStatus status); static Task changeTaskStatus(Task task, TaskState taskState, String reasonCode, String reasonMessage, Clock clock); static Task fixArchivedTaskStatus(Task task, Clock clock); static Task addAllocatedResourcesToTask(Task task, TaskStatus status, TwoLevelResource twoLevelResource, Map<String, String> taskContext); static Function<Job<E>, Job<E>> withJobId(String jobId); static Function<JobDescriptor<BatchJobExt>, JobDescriptor<BatchJobExt>> ofBatchSize(int size); static Function<JobDescriptor<ServiceJobExt>, JobDescriptor<ServiceJobExt>> ofServiceSize(int size); static Function<JobDescriptor<E>, JobDescriptor<E>> havingProvider(String name, String... attributes); static Function<JobDescriptor<E>, JobDescriptor<E>> withDisruptionBudget(DisruptionBudget disruptionBudget); static Function<Job<E>, Job<E>> withJobDisruptionBudget(DisruptionBudget disruptionBudget); static Retryer retryerFrom(RetryPolicy retryPolicy); static Retryer retryer(Job<?> job); static RetryPolicy getRetryPolicy(Job<?> job); static JobDescriptor<E> changeRetryPolicy(JobDescriptor<E> input, RetryPolicy retryPolicy); static JobDescriptor<BatchJobExt> changeRetryLimit(JobDescriptor<BatchJobExt> input, int retryLimit); static JobDescriptor<E> changeDisruptionBudget(JobDescriptor<E> input, DisruptionBudget disruptionBudget); static Job<E> changeDisruptionBudget(Job<E> input, DisruptionBudget disruptionBudget); static Job<E> updateJobAttributes(Job<E> input, Map<String, String> attributes); static Job<E> deleteJobAttributes(Job<E> input, Set<String> keys); static Optional<Long> getTimeInState(Task task, TaskState checkedState, Clock clock); static Task moveTask(String jobIdFrom, String jobIdTo, Task taskBefore); static boolean containsExactlyTaskStates(Task task, TaskState... expectedStates); static Optional<JobStatus> findJobStatus(Job<?> job, JobState checkedState); static Optional<TaskStatus> findTaskStatus(Task task, TaskState checkedState); static Optional<TaskStatus> findStatusAfter(Task task, TaskState before); static Optional<Integer> getOpportunisticCpuCount(Task task); static Optional<Duration> getJobRuntimePrediction(Job job); @SuppressWarnings("unchecked") static Optional<Duration> getJobRuntimePrediction(JobDescriptor jobDescriptor); static boolean isOwnedByKubeScheduler(Task task); static String getEffectiveCapacityGroup(Job job);  }