Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HDFS-5489. Merge change r1542158 from trunk. 

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1542159 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jing Zhao 2013-11-15 02:18:33 +00:00
parent eba4c42d6a
commit 6f0a0e0acd
6 changed files with 157 additions and 310 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
View File

@ -119,6 +119,8 @@ Release 2.3.0 - UNRELEASED
HDFS-5506. Use URLConnectionFactory in DelegationTokenFetcher. (Haohui Mai
via jing9)
HDFS-5489. Use TokenAspect in WebHDFSFileSystem. (Haohui Mai via jing9)
OPTIMIZATIONS
HDFS-5239. Allow FSNamesystem lock fairness to be configurable (daryn)

4
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/web/TokenAspect.java
View File

@ -144,6 +144,10 @@ final class TokenAspect<T extends FileSystem & Renewable> {
}
}
public synchronized void reset() {
hasInitedToken = false;
}
synchronized void initDelegationToken(UserGroupInformation ugi) {
Token<?> token = selectDelegationToken(ugi);
if (token != null) {

75
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/web/WebHdfsFileSystem.java
View File

@ -119,38 +119,11 @@ public class WebHdfsFileSystem extends FileSystem
/** Delegation token kind */
public static final Text TOKEN_KIND = new Text("WEBHDFS delegation");
/** Token selector */
public static final DTSelecorByKind DT_SELECTOR
= new DTSelecorByKind(TOKEN_KIND);
private DelegationTokenRenewer dtRenewer = null;
@VisibleForTesting
DelegationTokenRenewer.RenewAction<?> action;
@Override
public URI getCanonicalUri() {
return super.getCanonicalUri();
}
@VisibleForTesting
protected synchronized void addRenewAction(final WebHdfsFileSystem webhdfs) {
if (dtRenewer == null) {
dtRenewer = DelegationTokenRenewer.getInstance();
}
action = dtRenewer.addRenewAction(webhdfs);
}
/** Is WebHDFS enabled in conf? */
public static boolean isEnabled(final Configuration conf, final Log log) {
final boolean b = conf.getBoolean(DFSConfigKeys.DFS_WEBHDFS_ENABLED_KEY,
DFSConfigKeys.DFS_WEBHDFS_ENABLED_DEFAULT);
return b;
}
protected TokenAspect<WebHdfsFileSystem> tokenAspect = new TokenAspect<WebHdfsFileSystem>(
this, TOKEN_KIND);
private UserGroupInformation ugi;
private URI uri;
private boolean hasInitedToken;
private Token<?> delegationToken;
private RetryPolicy retryPolicy = null;
private Path workingDir;
@ -213,41 +186,27 @@ public class WebHdfsFileSystem extends FileSystem
this.workingDir = getHomeDirectory();
if (UserGroupInformation.isSecurityEnabled()) {
initDelegationToken();
tokenAspect.initDelegationToken(ugi);
}
}
protected void initDelegationToken() throws IOException {
// look for webhdfs token, then try hdfs
Token<?> token = selectDelegationToken(ugi);
if (token != null) {
LOG.debug("Found existing DT for " + token.getService());
setDelegationToken(token);
hasInitedToken = true;
}
@Override
public URI getCanonicalUri() {
return super.getCanonicalUri();
}
/** Is WebHDFS enabled in conf? */
public static boolean isEnabled(final Configuration conf, final Log log) {
final boolean b = conf.getBoolean(DFSConfigKeys.DFS_WEBHDFS_ENABLED_KEY,
DFSConfigKeys.DFS_WEBHDFS_ENABLED_DEFAULT);
return b;
}
protected synchronized Token<?> getDelegationToken() throws IOException {
// we haven't inited yet, or we used to have a token but it expired
if (!hasInitedToken || (action != null && !action.isValid())) {
//since we don't already have a token, go get one
Token<?> token = getDelegationToken(null);
// security might be disabled
if (token != null) {
setDelegationToken(token);
addRenewAction(this);
LOG.debug("Created new DT for " + token.getService());
}
hasInitedToken = true;
}
tokenAspect.ensureTokenInitialized();
return delegationToken;
}
protected Token<DelegationTokenIdentifier> selectDelegationToken(
UserGroupInformation ugi) {
return DT_SELECTOR.selectToken(getCanonicalUri(), ugi.getTokens(), getConf());
}
@Override
protected int getDefaultPort() {
return getConf().getInt(DFSConfigKeys.DFS_NAMENODE_HTTP_PORT_KEY,
@ -371,7 +330,7 @@ public class WebHdfsFileSystem extends FileSystem
private synchronized void resetStateToFailOver() {
currentNNAddrIndex = (currentNNAddrIndex + 1) % nnAddrs.length;
delegationToken = null;
hasInitedToken = false;
tokenAspect.reset();
}
/**
@ -882,9 +841,7 @@ public class WebHdfsFileSystem extends FileSystem
@Override
public void close() throws IOException {
super.close();
if (dtRenewer != null) {
dtRenewer.removeRenewAction(this); // blocks
}
tokenAspect.removeRenewAction();
}
class OffsetUrlOpener extends ByteRangeInputStream.URLOpener {

89
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/web/TestTokenAspect.java
View File

@ -19,13 +19,19 @@
package org.apache.hadoop.hdfs.web;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.io.FileNotFoundException;
@ -35,6 +41,7 @@ import java.net.URISyntaxException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.DelegationTokenRenewer;
import org.apache.hadoop.fs.DelegationTokenRenewer.RenewAction;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
@ -163,15 +170,44 @@ public class TestTokenAspect {
}
}
private static RenewAction<?> getActionFromTokenAspect(
TokenAspect<DummyFs> tokenAspect) {
return (RenewAction<?>) Whitebox.getInternalState(tokenAspect, "action");
}
@Test
public void testGetRemoteToken() throws IOException, URISyntaxException {
public void testCachedInitialization() throws IOException, URISyntaxException {
Configuration conf = new Configuration();
UserGroupInformation.setConfiguration(conf);
DummyFs fs = spy(new DummyFs());
Token<TokenIdentifier> token = new Token<TokenIdentifier>(new byte[0],
new byte[0], DummyFs.TOKEN_KIND, new Text("127.0.0.1:1234"));
doReturn(token).when(fs).getDelegationToken(anyString());
doReturn(token).when(fs).getRenewToken();
fs.emulateSecurityEnabled = true;
fs.initialize(new URI("dummyfs://127.0.0.1:1234"), conf);
fs.tokenAspect.ensureTokenInitialized();
verify(fs, times(1)).getDelegationToken(null);
verify(fs, times(1)).setDelegationToken(token);
// For the second iteration, the token should be cached.
fs.tokenAspect.ensureTokenInitialized();
verify(fs, times(1)).getDelegationToken(null);
verify(fs, times(1)).setDelegationToken(token);
}
@Test
public void testGetRemoteToken() throws IOException, URISyntaxException {
Configuration conf = new Configuration();
DummyFs fs = spy(new DummyFs());
Token<TokenIdentifier> token = new Token<TokenIdentifier>(new byte[0],
new byte[0], DummyFs.TOKEN_KIND, new Text("127.0.0.1:1234"));
doReturn(token).when(fs).getDelegationToken(anyString());
doReturn(token).when(fs).getRenewToken();
fs.initialize(new URI("dummyfs://127.0.0.1:1234"), conf);
fs.tokenAspect.ensureTokenInitialized();
@ -186,7 +222,6 @@ public class TestTokenAspect {
public void testGetRemoteTokenFailure() throws IOException,
URISyntaxException {
Configuration conf = new Configuration();
UserGroupInformation.setConfiguration(conf);
DummyFs fs = spy(new DummyFs());
IOException e = new IOException();
doThrow(e).when(fs).getDelegationToken(anyString());
@ -203,7 +238,6 @@ public class TestTokenAspect {
@Test
public void testInitWithNoTokens() throws IOException, URISyntaxException {
Configuration conf = new Configuration();
UserGroupInformation.setConfiguration(conf);
DummyFs fs = spy(new DummyFs());
doReturn(null).when(fs).getDelegationToken(anyString());
fs.initialize(new URI("dummyfs://127.0.0.1:1234"), conf);
@ -218,7 +252,6 @@ public class TestTokenAspect {
@Test
public void testInitWithUGIToken() throws IOException, URISyntaxException {
Configuration conf = new Configuration();
UserGroupInformation.setConfiguration(conf);
DummyFs fs = spy(new DummyFs());
doReturn(null).when(fs).getDelegationToken(anyString());
@ -241,6 +274,51 @@ public class TestTokenAspect {
assertNull(Whitebox.getInternalState(fs.tokenAspect, "action"));
}
@Test
public void testRenewal() throws Exception {
Configuration conf = new Configuration();
Token<?> token1 = mock(Token.class);
Token<?> token2 = mock(Token.class);
final long renewCycle = 100;
DelegationTokenRenewer.renewCycle = renewCycle;
UserGroupInformation ugi = UserGroupInformation.createUserForTesting("foo",
new String[] { "bar" });
DummyFs fs = spy(new DummyFs());
doReturn(token1).doReturn(token2).when(fs).getDelegationToken(null);
doReturn(token1).when(fs).getRenewToken();
// cause token renewer to abandon the token
doThrow(new IOException("renew failed")).when(token1).renew(conf);
doThrow(new IOException("get failed")).when(fs).addDelegationTokens(null,
null);
TokenAspect<DummyFs> tokenAspect = new TokenAspect<DummyFs>(fs,
DummyFs.TOKEN_KIND);
fs.initialize(new URI("dummyfs://127.0.0.1:1234"), conf);
tokenAspect.initDelegationToken(ugi);
// trigger token acquisition
tokenAspect.ensureTokenInitialized();
DelegationTokenRenewer.RenewAction<?> action = getActionFromTokenAspect(tokenAspect);
verify(fs).setDelegationToken(token1);
assertTrue(action.isValid());
// upon renewal, token will go bad based on above stubbing
Thread.sleep(renewCycle * 2);
assertSame(action, getActionFromTokenAspect(tokenAspect));
assertFalse(action.isValid());
// now that token is invalid, should get a new one
tokenAspect.ensureTokenInitialized();
verify(fs, times(2)).getDelegationToken(anyString());
verify(fs).setDelegationToken(token2);
assertNotSame(action, getActionFromTokenAspect(tokenAspect));
action = getActionFromTokenAspect(tokenAspect);
assertTrue(action.isValid());
}
@Test
public void testTokenSelectionPreferences() throws IOException,
URISyntaxException {
@ -252,7 +330,6 @@ public class TestTokenAspect {
DummyFs.TOKEN_KIND);
UserGroupInformation ugi = UserGroupInformation.createUserForTesting("foo",
new String[] { "bar" });
UserGroupInformation.setConfiguration(conf);
// use ip-based tokens
SecurityUtilTestHelper.setTokenServiceUseIp(true);

167
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/web/TestWebHdfsTokens.java
View File

@ -19,16 +19,20 @@
package org.apache.hadoop.hdfs.web;
import static org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod.KERBEROS;
import static org.junit.Assert.*;
import static org.mockito.Matchers.*;
import static org.mockito.Mockito.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import java.io.IOException;
import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.DelegationTokenRenewer;
import org.apache.hadoop.fs.DelegationTokenRenewer.RenewAction;
import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
import org.apache.hadoop.hdfs.web.resources.DeleteOpParam;
import org.apache.hadoop.hdfs.web.resources.GetOpParam;
@ -40,96 +44,38 @@ import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.internal.util.reflection.Whitebox;
public class TestWebHdfsTokens {
static Configuration conf;
static UserGroupInformation ugi;
private static Configuration conf;
@BeforeClass
public static void setup() throws IOException {
public static void setUp() {
conf = new Configuration();
SecurityUtil.setAuthenticationMethod(KERBEROS, conf);
UserGroupInformation.setConfiguration(conf);
ugi = UserGroupInformation.getCurrentUser();
}
private WebHdfsFileSystem spyWebhdfsInSecureSetup() throws IOException {
WebHdfsFileSystem fsOrig = new WebHdfsFileSystem();
fsOrig.initialize(URI.create("webhdfs://127.0.0.1:0"), conf);
WebHdfsFileSystem fs = spy(fsOrig);
Whitebox.setInternalState(fsOrig.tokenAspect, "fs", fs);
return fs;
}
@SuppressWarnings("unchecked")
@Test(timeout=1000)
public void testInitWithNoToken() throws IOException {
WebHdfsFileSystem fs = spy(new WebHdfsFileSystem());
doReturn(null).when(fs).getDelegationToken(anyString());
doNothing().when(fs).addRenewAction(any(WebHdfsFileSystem.class));
fs.initialize(URI.create("webhdfs://127.0.0.1:0"), conf);
// when not in ugi, don't get one
verify(fs).initDelegationToken();
verify(fs).selectDelegationToken(ugi);
verify(fs, never()).setDelegationToken(any(Token.class));
verify(fs, never()).getDelegationToken();
verify(fs, never()).getDelegationToken(anyString());
}
@SuppressWarnings("unchecked")
@Test(timeout=1000)
public void testInitWithUGIToken() throws IOException {
WebHdfsFileSystem fs = spy(new WebHdfsFileSystem());
Token<DelegationTokenIdentifier> token = mock(Token.class);
doReturn(token).when(fs).selectDelegationToken(ugi);
doReturn(null).when(fs).getDelegationToken(anyString());
doNothing().when(fs).addRenewAction(any(WebHdfsFileSystem.class));
fs.initialize(URI.create("webhdfs://127.0.0.1:0"), conf);
// when in the ugi, store it but don't renew it
verify(fs).initDelegationToken();
verify(fs).selectDelegationToken(ugi);
verify(fs).setDelegationToken(token);
verify(fs, never()).getDelegationToken();
verify(fs, never()).getDelegationToken(anyString());
verify(fs, never()).addRenewAction(fs);
}
@SuppressWarnings("unchecked")
@Test(timeout=1000)
public void testInternalGetDelegationToken() throws IOException {
WebHdfsFileSystem fs = spy(new WebHdfsFileSystem());
Token<DelegationTokenIdentifier> token = mock(Token.class);
doReturn(null).when(fs).selectDelegationToken(ugi);
doReturn(token).when(fs).getDelegationToken(anyString());
doNothing().when(fs).addRenewAction(any(WebHdfsFileSystem.class));
fs.initialize(URI.create("webhdfs://127.0.0.1:0"), conf);
// get token, store it, and renew it
Token<?> token2 = fs.getDelegationToken();
assertEquals(token2, token);
verify(fs).getDelegationToken(null);
verify(fs).setDelegationToken(token);
verify(fs).addRenewAction(fs);
reset(fs);
// just return token, don't get/set/renew
token2 = fs.getDelegationToken();
assertEquals(token2, token);
verify(fs, never()).getDelegationToken(null);
verify(fs, never()).setDelegationToken(any(Token.class));
verify(fs, never()).addRenewAction(fs);
}
@SuppressWarnings("unchecked")
@Test(timeout=1000)
@Test(timeout = 1000)
public void testTokenForNonTokenOp() throws IOException {
WebHdfsFileSystem fs = spy(new WebHdfsFileSystem());
WebHdfsFileSystem fs = spyWebhdfsInSecureSetup();
Token<DelegationTokenIdentifier> token = mock(Token.class);
doReturn(null).when(fs).selectDelegationToken(ugi);
doReturn(token).when(fs).getDelegationToken(null);
doNothing().when(fs).addRenewAction(any(WebHdfsFileSystem.class));
fs.initialize(URI.create("webhdfs://127.0.0.1:0"), conf);
// should get/set/renew token
fs.toUrl(GetOpParam.Op.OPEN, null);
verify(fs).getDelegationToken();
verify(fs).getDelegationToken(null);
verify(fs).setDelegationToken(token);
verify(fs).addRenewAction(fs);
reset(fs);
// should return prior token
@ -137,30 +83,27 @@ public class TestWebHdfsTokens {
verify(fs).getDelegationToken();
verify(fs, never()).getDelegationToken(null);
verify(fs, never()).setDelegationToken(token);
verify(fs, never()).addRenewAction(fs);
}
@Test(timeout=1000)
@Test(timeout = 1000)
public void testNoTokenForGetToken() throws IOException {
checkNoTokenForOperation(GetOpParam.Op.GETDELEGATIONTOKEN);
}
@Test(timeout=1000)
@Test(timeout = 1000)
public void testNoTokenForCanclToken() throws IOException {
checkNoTokenForOperation(PutOpParam.Op.RENEWDELEGATIONTOKEN);
}
@Test(timeout=1000)
@Test(timeout = 1000)
public void testNoTokenForCancelToken() throws IOException {
checkNoTokenForOperation(PutOpParam.Op.CANCELDELEGATIONTOKEN);
}
@SuppressWarnings("unchecked")
private void checkNoTokenForOperation(HttpOpParam.Op op) throws IOException {
WebHdfsFileSystem fs = spy(new WebHdfsFileSystem());
doReturn(null).when(fs).selectDelegationToken(ugi);
WebHdfsFileSystem fs = spyWebhdfsInSecureSetup();
doReturn(null).when(fs).getDelegationToken(null);
doNothing().when(fs).addRenewAction(any(WebHdfsFileSystem.class));
fs.initialize(URI.create("webhdfs://127.0.0.1:0"), conf);
// do not get a token!
@ -168,10 +111,9 @@ public class TestWebHdfsTokens {
verify(fs, never()).getDelegationToken();
verify(fs, never()).getDelegationToken(null);
verify(fs, never()).setDelegationToken(any(Token.class));
verify(fs, never()).addRenewAction(fs);
}
@Test(timeout=1000)
@Test(timeout = 1000)
public void testGetOpRequireAuth() {
for (HttpOpParam.Op op : GetOpParam.Op.values()) {
boolean expect = (op == GetOpParam.Op.GETDELEGATIONTOKEN);
@ -179,72 +121,25 @@ public class TestWebHdfsTokens {
}
}
@Test(timeout=1000)
@Test(timeout = 1000)
public void testPutOpRequireAuth() {
for (HttpOpParam.Op op : PutOpParam.Op.values()) {
boolean expect = (op == PutOpParam.Op.RENEWDELEGATIONTOKEN ||
op == PutOpParam.Op.CANCELDELEGATIONTOKEN);
boolean expect = (op == PutOpParam.Op.RENEWDELEGATIONTOKEN || op == PutOpParam.Op.CANCELDELEGATIONTOKEN);
assertEquals(expect, op.getRequireAuth());
}
}
@Test(timeout=1000)
@Test(timeout = 1000)
public void testPostOpRequireAuth() {
for (HttpOpParam.Op op : PostOpParam.Op.values()) {
assertFalse(op.getRequireAuth());
}
}
@Test(timeout=1000)
@Test(timeout = 1000)
public void testDeleteOpRequireAuth() {
for (HttpOpParam.Op op : DeleteOpParam.Op.values()) {
assertFalse(op.getRequireAuth());
}
}
@Test
public void testGetTokenAfterFailure() throws Exception {
Configuration conf = mock(Configuration.class);
Token<?> token1 = mock(Token.class);
Token<?> token2 = mock(Token.class);
long renewCycle = 1000;
DelegationTokenRenewer.renewCycle = renewCycle;
WebHdfsFileSystem fs = spy(new WebHdfsFileSystem());
doReturn(conf).when(fs).getConf();
doReturn(token1).doReturn(token2).when(fs).getDelegationToken(null);
// cause token renewer to abandon the token
doThrow(new IOException("renew failed")).when(token1).renew(conf);
doThrow(new IOException("get failed")).when(fs).addDelegationTokens(null, null);
// trigger token acquisition
Token<?> token = fs.getDelegationToken();
RenewAction<?> action = fs.action;
assertSame(token1, token);
assertTrue(action.isValid());
// fetch again and make sure it's the same as before
token = fs.getDelegationToken();
assertSame(token1, token);
assertSame(action, fs.action);
assertTrue(fs.action.isValid());
// upon renewal, token will go bad based on above stubbing
Thread.sleep(renewCycle);
assertSame(action, fs.action);
assertFalse(fs.action.isValid());
// now that token is invalid, should get a new one
token = fs.getDelegationToken();
assertSame(token2, token);
assertNotSame(action, fs.action);
assertTrue(fs.action.isValid());
action = fs.action;
// should get same one again
token = fs.getDelegationToken();
assertSame(token2, token);
assertSame(action, fs.action);
assertTrue(fs.action.isValid());
}
}

106
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/web/TestWebHdfsUrl.java
View File

@ -20,8 +20,6 @@ package org.apache.hadoop.hdfs.web;
import static org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod.KERBEROS;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.mockito.Mockito.mock;
import java.io.IOException;
@ -36,15 +34,20 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenSecretManager;
import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
import org.apache.hadoop.hdfs.web.resources.*;
import org.apache.hadoop.hdfs.web.resources.DelegationParam;
import org.apache.hadoop.hdfs.web.resources.DoAsParam;
import org.apache.hadoop.hdfs.web.resources.GetOpParam;
import org.apache.hadoop.hdfs.web.resources.PutOpParam;
import org.apache.hadoop.hdfs.web.resources.TokenArgumentParam;
import org.apache.hadoop.hdfs.web.resources.UserParam;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.security.SecurityUtilTestHelper;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.security.token.TokenIdentifier;
import org.junit.*;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
public class TestWebHdfsUrl {
// NOTE: port is never used
@ -306,95 +309,4 @@ public class TestWebHdfsUrl {
}
return (WebHdfsFileSystem) FileSystem.get(uri, conf);
}
@Test(timeout=60000)
public void testSelectHdfsDelegationToken() throws Exception {
SecurityUtilTestHelper.setTokenServiceUseIp(true);
Configuration conf = new Configuration();
conf.setClass("fs.webhdfs.impl", MyWebHdfsFileSystem.class, FileSystem.class);
// test with implicit default port
URI fsUri = URI.create("webhdfs://localhost");
MyWebHdfsFileSystem fs = (MyWebHdfsFileSystem) FileSystem.get(fsUri, conf);
checkTokenSelection(fs, conf);
// test with explicit default port
fsUri = URI.create("webhdfs://localhost:"+fs.getDefaultPort());
fs = (MyWebHdfsFileSystem) FileSystem.get(fsUri, conf);
checkTokenSelection(fs, conf);
// test with non-default port
fsUri = URI.create("webhdfs://localhost:"+(fs.getDefaultPort()-1));
fs = (MyWebHdfsFileSystem) FileSystem.get(fsUri, conf);
checkTokenSelection(fs, conf);
}
private void checkTokenSelection(MyWebHdfsFileSystem fs,
Configuration conf) throws IOException {
int port = fs.getCanonicalUri().getPort();
// can't clear tokens from ugi, so create a new user everytime
UserGroupInformation ugi =
UserGroupInformation.createUserForTesting(fs.getUri().getAuthority(), new String[]{});
// use ip-based tokens
SecurityUtilTestHelper.setTokenServiceUseIp(true);
// test fallback to hdfs token
Token<?> hdfsToken = new Token<TokenIdentifier>(
new byte[0], new byte[0],
DelegationTokenIdentifier.HDFS_DELEGATION_KIND,
new Text("127.0.0.1:8020"));
ugi.addToken(hdfsToken);
// test fallback to hdfs token
Token<?> token = fs.selectDelegationToken(ugi);
assertNotNull(token);
assertEquals(hdfsToken, token);
// test webhdfs is favored over hdfs
Token<?> webHdfsToken = new Token<TokenIdentifier>(
new byte[0], new byte[0],
WebHdfsFileSystem.TOKEN_KIND, new Text("127.0.0.1:"+port));
ugi.addToken(webHdfsToken);
token = fs.selectDelegationToken(ugi);
assertNotNull(token);
assertEquals(webHdfsToken, token);
// switch to using host-based tokens, no token should match
SecurityUtilTestHelper.setTokenServiceUseIp(false);
token = fs.selectDelegationToken(ugi);
assertNull(token);
// test fallback to hdfs token
hdfsToken = new Token<TokenIdentifier>(
new byte[0], new byte[0],
DelegationTokenIdentifier.HDFS_DELEGATION_KIND,
new Text("localhost:8020"));
ugi.addToken(hdfsToken);
token = fs.selectDelegationToken(ugi);
assertNotNull(token);
assertEquals(hdfsToken, token);
// test webhdfs is favored over hdfs
webHdfsToken = new Token<TokenIdentifier>(
new byte[0], new byte[0],
WebHdfsFileSystem.TOKEN_KIND, new Text("localhost:"+port));
ugi.addToken(webHdfsToken);
token = fs.selectDelegationToken(ugi);
assertNotNull(token);
assertEquals(webHdfsToken, token);
}
static class MyWebHdfsFileSystem extends WebHdfsFileSystem {
@Override
public URI getCanonicalUri() {
return super.getCanonicalUri();
}
@Override
public int getDefaultPort() {
return super.getDefaultPort();
}
}
}