Fixing overlord issued too many redirects (#12908)

* Fixing race in overlord redirects where the node was redirecting to itself

* Fixing test cases

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

@@ -232,6 +232,19 @@ public class TaskMaster implements TaskCountStatsProvider, TaskSlotCountStatsPro
     return overlordLeaderSelector.getCurrentLeader();
   }
 
+  public Optional<String> getRedirectLocation()
+  {
+    String leader = overlordLeaderSelector.getCurrentLeader();
+    // do not redirect when
+    // leader is not elected
+    // leader is the current node
+    if (leader == null || leader.isEmpty() || overlordLeaderSelector.isLeader()) {
+      return Optional.absent();
+    } else {
+      return Optional.of(leader);
+    }
+  }
+
   public Optional<TaskRunner> getTaskRunner()
   {
     if (isLeader()) {

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

@@ -19,6 +19,7 @@
 
 package org.apache.druid.indexing.overlord.http;
 
+import com.google.common.base.Optional;
 import com.google.common.collect.ImmutableSet;
 import com.google.inject.Inject;
 import org.apache.druid.indexing.overlord.TaskMaster;
@@ -55,13 +56,12 @@ public class OverlordRedirectInfo implements RedirectInfo
   public URL getRedirectURL(String queryString, String requestURI)
   {
     try {
-      final String leader = taskMaster.getCurrentLeader();
+      final Optional<String> redirectLocation = taskMaster.getRedirectLocation();
-      if (leader == null || leader.isEmpty()) {
+      if (!redirectLocation.isPresent()) {
         return null;
       }
 
-      String location = StringUtils.format("%s%s", leader, requestURI);
+      String location = StringUtils.format("%s%s", redirectLocation.get(), requestURI);
-
       if (queryString != null) {
         location = StringUtils.format("%s?%s", location, queryString);
       }

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

@@ -19,6 +19,7 @@
 
 package org.apache.druid.indexing.overlord.http;
 
+import com.google.common.base.Optional;
 import org.apache.druid.indexing.overlord.TaskMaster;
 import org.easymock.EasyMock;
 import org.junit.Assert;
@@ -66,19 +67,9 @@ public class OverlordRedirectInfoTest
   }
 
   @Test
-  public void testGetRedirectURLNull()
+  public void testGetRedirectURLWithEmptyLocation()
   {
-    EasyMock.expect(taskMaster.getCurrentLeader()).andReturn(null).anyTimes();
+    EasyMock.expect(taskMaster.getRedirectLocation()).andReturn(Optional.absent()).anyTimes();
-    EasyMock.replay(taskMaster);
-    URL url = redirectInfo.getRedirectURL("query", "/request");
-    Assert.assertNull(url);
-    EasyMock.verify(taskMaster);
-  }
-
-  @Test
-  public void testGetRedirectURLEmpty()
-  {
-    EasyMock.expect(taskMaster.getCurrentLeader()).andReturn("").anyTimes();
     EasyMock.replay(taskMaster);
     URL url = redirectInfo.getRedirectURL("query", "/request");
     Assert.assertNull(url);
@@ -91,7 +82,7 @@ public class OverlordRedirectInfoTest
     String host = "http://localhost";
     String query = "foo=bar&x=y";
     String request = "/request";
-    EasyMock.expect(taskMaster.getCurrentLeader()).andReturn(host).anyTimes();
+    EasyMock.expect(taskMaster.getRedirectLocation()).andReturn(Optional.of(host)).anyTimes();
     EasyMock.replay(taskMaster);
     URL url = redirectInfo.getRedirectURL(query, request);
     Assert.assertEquals("http://localhost/request?foo=bar&x=y", url.toString());
@@ -107,7 +98,7 @@ public class OverlordRedirectInfoTest
         "UTF-8"
     ) + "/status";
 
-    EasyMock.expect(taskMaster.getCurrentLeader()).andReturn(host).anyTimes();
+    EasyMock.expect(taskMaster.getRedirectLocation()).andReturn(Optional.of(host)).anyTimes();
     EasyMock.replay(taskMaster);
     URL url = redirectInfo.getRedirectURL(null, request);
     Assert.assertEquals(

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

@@ -80,7 +80,6 @@ import org.junit.Test;
 import javax.annotation.Nullable;
 import javax.servlet.http.HttpServletRequest;
 import javax.ws.rs.core.Response;
-
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -217,6 +216,7 @@ public class OverlordTest
       Thread.sleep(10);
     }
     Assert.assertEquals(taskMaster.getCurrentLeader(), druidNode.getHostAndPort());
+    Assert.assertEquals(Optional.absent(), taskMaster.getRedirectLocation());
 
     final TaskStorageQueryAdapter taskStorageQueryAdapter = new TaskStorageQueryAdapter(taskStorage, taskLockbox);
     final WorkerTaskRunnerQueryAdapter workerTaskRunnerQueryAdapter = new WorkerTaskRunnerQueryAdapter(taskMaster, null);

server/src/main/java/org/apache/druid/rpc/ServiceClientImpl.java

@@ -200,7 +200,12 @@ public class ServiceClientImpl implements ServiceClient
                         final String newUri = result.error().getResponse().headers().get("Location");
 
                         if (redirectCount >= MAX_REDIRECTS) {
-                          retVal.setException(new RpcException("Service [%s] issued too many redirects", serviceName));
+                          retVal.setException(new RpcException(
+                              "Service [%s] redirected too many times [%d] to invalid url %s",
+                              serviceName,
+                              redirectCount,
+                              newUri
+                          ));
                         } else {
                           // Update preferredLocationNoPath if we got a redirect.
                           final ServiceLocation redirectLocationNoPath = serviceLocationNoPathFromUri(newUri);
@@ -212,7 +217,12 @@ public class ServiceClientImpl implements ServiceClient
                             );
                           } else {
                             retVal.setException(
-                                new RpcException("Service [%s] redirected to invalid URL [%s]", serviceName, newUri)
+                                new RpcException(
+                                    "Service [%s] redirected [%d] times to invalid URL [%s]",
+                                    serviceName,
+                                    redirectCount,
+                                    newUri
+                                )
                             );
                           }
                         }

server/src/test/java/org/apache/druid/rpc/ServiceClientImplTest.java

@@ -289,7 +289,7 @@ public class ServiceClientImplTest
     MatcherAssert.assertThat(e.getCause(), CoreMatchers.instanceOf(RpcException.class));
     MatcherAssert.assertThat(
         e.getCause(),
-        ThrowableMessageMatcher.hasMessage(CoreMatchers.containsString("too many redirects"))
+        ThrowableMessageMatcher.hasMessage(CoreMatchers.containsString("redirected too many times"))
     );
   }
 
@@ -313,7 +313,8 @@ public class ServiceClientImplTest
     MatcherAssert.assertThat(e.getCause(), CoreMatchers.instanceOf(RpcException.class));
     MatcherAssert.assertThat(
         e.getCause(),
-        ThrowableMessageMatcher.hasMessage(CoreMatchers.containsString("redirected to invalid URL [invalid-url]"))
+        ThrowableMessageMatcher.hasMessage(
+            CoreMatchers.containsString("redirected [0] times to invalid URL [invalid-url]"))
     );
   }
 
@@ -337,7 +338,7 @@ public class ServiceClientImplTest
     MatcherAssert.assertThat(e.getCause(), CoreMatchers.instanceOf(RpcException.class));
     MatcherAssert.assertThat(
         e.getCause(),
-        ThrowableMessageMatcher.hasMessage(CoreMatchers.containsString("redirected to invalid URL [null]"))
+        ThrowableMessageMatcher.hasMessage(CoreMatchers.containsString("redirected [0] times to invalid URL [null]"))
     );
   }
 
@@ -361,7 +362,7 @@ public class ServiceClientImplTest
     MatcherAssert.assertThat(e.getCause(), CoreMatchers.instanceOf(RpcException.class));
     MatcherAssert.assertThat(
         e.getCause(),
-        ThrowableMessageMatcher.hasMessage(CoreMatchers.containsString("too many redirects"))
+        ThrowableMessageMatcher.hasMessage(CoreMatchers.containsString("redirected too many times"))
     );
   }