mirror of
https://github.com/apache/druid.git
synced 2025-02-26 04:55:24 +00:00
Refresh DruidLeaderClient cache selectively for non-200 responses (#14092)
* Refresh DruidLeaderClient cache for non-200 responses * Change local variable name to avoid confusion * Implicit retries for 503 and 504 * Remove unused imports * Use argumentmatcher instead of Mockito for #any in test * Remove flag to disable retry for 503/504 * Remove unused import from test * Add log line for internal retry --------- Co-authored-by: Abhishek Singh Chouhan <abhishek.chouhan@salesforce.com>
This commit is contained in:
Abhishek Singh Chouhan
GitHub
parent
4fffee1776
commit
895abd8929
@ -122,6 +122,14 @@ public class DruidLeaderClient
|
||||
return new Request(httpMethod, new URL(StringUtils.format("%s%s", getCurrentKnownLeader(true), urlPath)));
|
||||
}
|
||||
|
||||
/**
|
||||
* Executes a Request object aimed at the leader. Throws IOException if the leader cannot be located.
|
||||
* Internal retries with cache invalidation are attempted if 503/504 response is received.
|
||||
*
|
||||
* @param request
|
||||
* @throws IOException
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
public StringFullResponseHolder go(Request request) throws IOException, InterruptedException
|
||||
{
|
||||
return go(request, new StringFullResponseHandler(StandardCharsets.UTF_8));
|
||||
@ -129,8 +137,17 @@ public class DruidLeaderClient
|
||||
|
||||
/**
|
||||
* Executes a Request object aimed at the leader. Throws IOException if the leader cannot be located.
|
||||
* Internal retries with cache invalidation are attempted if 503/504 response is received.
|
||||
*
|
||||
* @param request
|
||||
* @param responseHandler
|
||||
* @throws IOException
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
public <T, H extends FullResponseHolder<T>> H go(Request request, HttpResponseHandler<H, H> responseHandler)
|
||||
public <T, H extends FullResponseHolder<T>> H go(
|
||||
Request request,
|
||||
HttpResponseHandler<H, H> responseHandler
|
||||
)
|
||||
throws IOException, InterruptedException
|
||||
{
|
||||
Preconditions.checkState(lifecycleLock.awaitStarted(1, TimeUnit.MILLISECONDS));
|
||||
@ -152,38 +169,11 @@ public class DruidLeaderClient
|
||||
catch (IOException | ChannelException ex) {
|
||||
// can happen if the node is stopped.
|
||||
log.warn(ex, "Request[%s] failed.", request.getUrl());
|
||||
|
||||
try {
|
||||
if (request.getUrl().getQuery() == null) {
|
||||
request = withUrl(
|
||||
request,
|
||||
new URL(StringUtils.format("%s%s", getCurrentKnownLeader(false), request.getUrl().getPath()))
|
||||
);
|
||||
} else {
|
||||
request = withUrl(
|
||||
request,
|
||||
new URL(StringUtils.format(
|
||||
"%s%s?%s",
|
||||
getCurrentKnownLeader(false),
|
||||
request.getUrl().getPath(),
|
||||
request.getUrl().getQuery()
|
||||
))
|
||||
);
|
||||
}
|
||||
request = getNewRequestUrlInvalidatingCache(request);
|
||||
continue;
|
||||
}
|
||||
catch (MalformedURLException e) {
|
||||
// Not an IOException; this is our own fault.
|
||||
throw new ISE(
|
||||
e,
|
||||
"failed to build url with path[%] and query string [%s].",
|
||||
request.getUrl().getPath(),
|
||||
request.getUrl().getQuery()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if (HttpResponseStatus.TEMPORARY_REDIRECT.equals(fullResponseHolder.getResponse().getStatus())) {
|
||||
HttpResponseStatus responseStatus = fullResponseHolder.getResponse().getStatus();
|
||||
if (HttpResponseStatus.TEMPORARY_REDIRECT.equals(responseStatus)) {
|
||||
String redirectUrlStr = fullResponseHolder.getResponse().headers().get("Location");
|
||||
if (redirectUrlStr == null) {
|
||||
throw new IOE("No redirect location is found in response from url[%s].", request.getUrl());
|
||||
@ -213,6 +203,16 @@ public class DruidLeaderClient
|
||||
));
|
||||
|
||||
request = withUrl(request, redirectUrl);
|
||||
} else if (HttpResponseStatus.SERVICE_UNAVAILABLE.equals(responseStatus)
|
||||
|| HttpResponseStatus.GATEWAY_TIMEOUT.equals(responseStatus)) {
|
||||
log.warn(
|
||||
"Request[%s] received a %s response. Attempt %s/%s",
|
||||
request.getUrl(),
|
||||
responseStatus,
|
||||
counter + 1,
|
||||
MAX_RETRIES
|
||||
);
|
||||
request = getNewRequestUrlInvalidatingCache(request);
|
||||
} else {
|
||||
return fullResponseHolder;
|
||||
}
|
||||
@ -295,4 +295,37 @@ public class DruidLeaderClient
|
||||
}
|
||||
return req;
|
||||
}
|
||||
|
||||
private Request getNewRequestUrlInvalidatingCache(Request oldRequest) throws IOException
|
||||
{
|
||||
try {
|
||||
Request newRequest;
|
||||
if (oldRequest.getUrl().getQuery() == null) {
|
||||
newRequest = withUrl(
|
||||
oldRequest,
|
||||
new URL(StringUtils.format("%s%s", getCurrentKnownLeader(false), oldRequest.getUrl().getPath()))
|
||||
);
|
||||
} else {
|
||||
newRequest = withUrl(
|
||||
oldRequest,
|
||||
new URL(StringUtils.format(
|
||||
"%s%s?%s",
|
||||
getCurrentKnownLeader(false),
|
||||
oldRequest.getUrl().getPath(),
|
||||
oldRequest.getUrl().getQuery()
|
||||
))
|
||||
);
|
||||
}
|
||||
return newRequest;
|
||||
}
|
||||
catch (MalformedURLException e) {
|
||||
// Not an IOException; this is our own fault.
|
||||
throw new ISE(
|
||||
e,
|
||||
"failed to build url with path[%] and query string [%s].",
|
||||
oldRequest.getUrl().getPath(),
|
||||
oldRequest.getUrl().getQuery()
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -21,6 +21,7 @@ package org.apache.druid.discovery;
|
||||
|
||||
import com.google.common.collect.ImmutableList;
|
||||
import com.google.common.collect.ImmutableMap;
|
||||
import com.google.common.util.concurrent.ListenableFutureTask;
|
||||
import com.google.inject.Binder;
|
||||
import com.google.inject.Inject;
|
||||
import com.google.inject.Injector;
|
||||
@ -39,6 +40,7 @@ import org.apache.druid.initialization.Initialization;
|
||||
import org.apache.druid.java.util.common.StringUtils;
|
||||
import org.apache.druid.java.util.http.client.HttpClient;
|
||||
import org.apache.druid.java.util.http.client.Request;
|
||||
import org.apache.druid.java.util.http.client.response.StringFullResponseHolder;
|
||||
import org.apache.druid.server.DruidNode;
|
||||
import org.apache.druid.server.initialization.BaseJettyTest;
|
||||
import org.apache.druid.server.initialization.jetty.JettyServerInitializer;
|
||||
@ -49,11 +51,16 @@ import org.eclipse.jetty.server.handler.HandlerList;
|
||||
import org.eclipse.jetty.servlet.DefaultServlet;
|
||||
import org.eclipse.jetty.servlet.ServletContextHandler;
|
||||
import org.eclipse.jetty.servlet.ServletHolder;
|
||||
import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
|
||||
import org.jboss.netty.handler.codec.http.HttpMethod;
|
||||
import org.jboss.netty.handler.codec.http.HttpResponseStatus;
|
||||
import org.jboss.netty.handler.codec.http.HttpVersion;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
import org.junit.rules.ExpectedException;
|
||||
import org.mockito.ArgumentMatchers;
|
||||
import org.mockito.Mockito;
|
||||
|
||||
import javax.ws.rs.GET;
|
||||
import javax.ws.rs.POST;
|
||||
@ -63,6 +70,7 @@ import javax.ws.rs.core.MediaType;
|
||||
import javax.ws.rs.core.Response;
|
||||
import java.io.IOException;
|
||||
import java.net.URI;
|
||||
import java.nio.charset.Charset;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
/**
|
||||
@ -211,6 +219,41 @@ public class DruidLeaderClientTest extends BaseJettyTest
|
||||
Assert.assertEquals("hello", druidLeaderClient.go(request).getContent());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void test503ResponseFromServerAndCacheRefresh() throws Exception
|
||||
{
|
||||
DruidNodeDiscovery druidNodeDiscovery = EasyMock.createMock(DruidNodeDiscovery.class);
|
||||
// Should be called twice. Second time is when we refresh the cache since we get 503 in the first request
|
||||
EasyMock.expect(druidNodeDiscovery.getAllNodes()).andReturn(ImmutableList.of(discoveryDruidNode)).times(2);
|
||||
|
||||
DruidNodeDiscoveryProvider druidNodeDiscoveryProvider = EasyMock.createMock(DruidNodeDiscoveryProvider.class);
|
||||
EasyMock.expect(druidNodeDiscoveryProvider.getForNodeRole(NodeRole.PEON)).andReturn(druidNodeDiscovery).anyTimes();
|
||||
|
||||
ListenableFutureTask task = EasyMock.createMock(ListenableFutureTask.class);
|
||||
EasyMock.expect(task.get()).andReturn(new StringFullResponseHolder(new DefaultHttpResponse(
|
||||
HttpVersion.HTTP_1_1,
|
||||
HttpResponseStatus.SERVICE_UNAVAILABLE
|
||||
), Charset.defaultCharset()));
|
||||
EasyMock.replay(druidNodeDiscovery, druidNodeDiscoveryProvider, task);
|
||||
|
||||
HttpClient spyHttpClient = Mockito.spy(this.httpClient);
|
||||
// Override behavior for the first call only
|
||||
Mockito.doReturn(task).doCallRealMethod().when(spyHttpClient).go(ArgumentMatchers.any(), ArgumentMatchers.any());
|
||||
|
||||
DruidLeaderClient druidLeaderClient = new DruidLeaderClient(
|
||||
spyHttpClient,
|
||||
druidNodeDiscoveryProvider,
|
||||
NodeRole.PEON,
|
||||
"/simple/leader"
|
||||
);
|
||||
druidLeaderClient.start();
|
||||
|
||||
Request request = druidLeaderClient.makeRequest(HttpMethod.POST, "/simple/direct");
|
||||
request.setContent("hello".getBytes(StandardCharsets.UTF_8));
|
||||
Assert.assertEquals("hello", druidLeaderClient.go(request).getContent());
|
||||
EasyMock.verify(druidNodeDiscovery);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFindCurrentLeader()
|
||||
{
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user