YARN-5309. Fix SSLFactory truststore reloader thread leak in TimelineClientImpl. Contributed by Weiwei Yang.
This commit is contained in:
parent
f2508d78b2
commit
478a2bb185
|
@ -144,6 +144,11 @@
|
||||||
<artifactId>junit</artifactId>
|
<artifactId>junit</artifactId>
|
||||||
<scope>test</scope>
|
<scope>test</scope>
|
||||||
</dependency>
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.bouncycastle</groupId>
|
||||||
|
<artifactId>bcprov-jdk16</artifactId>
|
||||||
|
<scope>test</scope>
|
||||||
|
</dependency>
|
||||||
<dependency>
|
<dependency>
|
||||||
<groupId>com.sun.jersey.jersey-test-framework</groupId>
|
<groupId>com.sun.jersey.jersey-test-framework</groupId>
|
||||||
<artifactId>jersey-test-framework-grizzly2</artifactId>
|
<artifactId>jersey-test-framework-grizzly2</artifactId>
|
||||||
|
|
|
@ -111,6 +111,7 @@ public class TimelineClientImpl extends TimelineClient {
|
||||||
private Configuration configuration;
|
private Configuration configuration;
|
||||||
private float timelineServiceVersion;
|
private float timelineServiceVersion;
|
||||||
private TimelineWriter timelineWriter;
|
private TimelineWriter timelineWriter;
|
||||||
|
private SSLFactory sslFactory;
|
||||||
|
|
||||||
@Private
|
@Private
|
||||||
@VisibleForTesting
|
@VisibleForTesting
|
||||||
|
@ -269,7 +270,7 @@ protected void serviceInit(Configuration conf) throws Exception {
|
||||||
}
|
}
|
||||||
ClientConfig cc = new DefaultClientConfig();
|
ClientConfig cc = new DefaultClientConfig();
|
||||||
cc.getClasses().add(YarnJacksonJaxbJsonProvider.class);
|
cc.getClasses().add(YarnJacksonJaxbJsonProvider.class);
|
||||||
connConfigurator = newConnConfigurator(conf);
|
connConfigurator = initConnConfigurator(conf);
|
||||||
if (UserGroupInformation.isSecurityEnabled()) {
|
if (UserGroupInformation.isSecurityEnabled()) {
|
||||||
authenticator = new KerberosDelegationTokenAuthenticator();
|
authenticator = new KerberosDelegationTokenAuthenticator();
|
||||||
} else {
|
} else {
|
||||||
|
@ -325,6 +326,9 @@ protected void serviceStop() throws Exception {
|
||||||
if (this.timelineWriter != null) {
|
if (this.timelineWriter != null) {
|
||||||
this.timelineWriter.close();
|
this.timelineWriter.close();
|
||||||
}
|
}
|
||||||
|
if (this.sslFactory != null) {
|
||||||
|
this.sslFactory.destroy();
|
||||||
|
}
|
||||||
super.serviceStop();
|
super.serviceStop();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -477,9 +481,9 @@ public HttpURLConnection getHttpURLConnection(final URL url) throws IOException
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private static ConnectionConfigurator newConnConfigurator(Configuration conf) {
|
private ConnectionConfigurator initConnConfigurator(Configuration conf) {
|
||||||
try {
|
try {
|
||||||
return newSslConnConfigurator(DEFAULT_SOCKET_TIMEOUT, conf);
|
return initSslConnConfigurator(DEFAULT_SOCKET_TIMEOUT, conf);
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
LOG.debug("Cannot load customized ssl related configuration. " +
|
LOG.debug("Cannot load customized ssl related configuration. " +
|
||||||
"Fallback to system-generic settings.", e);
|
"Fallback to system-generic settings.", e);
|
||||||
|
@ -497,16 +501,15 @@ public HttpURLConnection configure(HttpURLConnection conn)
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
private static ConnectionConfigurator newSslConnConfigurator(final int timeout,
|
private ConnectionConfigurator initSslConnConfigurator(final int timeout,
|
||||||
Configuration conf) throws IOException, GeneralSecurityException {
|
Configuration conf) throws IOException, GeneralSecurityException {
|
||||||
final SSLFactory factory;
|
|
||||||
final SSLSocketFactory sf;
|
final SSLSocketFactory sf;
|
||||||
final HostnameVerifier hv;
|
final HostnameVerifier hv;
|
||||||
|
|
||||||
factory = new SSLFactory(SSLFactory.Mode.CLIENT, conf);
|
sslFactory = new SSLFactory(SSLFactory.Mode.CLIENT, conf);
|
||||||
factory.init();
|
sslFactory.init();
|
||||||
sf = factory.createSSLSocketFactory();
|
sf = sslFactory.createSSLSocketFactory();
|
||||||
hv = factory.getHostnameVerifier();
|
hv = sslFactory.getHostnameVerifier();
|
||||||
|
|
||||||
return new ConnectionConfigurator() {
|
return new ConnectionConfigurator() {
|
||||||
@Override
|
@Override
|
||||||
|
|
|
@ -25,6 +25,7 @@
|
||||||
import static org.mockito.Mockito.spy;
|
import static org.mockito.Mockito.spy;
|
||||||
import static org.mockito.Mockito.when;
|
import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
|
import java.io.File;
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
import java.net.ConnectException;
|
import java.net.ConnectException;
|
||||||
import java.net.SocketTimeoutException;
|
import java.net.SocketTimeoutException;
|
||||||
|
@ -33,8 +34,10 @@
|
||||||
|
|
||||||
import org.apache.hadoop.conf.Configuration;
|
import org.apache.hadoop.conf.Configuration;
|
||||||
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
|
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
|
||||||
|
import org.apache.hadoop.fs.FileUtil;
|
||||||
import org.apache.hadoop.io.Text;
|
import org.apache.hadoop.io.Text;
|
||||||
import org.apache.hadoop.security.UserGroupInformation;
|
import org.apache.hadoop.security.UserGroupInformation;
|
||||||
|
import org.apache.hadoop.security.ssl.KeyStoreTestUtil;
|
||||||
import org.apache.hadoop.security.token.Token;
|
import org.apache.hadoop.security.token.Token;
|
||||||
import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenSecretManager;
|
import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenSecretManager;
|
||||||
import org.apache.hadoop.yarn.api.records.timeline.TimelineDomain;
|
import org.apache.hadoop.yarn.api.records.timeline.TimelineDomain;
|
||||||
|
@ -58,9 +61,15 @@ public class TestTimelineClient {
|
||||||
|
|
||||||
private TimelineClientImpl client;
|
private TimelineClientImpl client;
|
||||||
private TimelineWriter spyTimelineWriter;
|
private TimelineWriter spyTimelineWriter;
|
||||||
|
private static final File testDir = new File(
|
||||||
|
System.getProperty("test.build.data",
|
||||||
|
System.getProperty("java.io.tmpdir")),
|
||||||
|
"TestTimelineClient");
|
||||||
|
|
||||||
@Before
|
@Before
|
||||||
public void setup() {
|
public void setup() {
|
||||||
|
FileUtil.fullyDelete(testDir);
|
||||||
|
testDir.mkdirs();
|
||||||
YarnConfiguration conf = new YarnConfiguration();
|
YarnConfiguration conf = new YarnConfiguration();
|
||||||
conf.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, true);
|
conf.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, true);
|
||||||
conf.setFloat(YarnConfiguration.TIMELINE_SERVICE_VERSION, 1.0f);
|
conf.setFloat(YarnConfiguration.TIMELINE_SERVICE_VERSION, 1.0f);
|
||||||
|
@ -69,6 +78,7 @@ public void setup() {
|
||||||
|
|
||||||
@After
|
@After
|
||||||
public void tearDown() {
|
public void tearDown() {
|
||||||
|
FileUtil.fullyDelete(testDir);
|
||||||
if (client != null) {
|
if (client != null) {
|
||||||
client.stop();
|
client.stop();
|
||||||
}
|
}
|
||||||
|
@ -434,6 +444,49 @@ private TimelineClientImpl createTimelineClientFakeTimelineClientRetryOp(
|
||||||
return client;
|
return client;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testTimelineClientCleanup() throws Exception {
|
||||||
|
YarnConfiguration conf = new YarnConfiguration();
|
||||||
|
conf.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, true);
|
||||||
|
conf.setInt(YarnConfiguration.TIMELINE_SERVICE_CLIENT_MAX_RETRIES, 0);
|
||||||
|
|
||||||
|
String sslConfDir =
|
||||||
|
KeyStoreTestUtil.getClasspathDir(TestTimelineClient.class);
|
||||||
|
KeyStoreTestUtil.setupSSLConfig(testDir.getAbsolutePath(),
|
||||||
|
sslConfDir, conf, false);
|
||||||
|
client = createTimelineClient(conf);
|
||||||
|
|
||||||
|
ThreadGroup threadGroup = Thread.currentThread().getThreadGroup();
|
||||||
|
|
||||||
|
while (threadGroup.getParent() != null) {
|
||||||
|
threadGroup = threadGroup.getParent();
|
||||||
|
}
|
||||||
|
|
||||||
|
Thread[] threads = new Thread[threadGroup.activeCount()];
|
||||||
|
|
||||||
|
threadGroup.enumerate(threads);
|
||||||
|
Thread reloaderThread = null;
|
||||||
|
for (Thread thread : threads) {
|
||||||
|
if ((thread.getName() != null)
|
||||||
|
&& (thread.getName().contains("Truststore reloader thread"))) {
|
||||||
|
reloaderThread = thread;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Assert.assertTrue("Reloader is not alive", reloaderThread.isAlive());
|
||||||
|
|
||||||
|
client.close();
|
||||||
|
|
||||||
|
boolean reloaderStillAlive = true;
|
||||||
|
for (int i = 0; i < 10; i++) {
|
||||||
|
reloaderStillAlive = reloaderThread.isAlive();
|
||||||
|
if (!reloaderStillAlive) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
Thread.sleep(1000);
|
||||||
|
}
|
||||||
|
Assert.assertFalse("Reloader is still alive", reloaderStillAlive);
|
||||||
|
}
|
||||||
|
|
||||||
private static class TestTimlineDelegationTokenSecretManager extends
|
private static class TestTimlineDelegationTokenSecretManager extends
|
||||||
AbstractDelegationTokenSecretManager<TimelineDelegationTokenIdentifier> {
|
AbstractDelegationTokenSecretManager<TimelineDelegationTokenIdentifier> {
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue