Fix shutdownAllTasks API for non-existing dataSource (#6706)

indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java

@@ -352,10 +352,14 @@ public class OverlordResource
           public Response apply(TaskQueue taskQueue)
           {
             final List<TaskInfo<Task, TaskStatus>> tasks = taskStorageQueryAdapter.getActiveTaskInfo(dataSource);
-            for (final TaskInfo<Task, TaskStatus> task : tasks) {
+            if (tasks.isEmpty()) {
-              taskQueue.shutdown(task.getId(), "Shutdown request from user");
+              return Response.status(Status.NOT_FOUND).build();
+            } else {
+              for (final TaskInfo<Task, TaskStatus> task : tasks) {
+                taskQueue.shutdown(task.getId(), "Shutdown request from user");
+              }
+              return Response.ok(ImmutableMap.of("dataSource", dataSource)).build();
             }
-            return Response.ok(ImmutableMap.of("dataSource", dataSource)).build();
           }
         }
     );

indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordResourceTest.java

@@ -65,7 +65,9 @@ import org.junit.rules.ExpectedException;
 
 import javax.servlet.http.HttpServletRequest;
 import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -975,6 +977,19 @@ public class OverlordResourceTest
     Assert.assertEquals("datasource", response.get("dataSource"));
   }
 
+  @Test
+  public void testShutdownAllTasksForNonExistingDataSource()
+  {
+    final TaskQueue taskQueue = EasyMock.createMock(TaskQueue.class);
+    EasyMock.expect(taskMaster.isLeader()).andReturn(true).anyTimes();
+    EasyMock.expect(taskMaster.getTaskQueue()).andReturn(Optional.of(taskQueue)).anyTimes();
+    EasyMock.expect(taskStorageQueryAdapter.getActiveTaskInfo(EasyMock.anyString())).andReturn(Collections.emptyList());
+    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+
+    final Response response = overlordResource.shutdownTasksForDataSource("notExisting");
+    Assert.assertEquals(Status.NOT_FOUND.getStatusCode(), response.getStatus());
+  }
+
   private void expectAuthorizationTokenCheck()
   {
     AuthenticationResult authenticationResult = new AuthenticationResult("druid", "druid", null, null);