Do not retry ON_DEMAND tasks which were killed by user request.

SingularityService/src/main/java/com/hubspot/singularity/scheduler/SingularityScheduler.java

@@ -849,10 +849,19 @@ private boolean shouldRetryImmediately(SingularityRequest request, SingularityDe
       return false;
     }
 
-    if (task.isPresent()
-        && task.get().getTaskRequest().getPendingTask().getPendingTaskId().getPendingType() == PendingType.IMMEDIATE
-        && request.getRequestType() == RequestType.SCHEDULED) {
-      return false; // don't retry UI triggered scheduled jobs (UI triggered on-demand jobs are okay to retry though)
+    if (task.isPresent()) {
+      if (task.get().getTaskRequest().getPendingTask().getPendingTaskId().getPendingType() == PendingType.IMMEDIATE
+          && request.getRequestType() == RequestType.SCHEDULED) {
+        return false; // don't retry UI triggered scheduled jobs (UI triggered on-demand jobs are okay to retry though)
+      }
+
+      Optional<SingularityTaskHistoryUpdate> taskHistoryUpdate = taskManager.getTaskHistoryUpdate(task.get().getTaskId(), ExtendedTaskState.TASK_CLEANING);
+
+      if (taskHistoryUpdate.isPresent()
+          && request.getRequestType() == RequestType.ON_DEMAND
+          && taskHistoryUpdate.get().getStatusMessage().orElse("").contains("USER_REQUESTED")) {
+        return false; // don't retry one-off launches of on-demand jobs if they were killed by the user
+      }
     }
 
     final int numRetriesInARow = deployStatistics.getNumSequentialRetries();

SingularityService/src/test/java/com/hubspot/singularity/scheduler/SingularitySchedulerTest.java

@@ -1619,6 +1619,40 @@ public void testRunNowOnDemandJobMayRetryOnFailure() {
     Assertions.assertEquals(1, deployStatistics.getNumSequentialRetries());
   }
 
+  @Test
+  public void testRunNowOnDemandJobsDoNotRetryAfterUserRequestedKill() {
+    initRequestWithType(RequestType.ON_DEMAND, false);
+    SingularityRequest request = requestResource.getRequest(requestId, singularityUser).getRequest();
+    SingularityRequest newRequest = request.toBuilder().setNumRetriesOnFailure(Optional.of(2)).build();
+    requestResource.postRequest(newRequest, singularityUser);
+    initFirstDeploy();
+
+    requestResource.scheduleImmediately(singularityUser, requestId, new SingularityRunNowRequestBuilder().setMessage("foo bar").build());
+    scheduler.drainPendingQueue();
+    resourceOffers();
+
+    SingularityTask task = taskManager.getActiveTasks().get(0);
+    taskManager.saveTaskCleanup(new SingularityTaskCleanup(
+        Optional.of(singularityUser.getId()),
+        TaskCleanupType.USER_REQUESTED,
+        System.currentTimeMillis(),
+        task.getTaskId(),
+        Optional.empty(),
+        Optional.empty(),
+        Optional.empty())
+    );
+    cleaner.drainCleanupQueue();
+    statusUpdate(task, TaskState.TASK_KILLED);
+    scheduler.drainPendingQueue();
+
+    SingularityDeployStatistics deployStatistics = deployManager.getDeployStatistics(task.getTaskId().getRequestId(), task.getTaskId().getDeployId()).get();
+
+    Assertions.assertEquals(0, taskManager.getPendingTaskIds().size());
+    Assertions.assertEquals(MesosTaskState.TASK_KILLED, deployStatistics.getLastTaskState().get().toTaskState().get());
+    Assertions.assertEquals(0, deployStatistics.getNumFailures());
+    Assertions.assertEquals(0, deployStatistics.getNumSequentialRetries());
+  }
+
   @Test
   public void testOnDemandRunNowJobRespectsSpecifiedRunAtTime() {
     initOnDemandRequest();