Return a 503 status code instead of 400 during transient errors (#15756)

* Fix up HTTP status error code

* Keep the unsupported proxy as 400

* Tests and rename

server/src/main/java/org/apache/druid/server/AsyncManagementForwardingServlet.java

@@ -113,16 +113,22 @@ public class AsyncManagementForwardingServlet extends AsyncProxyServlet
       handleEnabledRequest(response);
       return;
     } else {
-      handleBadRequest(response, StringUtils.format("Unsupported proxy destination [%s]", request.getRequestURI()));
+      handleInvalidRequest(
+          response,
+          StringUtils.format("Unsupported proxy destination[%s]", request.getRequestURI()),
+          HttpServletResponse.SC_BAD_REQUEST
+      );
       return;
     }
 
     if (currentLeader == null) {
-      handleBadRequest(
+      handleInvalidRequest(
           response,
           StringUtils.format(
-              "Unable to determine destination for [%s]; is your coordinator/overlord running?", request.getRequestURI()
+              "Unable to determine destination[%s]; is your coordinator/overlord running?",
-          )
+              request.getRequestURI()
+          ),
+          HttpServletResponse.SC_SERVICE_UNAVAILABLE
       );
       return;
     }
@@ -185,11 +191,11 @@ public class AsyncManagementForwardingServlet extends AsyncProxyServlet
     super.onServerResponseHeaders(clientRequest, proxyResponse, serverResponse);
   }
 
-  private void handleBadRequest(HttpServletResponse response, String errorMessage) throws IOException
+  private void handleInvalidRequest(HttpServletResponse response, String errorMessage, int statusCode) throws IOException
   {
     if (!response.isCommitted()) {
       response.resetBuffer();
-      response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
+      response.setStatus(statusCode);
       jsonMapper.writeValue(response.getOutputStream(), ImmutableMap.of("error", errorMessage));
     }
     response.flushBuffer();

server/src/test/java/org/apache/druid/server/AsyncManagementForwardingServletTest.java

@@ -71,6 +71,7 @@ public class AsyncManagementForwardingServletTest extends BaseJettyTest
 
   private static int coordinatorPort;
   private static int overlordPort;
+  private static boolean isValidLeader;
 
   private Server coordinator;
   private Server overlord;
@@ -109,6 +110,7 @@ public class AsyncManagementForwardingServletTest extends BaseJettyTest
 
     coordinator.start();
     overlord.start();
+    isValidLeader = true;
   }
 
   @After
@@ -119,6 +121,7 @@ public class AsyncManagementForwardingServletTest extends BaseJettyTest
 
     COORDINATOR_EXPECTED_REQUEST.reset();
     OVERLORD_EXPECTED_REQUEST.reset();
+    isValidLeader = true;
   }
 
   @Override
@@ -344,6 +347,45 @@ public class AsyncManagementForwardingServletTest extends BaseJettyTest
     Assert.assertFalse("overlord called", OVERLORD_EXPECTED_REQUEST.called);
   }
 
+  @Test
+  public void testCoordinatorLeaderUnknown() throws Exception
+  {
+    isValidLeader = false;
+    HttpURLConnection connection = ((HttpURLConnection)
+        new URL(StringUtils.format("http://localhost:%d/druid/coordinator", port)).openConnection());
+    connection.setRequestMethod("GET");
+
+    Assert.assertEquals(503, connection.getResponseCode());
+    Assert.assertFalse("coordinator called", COORDINATOR_EXPECTED_REQUEST.called);
+    Assert.assertFalse("overlord called", OVERLORD_EXPECTED_REQUEST.called);
+  }
+
+  @Test
+  public void testOverlordLeaderUnknown() throws Exception
+  {
+    isValidLeader = false;
+    HttpURLConnection connection = ((HttpURLConnection)
+        new URL(StringUtils.format("http://localhost:%d/druid/indexer", port)).openConnection());
+    connection.setRequestMethod("GET");
+
+    Assert.assertEquals(503, connection.getResponseCode());
+    Assert.assertFalse("coordinator called", COORDINATOR_EXPECTED_REQUEST.called);
+    Assert.assertFalse("overlord called", OVERLORD_EXPECTED_REQUEST.called);
+    isValidLeader = true;
+  }
+
+  @Test
+  public void testUnsupportedProxyDestination() throws Exception
+  {
+    HttpURLConnection connection = ((HttpURLConnection)
+        new URL(StringUtils.format("http://localhost:%d/proxy/other/status2", port)).openConnection());
+    connection.setRequestMethod("GET");
+
+    Assert.assertEquals(400, connection.getResponseCode());
+    Assert.assertFalse("coordinator called", COORDINATOR_EXPECTED_REQUEST.called);
+    Assert.assertFalse("overlord called", OVERLORD_EXPECTED_REQUEST.called);
+  }
+
   @Test
   public void testLocalRequest() throws Exception
   {
@@ -422,7 +464,11 @@ public class AsyncManagementForwardingServletTest extends BaseJettyTest
         @Override
         public String getCurrentLeader()
         {
-          return StringUtils.format("http://localhost:%d", coordinatorPort);
+          if (isValidLeader) {
+            return StringUtils.format("http://localhost:%d", coordinatorPort);
+          } else {
+            return null;
+          }
         }
       };
 
@@ -431,7 +477,11 @@ public class AsyncManagementForwardingServletTest extends BaseJettyTest
         @Override
         public String getCurrentLeader()
         {
-          return StringUtils.format("http://localhost:%d", overlordPort);
+          if (isValidLeader) {
+            return StringUtils.format("http://localhost:%d", overlordPort);
+          } else {
+            return null;
+          }
         }
       };