Skip to content

Refresh DruidLeaderClient cache selectively for non-200 responses #14092

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gianm merged 8 commits into from
Apr 20, 2023
Merged

Refresh DruidLeaderClient cache selectively for non-200 responses #14092

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
73769ff
Refresh DruidLeaderClient cache for non-200 responses
abhishek-chouhan Apr 14, 2023
cbbbecf
Change local variable name to avoid confusion
abhishek-chouhan Apr 14, 2023
ce73e52
Implicit retries for 503 and 504
abhishek-chouhan Apr 19, 2023
f8e5122
Remove unused imports
abhishek-chouhan Apr 19, 2023
a6662d2
Use argumentmatcher instead of Mockito for #any in test
abhishek-chouhan Apr 19, 2023
e51c352
Remove flag to disable retry for 503/504
Apr 19, 2023
e288a90
Remove unused import from test
abhishek-chouhan Apr 19, 2023
beb8817
Add log line for internal retry
abhishek-chouhan Apr 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
97 changes: 65 additions & 32 deletions server/src/main/java/org/apache/druid/discovery/DruidLeaderClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,15 +122,32 @@ public Request makeRequest(HttpMethod httpMethod, String urlPath) throws IOExcep
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));
}

/**
* 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));
Expand All @@ -152,38 +169,11 @@ public <T, H extends FullResponseHolder<T>> H go(Request request, HttpResponseHa
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()
))
);
}
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()
);
}
request = getNewRequestUrlInvalidatingCache(request);
continue;
}

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());
Expand Down Expand Up @@ -213,6 +203,16 @@ public <T, H extends FullResponseHolder<T>> H go(Request request, HttpResponseHa
));

request = withUrl(request, redirectUrl);
} else if (HttpResponseStatus.SERVICE_UNAVAILABLE.equals(responseStatus)
Copy link
Copy Markdown
Contributor

@abhishekagarwal87 abhishekagarwal87 Apr 20, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I missed this but if there is a retry happening in response to 503/504, there will be no way to see that happening in logs. We should be logging a warning here. what do you think?

Copy link
Copy Markdown
Contributor Author

@abhishek-chouhan abhishek-chouhan Apr 20, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Makes sense. Added a log.

|| 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;
}
Expand Down Expand Up @@ -295,4 +295,37 @@ private Request withUrl(Request old, URL url)
}
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()
);
}
}
}
43 changes: 43 additions & 0 deletions server/src/test/java/org/apache/druid/discovery/DruidLeaderClientTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@

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;
Expand All @@ -39,6 +40,7 @@
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;
Expand All @@ -49,11 +51,16 @@
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;
Expand All @@ -63,6 +70,7 @@
import javax.ws.rs.core.Response;
import java.io.IOException;
import java.net.URI;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;

/**
Expand Down Expand Up @@ -211,6 +219,41 @@ public void testServerFailureAndRedirect() throws Exception
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()
{
Expand Down