Fixed TimelineClient to retry SocketTimeoutException too. Contributed by Xuan Gong.

This commit is contained in:
Vinod Kumar Vavilapalli 2016-04-18 11:47:06 -07:00
parent d8b729e16f
commit 477003730e
2 changed files with 93 additions and 22 deletions

View File

@ -24,6 +24,7 @@
import java.net.ConnectException; import java.net.ConnectException;
import java.net.HttpURLConnection; import java.net.HttpURLConnection;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.net.SocketTimeoutException;
import java.net.URI; import java.net.URI;
import java.net.URL; import java.net.URL;
import java.net.URLConnection; import java.net.URLConnection;
@ -116,7 +117,9 @@ public class TimelineClientImpl extends TimelineClient {
TimelineClientConnectionRetry connectionRetry; TimelineClientConnectionRetry connectionRetry;
// Abstract class for an operation that should be retried by timeline client // Abstract class for an operation that should be retried by timeline client
private static abstract class TimelineClientRetryOp { @Private
@VisibleForTesting
public static abstract class TimelineClientRetryOp {
// The operation that should be retried // The operation that should be retried
public abstract Object run() throws IOException; public abstract Object run() throws IOException;
// The method to indicate if we should retry given the incoming exception // The method to indicate if we should retry given the incoming exception
@ -449,27 +452,8 @@ private Object operateDelegationToken(
final PrivilegedExceptionAction<?> action) final PrivilegedExceptionAction<?> action)
throws IOException, YarnException { throws IOException, YarnException {
// Set up the retry operation // Set up the retry operation
TimelineClientRetryOp tokenRetryOp = new TimelineClientRetryOp() { TimelineClientRetryOp tokenRetryOp =
createTimelineClientRetryOpForOperateDelegationToken(action);
@Override
public Object run() throws IOException {
// Try pass the request, if fail, keep retrying
authUgi.checkTGTAndReloginFromKeytab();
try {
return authUgi.doAs(action);
} catch (UndeclaredThrowableException e) {
throw new IOException(e.getCause());
} catch (InterruptedException e) {
throw new IOException(e);
}
}
@Override
public boolean shouldRetryOn(Exception e) {
// Only retry on connection exceptions
return (e instanceof ConnectException);
}
};
return connectionRetry.retryOn(tokenRetryOp); return connectionRetry.retryOn(tokenRetryOp);
} }
@ -680,4 +664,50 @@ public void putDomain(ApplicationAttemptId appAttemptId,
public void setTimelineWriter(TimelineWriter writer) { public void setTimelineWriter(TimelineWriter writer) {
this.timelineWriter = writer; this.timelineWriter = writer;
} }
@Private
@VisibleForTesting
public TimelineClientRetryOp
createTimelineClientRetryOpForOperateDelegationToken(
final PrivilegedExceptionAction<?> action) throws IOException {
return new TimelineClientRetryOpForOperateDelegationToken(
this.authUgi, action);
}
@Private
@VisibleForTesting
public class TimelineClientRetryOpForOperateDelegationToken
extends TimelineClientRetryOp {
private final UserGroupInformation authUgi;
private final PrivilegedExceptionAction<?> action;
public TimelineClientRetryOpForOperateDelegationToken(
UserGroupInformation authUgi, PrivilegedExceptionAction<?> action) {
this.authUgi = authUgi;
this.action = action;
}
@Override
public Object run() throws IOException {
// Try pass the request, if fail, keep retrying
authUgi.checkTGTAndReloginFromKeytab();
try {
return authUgi.doAs(action);
} catch (UndeclaredThrowableException e) {
throw new IOException(e.getCause());
} catch (InterruptedException e) {
throw new IOException(e);
}
}
@Override
public boolean shouldRetryOn(Exception e) {
// retry on connection exceptions
// and SocketTimeoutException
return (e instanceof ConnectException
|| e instanceof SocketTimeoutException);
}
}
} }

View File

@ -27,7 +27,9 @@
import java.io.IOException; import java.io.IOException;
import java.net.ConnectException; import java.net.ConnectException;
import java.net.SocketTimeoutException;
import java.net.URI; import java.net.URI;
import java.security.PrivilegedExceptionAction;
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic; import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
@ -234,6 +236,8 @@ public void testDelegationTokenOperationsRetry() throws Exception {
UserGroupInformation.setConfiguration(conf); UserGroupInformation.setConfiguration(conf);
TimelineClientImpl client = createTimelineClient(conf); TimelineClientImpl client = createTimelineClient(conf);
TimelineClientImpl clientFake =
createTimelineClientFakeTimelineClientRetryOp(conf);
TestTimlineDelegationTokenSecretManager dtManager = TestTimlineDelegationTokenSecretManager dtManager =
new TestTimlineDelegationTokenSecretManager(); new TestTimlineDelegationTokenSecretManager();
try { try {
@ -278,8 +282,24 @@ public void testDelegationTokenOperationsRetry() throws Exception {
} catch (RuntimeException ce) { } catch (RuntimeException ce) {
assertException(client, ce); assertException(client, ce);
} }
// Test DelegationTokenOperationsRetry on SocketTimeoutException
try {
TimelineDelegationTokenIdentifier timelineDT =
new TimelineDelegationTokenIdentifier(
new Text("tester"), new Text("tester"), new Text("tester"));
clientFake.cancelDelegationToken(
new Token<TimelineDelegationTokenIdentifier>(timelineDT.getBytes(),
dtManager.createPassword(timelineDT),
timelineDT.getKind(),
new Text("0.0.0.0:8188")));
assertFail();
} catch (RuntimeException ce) {
assertException(clientFake, ce);
}
} finally { } finally {
client.stop(); client.stop();
clientFake.stop();
dtManager.stopThreads(); dtManager.stopThreads();
} }
} }
@ -393,6 +413,27 @@ protected TimelineWriter createTimelineWriter(Configuration conf,
return client; return client;
} }
private TimelineClientImpl createTimelineClientFakeTimelineClientRetryOp(
YarnConfiguration conf) {
TimelineClientImpl client = new TimelineClientImpl() {
@Override
public TimelineClientRetryOp
createTimelineClientRetryOpForOperateDelegationToken(
final PrivilegedExceptionAction<?> action) throws IOException {
TimelineClientRetryOpForOperateDelegationToken op =
spy(new TimelineClientRetryOpForOperateDelegationToken(
UserGroupInformation.getCurrentUser(), action));
doThrow(new SocketTimeoutException("Test socketTimeoutException"))
.when(op).run();
return op;
}
};
client.init(conf);
client.start();
return client;
}
private static class TestTimlineDelegationTokenSecretManager extends private static class TestTimlineDelegationTokenSecretManager extends
AbstractDelegationTokenSecretManager<TimelineDelegationTokenIdentifier> { AbstractDelegationTokenSecretManager<TimelineDelegationTokenIdentifier> {