<reponame>dmgerman/hadoop
begin_unit|revision:0.9.5;language:Java;cregit-version:0.0.1
begin_comment
comment|/**  * Licensed to the Apache Software Foundation (ASF) under one  * or more contributor license agreements.  See the NOTICE file  * distributed with this work for additional information  * regarding copyright ownership.  The ASF licenses this file  * to you under the Apache License, Version 2.0 (the  * "License"); you may not use this file except in compliance  * with the License.  You may obtain a copy of the License at  *  *     http://www.apache.org/licenses/LICENSE-2.0  *  * Unless required by applicable law or agreed to in writing, software  * distributed under the License is distributed on an "AS IS" BASIS,  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  * See the License for the specific language governing permissions and  * limitations under the License.  */
end_comment

begin_package
DECL|package|org.apache.hadoop.security.token.delegation
package|package
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|security
operator|.
name|token
operator|.
name|delegation
package|;
end_package

begin_import
import|import
name|java
operator|.
name|io
operator|.
name|IOException
import|;
end_import

begin_import
import|import
name|java
operator|.
name|util
operator|.
name|ArrayList
import|;
end_import

begin_import
import|import
name|java
operator|.
name|util
operator|.
name|List
import|;
end_import

begin_import
import|import
name|java
operator|.
name|util
operator|.
name|concurrent
operator|.
name|Callable
import|;
end_import

begin_import
import|import
name|java
operator|.
name|util
operator|.
name|concurrent
operator|.
name|ExecutorService
import|;
end_import

begin_import
import|import
name|com
operator|.
name|google
operator|.
name|common
operator|.
name|base
operator|.
name|Supplier
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|curator
operator|.
name|RetryPolicy
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|curator
operator|.
name|framework
operator|.
name|CuratorFramework
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|curator
operator|.
name|framework
operator|.
name|CuratorFrameworkFactory
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|curator
operator|.
name|framework
operator|.
name|api
operator|.
name|ACLProvider
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|curator
operator|.
name|retry
operator|.
name|ExponentialBackoffRetry
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|curator
operator|.
name|test
operator|.
name|TestingServer
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|conf
operator|.
name|Configuration
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|io
operator|.
name|Text
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|security
operator|.
name|UserGroupInformation
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|security
operator|.
name|token
operator|.
name|SecretManager
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|security
operator|.
name|token
operator|.
name|Token
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|security
operator|.
name|token
operator|.
name|delegation
operator|.
name|web
operator|.
name|DelegationTokenIdentifier
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|security
operator|.
name|token
operator|.
name|delegation
operator|.
name|web
operator|.
name|DelegationTokenManager
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|test
operator|.
name|GenericTestUtils
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|zookeeper
operator|.
name|ZooDefs
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|zookeeper
operator|.
name|data
operator|.
name|ACL
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|zookeeper
operator|.
name|data
operator|.
name|Id
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|zookeeper
operator|.
name|server
operator|.
name|auth
operator|.
name|DigestAuthenticationProvider
import|;
end_import

begin_import
import|import
name|org
operator|.
name|junit
operator|.
name|After
import|;
end_import

begin_import
import|import
name|org
operator|.
name|junit
operator|.
name|Assert
import|;
end_import

begin_import
import|import
name|org
operator|.
name|junit
operator|.
name|Before
import|;
end_import

begin_import
import|import
name|org
operator|.
name|junit
operator|.
name|Rule
import|;
end_import

begin_import
import|import
name|org
operator|.
name|junit
operator|.
name|Test
import|;
end_import

begin_import
import|import
name|org
operator|.
name|junit
operator|.
name|rules
operator|.
name|Timeout
import|;
end_import

begin_import
import|import
name|org
operator|.
name|slf4j
operator|.
name|Logger
import|;
end_import

begin_import
import|import
name|org
operator|.
name|slf4j
operator|.
name|LoggerFactory
import|;
end_import

begin_import
import|import static
name|org
operator|.
name|junit
operator|.
name|Assert
operator|.
name|fail
import|;
end_import

begin_class
DECL|class|TestZKDelegationTokenSecretManager
specifier|public
class|class
name|TestZKDelegationTokenSecretManager
block|{
DECL|field|LOG
specifier|private
specifier|static
specifier|final
name|Logger
name|LOG
init|=
name|LoggerFactory
operator|.
name|getLogger
argument_list|(
name|TestZKDelegationTokenSecretManager
operator|.
name|class
argument_list|)
decl_stmt|;
DECL|field|TEST_RETRIES
specifier|private
specifier|static
specifier|final
name|int
name|TEST_RETRIES
init|=
literal|2
decl_stmt|;
DECL|field|RETRY_COUNT
specifier|private
specifier|static
specifier|final
name|int
name|RETRY_COUNT
init|=
literal|5
decl_stmt|;
DECL|field|RETRY_WAIT
specifier|private
specifier|static
specifier|final
name|int
name|RETRY_WAIT
init|=
literal|1000
decl_stmt|;
DECL|field|DAY_IN_SECS
specifier|private
specifier|static
specifier|final
name|long
name|DAY_IN_SECS
init|=
literal|86400
decl_stmt|;
DECL|field|zkServer
specifier|private
name|TestingServer
name|zkServer
decl_stmt|;
annotation|@
name|Rule
DECL|field|globalTimeout
specifier|public
name|Timeout
name|globalTimeout
init|=
operator|new
name|Timeout
argument_list|(
literal|300000
argument_list|)
decl_stmt|;
annotation|@
name|Before
DECL|method|setup ()
specifier|public
name|void
name|setup
parameter_list|()
throws|throws
name|Exception
block|{
name|zkServer
operator|=
operator|new
name|TestingServer
argument_list|()
expr_stmt|;
name|zkServer
operator|.
name|start
argument_list|()
expr_stmt|;
block|}
annotation|@
name|After
DECL|method|tearDown ()
specifier|public
name|void
name|tearDown
parameter_list|()
throws|throws
name|Exception
block|{
if|if
condition|(
name|zkServer
operator|!=
literal|null
condition|)
block|{
name|zkServer
operator|.
name|close
argument_list|()
expr_stmt|;
block|}
block|}
DECL|method|getSecretConf (String connectString)
specifier|protected
name|Configuration
name|getSecretConf
parameter_list|(
name|String
name|connectString
parameter_list|)
block|{
name|Configuration
name|conf
init|=
operator|new
name|Configuration
argument_list|()
decl_stmt|;
name|conf
operator|.
name|setBoolean
argument_list|(
name|DelegationTokenManager
operator|.
name|ENABLE_ZK_KEY
argument_list|,
literal|true
argument_list|)
expr_stmt|;
name|conf
operator|.
name|set
argument_list|(
name|ZKDelegationTokenSecretManager
operator|.
name|ZK_DTSM_ZK_CONNECTION_STRING
argument_list|,
name|connectString
argument_list|)
expr_stmt|;
name|conf
operator|.
name|set
argument_list|(
name|ZKDelegationTokenSecretManager
operator|.
name|ZK_DTSM_ZNODE_WORKING_PATH
argument_list|,
literal|"testPath"
argument_list|)
expr_stmt|;
name|conf
operator|.
name|set
argument_list|(
name|ZKDelegationTokenSecretManager
operator|.
name|ZK_DTSM_ZK_AUTH_TYPE
argument_list|,
literal|"none"
argument_list|)
expr_stmt|;
name|conf
operator|.
name|setLong
argument_list|(
name|ZKDelegationTokenSecretManager
operator|.
name|ZK_DTSM_ZK_SHUTDOWN_TIMEOUT
argument_list|,
literal|100
argument_list|)
expr_stmt|;
name|conf
operator|.
name|setLong
argument_list|(
name|DelegationTokenManager
operator|.
name|UPDATE_INTERVAL
argument_list|,
name|DAY_IN_SECS
argument_list|)
expr_stmt|;
name|conf
operator|.
name|setLong
argument_list|(
name|DelegationTokenManager
operator|.
name|MAX_LIFETIME
argument_list|,
name|DAY_IN_SECS
argument_list|)
expr_stmt|;
name|conf
operator|.
name|setLong
argument_list|(
name|DelegationTokenManager
operator|.
name|RENEW_INTERVAL
argument_list|,
name|DAY_IN_SECS
argument_list|)
expr_stmt|;
name|conf
operator|.
name|setLong
argument_list|(
name|DelegationTokenManager
operator|.
name|REMOVAL_SCAN_INTERVAL
argument_list|,
name|DAY_IN_SECS
argument_list|)
expr_stmt|;
return|return
name|conf
return|;
block|}
annotation|@
name|SuppressWarnings
argument_list|(
literal|"unchecked"
argument_list|)
annotation|@
name|Test
DECL|method|testMultiNodeOperations ()
specifier|public
name|void
name|testMultiNodeOperations
parameter_list|()
throws|throws
name|Exception
block|{
for|for
control|(
name|int
name|i
init|=
literal|0
init|;
name|i
operator|<
name|TEST_RETRIES
condition|;
name|i
operator|++
control|)
block|{
name|DelegationTokenManager
name|tm1
decl_stmt|,
name|tm2
init|=
literal|null
decl_stmt|;
name|String
name|connectString
init|=
name|zkServer
operator|.
name|getConnectString
argument_list|()
decl_stmt|;
name|Configuration
name|conf
init|=
name|getSecretConf
argument_list|(
name|connectString
argument_list|)
decl_stmt|;
name|tm1
operator|=
operator|new
name|DelegationTokenManager
argument_list|(
name|conf
argument_list|,
operator|new
name|Text
argument_list|(
literal|"bla"
argument_list|)
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|init
argument_list|()
expr_stmt|;
name|tm2
operator|=
operator|new
name|DelegationTokenManager
argument_list|(
name|conf
argument_list|,
operator|new
name|Text
argument_list|(
literal|"bla"
argument_list|)
argument_list|)
expr_stmt|;
name|tm2
operator|.
name|init
argument_list|()
expr_stmt|;
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
name|token
init|=
operator|(
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
operator|)
name|tm1
operator|.
name|createToken
argument_list|(
name|UserGroupInformation
operator|.
name|getCurrentUser
argument_list|()
argument_list|,
literal|"foo"
argument_list|)
decl_stmt|;
name|Assert
operator|.
name|assertNotNull
argument_list|(
name|token
argument_list|)
expr_stmt|;
name|tm2
operator|.
name|verifyToken
argument_list|(
name|token
argument_list|)
expr_stmt|;
name|tm2
operator|.
name|renewToken
argument_list|(
name|token
argument_list|,
literal|"foo"
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|verifyToken
argument_list|(
name|token
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|cancelToken
argument_list|(
name|token
argument_list|,
literal|"foo"
argument_list|)
expr_stmt|;
try|try
block|{
name|verifyTokenFail
argument_list|(
name|tm2
argument_list|,
name|token
argument_list|)
expr_stmt|;
name|fail
argument_list|(
literal|"Expected InvalidToken"
argument_list|)
expr_stmt|;
block|}
catch|catch
parameter_list|(
name|SecretManager
operator|.
name|InvalidToken
name|it
parameter_list|)
block|{
comment|// Ignore
block|}
name|token
operator|=
operator|(
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
operator|)
name|tm2
operator|.
name|createToken
argument_list|(
name|UserGroupInformation
operator|.
name|getCurrentUser
argument_list|()
argument_list|,
literal|"bar"
argument_list|)
expr_stmt|;
name|Assert
operator|.
name|assertNotNull
argument_list|(
name|token
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|verifyToken
argument_list|(
name|token
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|renewToken
argument_list|(
name|token
argument_list|,
literal|"bar"
argument_list|)
expr_stmt|;
name|tm2
operator|.
name|verifyToken
argument_list|(
name|token
argument_list|)
expr_stmt|;
name|tm2
operator|.
name|cancelToken
argument_list|(
name|token
argument_list|,
literal|"bar"
argument_list|)
expr_stmt|;
try|try
block|{
name|verifyTokenFail
argument_list|(
name|tm1
argument_list|,
name|token
argument_list|)
expr_stmt|;
name|fail
argument_list|(
literal|"Expected InvalidToken"
argument_list|)
expr_stmt|;
block|}
catch|catch
parameter_list|(
name|SecretManager
operator|.
name|InvalidToken
name|it
parameter_list|)
block|{
comment|// Ignore
block|}
name|verifyDestroy
argument_list|(
name|tm1
argument_list|,
name|conf
argument_list|)
expr_stmt|;
name|verifyDestroy
argument_list|(
name|tm2
argument_list|,
name|conf
argument_list|)
expr_stmt|;
block|}
block|}
annotation|@
name|SuppressWarnings
argument_list|(
literal|"unchecked"
argument_list|)
annotation|@
name|Test
DECL|method|testNodeUpAferAWhile ()
specifier|public
name|void
name|testNodeUpAferAWhile
parameter_list|()
throws|throws
name|Exception
block|{
for|for
control|(
name|int
name|i
init|=
literal|0
init|;
name|i
operator|<
name|TEST_RETRIES
condition|;
name|i
operator|++
control|)
block|{
name|String
name|connectString
init|=
name|zkServer
operator|.
name|getConnectString
argument_list|()
decl_stmt|;
name|Configuration
name|conf
init|=
name|getSecretConf
argument_list|(
name|connectString
argument_list|)
decl_stmt|;
name|DelegationTokenManager
name|tm1
init|=
operator|new
name|DelegationTokenManager
argument_list|(
name|conf
argument_list|,
operator|new
name|Text
argument_list|(
literal|"bla"
argument_list|)
argument_list|)
decl_stmt|;
name|tm1
operator|.
name|init
argument_list|()
expr_stmt|;
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
name|token1
init|=
operator|(
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
operator|)
name|tm1
operator|.
name|createToken
argument_list|(
name|UserGroupInformation
operator|.
name|getCurrentUser
argument_list|()
argument_list|,
literal|"foo"
argument_list|)
decl_stmt|;
name|Assert
operator|.
name|assertNotNull
argument_list|(
name|token1
argument_list|)
expr_stmt|;
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
name|token2
init|=
operator|(
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
operator|)
name|tm1
operator|.
name|createToken
argument_list|(
name|UserGroupInformation
operator|.
name|getCurrentUser
argument_list|()
argument_list|,
literal|"bar"
argument_list|)
decl_stmt|;
name|Assert
operator|.
name|assertNotNull
argument_list|(
name|token2
argument_list|)
expr_stmt|;
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
name|token3
init|=
operator|(
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
operator|)
name|tm1
operator|.
name|createToken
argument_list|(
name|UserGroupInformation
operator|.
name|getCurrentUser
argument_list|()
argument_list|,
literal|"boo"
argument_list|)
decl_stmt|;
name|Assert
operator|.
name|assertNotNull
argument_list|(
name|token3
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|verifyToken
argument_list|(
name|token1
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|verifyToken
argument_list|(
name|token2
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|verifyToken
argument_list|(
name|token3
argument_list|)
expr_stmt|;
comment|// Cancel one token
name|tm1
operator|.
name|cancelToken
argument_list|(
name|token1
argument_list|,
literal|"foo"
argument_list|)
expr_stmt|;
comment|// Start second node after some time..
name|Thread
operator|.
name|sleep
argument_list|(
literal|1000
argument_list|)
expr_stmt|;
name|DelegationTokenManager
name|tm2
init|=
operator|new
name|DelegationTokenManager
argument_list|(
name|conf
argument_list|,
operator|new
name|Text
argument_list|(
literal|"bla"
argument_list|)
argument_list|)
decl_stmt|;
name|tm2
operator|.
name|init
argument_list|()
expr_stmt|;
name|tm2
operator|.
name|verifyToken
argument_list|(
name|token2
argument_list|)
expr_stmt|;
name|tm2
operator|.
name|verifyToken
argument_list|(
name|token3
argument_list|)
expr_stmt|;
try|try
block|{
name|verifyTokenFail
argument_list|(
name|tm2
argument_list|,
name|token1
argument_list|)
expr_stmt|;
name|fail
argument_list|(
literal|"Expected InvalidToken"
argument_list|)
expr_stmt|;
block|}
catch|catch
parameter_list|(
name|SecretManager
operator|.
name|InvalidToken
name|it
parameter_list|)
block|{
comment|// Ignore
block|}
comment|// Create a new token thru the new ZKDTSM
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
name|token4
init|=
operator|(
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
operator|)
name|tm2
operator|.
name|createToken
argument_list|(
name|UserGroupInformation
operator|.
name|getCurrentUser
argument_list|()
argument_list|,
literal|"xyz"
argument_list|)
decl_stmt|;
name|Assert
operator|.
name|assertNotNull
argument_list|(
name|token4
argument_list|)
expr_stmt|;
name|tm2
operator|.
name|verifyToken
argument_list|(
name|token4
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|verifyToken
argument_list|(
name|token4
argument_list|)
expr_stmt|;
comment|// Bring down tm2
name|verifyDestroy
argument_list|(
name|tm2
argument_list|,
name|conf
argument_list|)
expr_stmt|;
comment|// Start third node after some time..
name|Thread
operator|.
name|sleep
argument_list|(
literal|1000
argument_list|)
expr_stmt|;
name|DelegationTokenManager
name|tm3
init|=
operator|new
name|DelegationTokenManager
argument_list|(
name|conf
argument_list|,
operator|new
name|Text
argument_list|(
literal|"bla"
argument_list|)
argument_list|)
decl_stmt|;
name|tm3
operator|.
name|init
argument_list|()
expr_stmt|;
name|tm3
operator|.
name|verifyToken
argument_list|(
name|token2
argument_list|)
expr_stmt|;
name|tm3
operator|.
name|verifyToken
argument_list|(
name|token3
argument_list|)
expr_stmt|;
name|tm3
operator|.
name|verifyToken
argument_list|(
name|token4
argument_list|)
expr_stmt|;
try|try
block|{
name|verifyTokenFail
argument_list|(
name|tm3
argument_list|,
name|token1
argument_list|)
expr_stmt|;
name|fail
argument_list|(
literal|"Expected InvalidToken"
argument_list|)
expr_stmt|;
block|}
catch|catch
parameter_list|(
name|SecretManager
operator|.
name|InvalidToken
name|it
parameter_list|)
block|{
comment|// Ignore
block|}
name|verifyDestroy
argument_list|(
name|tm3
argument_list|,
name|conf
argument_list|)
expr_stmt|;
name|verifyDestroy
argument_list|(
name|tm1
argument_list|,
name|conf
argument_list|)
expr_stmt|;
block|}
block|}
annotation|@
name|SuppressWarnings
argument_list|(
literal|"unchecked"
argument_list|)
annotation|@
name|Test
DECL|method|testRenewTokenSingleManager ()
specifier|public
name|void
name|testRenewTokenSingleManager
parameter_list|()
throws|throws
name|Exception
block|{
for|for
control|(
name|int
name|i
init|=
literal|0
init|;
name|i
operator|<
name|TEST_RETRIES
condition|;
name|i
operator|++
control|)
block|{
name|DelegationTokenManager
name|tm1
init|=
literal|null
decl_stmt|;
name|String
name|connectString
init|=
name|zkServer
operator|.
name|getConnectString
argument_list|()
decl_stmt|;
name|Configuration
name|conf
init|=
name|getSecretConf
argument_list|(
name|connectString
argument_list|)
decl_stmt|;
name|tm1
operator|=
operator|new
name|DelegationTokenManager
argument_list|(
name|conf
argument_list|,
operator|new
name|Text
argument_list|(
literal|"foo"
argument_list|)
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|init
argument_list|()
expr_stmt|;
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
name|token
init|=
operator|(
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
operator|)
name|tm1
operator|.
name|createToken
argument_list|(
name|UserGroupInformation
operator|.
name|getCurrentUser
argument_list|()
argument_list|,
literal|"foo"
argument_list|)
decl_stmt|;
name|Assert
operator|.
name|assertNotNull
argument_list|(
name|token
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|renewToken
argument_list|(
name|token
argument_list|,
literal|"foo"
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|verifyToken
argument_list|(
name|token
argument_list|)
expr_stmt|;
name|verifyDestroy
argument_list|(
name|tm1
argument_list|,
name|conf
argument_list|)
expr_stmt|;
block|}
block|}
annotation|@
name|SuppressWarnings
argument_list|(
literal|"unchecked"
argument_list|)
annotation|@
name|Test
DECL|method|testCancelTokenSingleManager ()
specifier|public
name|void
name|testCancelTokenSingleManager
parameter_list|()
throws|throws
name|Exception
block|{
for|for
control|(
name|int
name|i
init|=
literal|0
init|;
name|i
operator|<
name|TEST_RETRIES
condition|;
name|i
operator|++
control|)
block|{
name|DelegationTokenManager
name|tm1
init|=
literal|null
decl_stmt|;
name|String
name|connectString
init|=
name|zkServer
operator|.
name|getConnectString
argument_list|()
decl_stmt|;
name|Configuration
name|conf
init|=
name|getSecretConf
argument_list|(
name|connectString
argument_list|)
decl_stmt|;
name|tm1
operator|=
operator|new
name|DelegationTokenManager
argument_list|(
name|conf
argument_list|,
operator|new
name|Text
argument_list|(
literal|"foo"
argument_list|)
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|init
argument_list|()
expr_stmt|;
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
name|token
init|=
operator|(
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
operator|)
name|tm1
operator|.
name|createToken
argument_list|(
name|UserGroupInformation
operator|.
name|getCurrentUser
argument_list|()
argument_list|,
literal|"foo"
argument_list|)
decl_stmt|;
name|Assert
operator|.
name|assertNotNull
argument_list|(
name|token
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|cancelToken
argument_list|(
name|token
argument_list|,
literal|"foo"
argument_list|)
expr_stmt|;
try|try
block|{
name|verifyTokenFail
argument_list|(
name|tm1
argument_list|,
name|token
argument_list|)
expr_stmt|;
name|fail
argument_list|(
literal|"Expected InvalidToken"
argument_list|)
expr_stmt|;
block|}
catch|catch
parameter_list|(
name|SecretManager
operator|.
name|InvalidToken
name|it
parameter_list|)
block|{
name|it
operator|.
name|printStackTrace
argument_list|()
expr_stmt|;
block|}
name|verifyDestroy
argument_list|(
name|tm1
argument_list|,
name|conf
argument_list|)
expr_stmt|;
block|}
block|}
annotation|@
name|SuppressWarnings
argument_list|(
literal|"rawtypes"
argument_list|)
DECL|method|verifyDestroy (DelegationTokenManager tm, Configuration conf)
specifier|protected
name|void
name|verifyDestroy
parameter_list|(
name|DelegationTokenManager
name|tm
parameter_list|,
name|Configuration
name|conf
parameter_list|)
throws|throws
name|Exception
block|{
name|AbstractDelegationTokenSecretManager
name|sm
init|=
name|tm
operator|.
name|getDelegationTokenSecretManager
argument_list|()
decl_stmt|;
name|ZKDelegationTokenSecretManager
name|zksm
init|=
operator|(
name|ZKDelegationTokenSecretManager
operator|)
name|sm
decl_stmt|;
name|ExecutorService
name|es
init|=
name|zksm
operator|.
name|getListenerThreadPool
argument_list|()
decl_stmt|;
name|tm
operator|.
name|destroy
argument_list|()
expr_stmt|;
name|Assert
operator|.
name|assertTrue
argument_list|(
name|es
operator|.
name|isShutdown
argument_list|()
argument_list|)
expr_stmt|;
comment|// wait for the pool to terminate
name|long
name|timeout
init|=
name|conf
operator|.
name|getLong
argument_list|(
name|ZKDelegationTokenSecretManager
operator|.
name|ZK_DTSM_ZK_SHUTDOWN_TIMEOUT
argument_list|,
name|ZKDelegationTokenSecretManager
operator|.
name|ZK_DTSM_ZK_SHUTDOWN_TIMEOUT_DEFAULT
argument_list|)
decl_stmt|;
name|Thread
operator|.
name|sleep
argument_list|(
name|timeout
operator|*
literal|3
argument_list|)
expr_stmt|;
name|Assert
operator|.
name|assertTrue
argument_list|(
name|es
operator|.
name|isTerminated
argument_list|()
argument_list|)
expr_stmt|;
block|}
annotation|@
name|SuppressWarnings
argument_list|(
block|{
literal|"unchecked"
block|,
literal|"rawtypes"
block|}
argument_list|)
annotation|@
name|Test
DECL|method|testStopThreads ()
specifier|public
name|void
name|testStopThreads
parameter_list|()
throws|throws
name|Exception
block|{
name|DelegationTokenManager
name|tm1
init|=
literal|null
decl_stmt|;
name|String
name|connectString
init|=
name|zkServer
operator|.
name|getConnectString
argument_list|()
decl_stmt|;
comment|// let's make the update interval short and the shutdown interval
comment|// comparatively longer, so if the update thread runs after shutdown,
comment|// it will cause an error.
specifier|final
name|long
name|updateIntervalSeconds
init|=
literal|1
decl_stmt|;
specifier|final
name|long
name|shutdownTimeoutMillis
init|=
name|updateIntervalSeconds
operator|*
literal|1000
operator|*
literal|5
decl_stmt|;
name|Configuration
name|conf
init|=
name|getSecretConf
argument_list|(
name|connectString
argument_list|)
decl_stmt|;
name|conf
operator|.
name|setLong
argument_list|(
name|DelegationTokenManager
operator|.
name|UPDATE_INTERVAL
argument_list|,
name|updateIntervalSeconds
argument_list|)
expr_stmt|;
name|conf
operator|.
name|setLong
argument_list|(
name|DelegationTokenManager
operator|.
name|REMOVAL_SCAN_INTERVAL
argument_list|,
name|updateIntervalSeconds
argument_list|)
expr_stmt|;
name|conf
operator|.
name|setLong
argument_list|(
name|DelegationTokenManager
operator|.
name|RENEW_INTERVAL
argument_list|,
name|updateIntervalSeconds
argument_list|)
expr_stmt|;
name|conf
operator|.
name|setLong
argument_list|(
name|ZKDelegationTokenSecretManager
operator|.
name|ZK_DTSM_ZK_SHUTDOWN_TIMEOUT
argument_list|,
name|shutdownTimeoutMillis
argument_list|)
expr_stmt|;
name|tm1
operator|=
operator|new
name|DelegationTokenManager
argument_list|(
name|conf
argument_list|,
operator|new
name|Text
argument_list|(
literal|"foo"
argument_list|)
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|init
argument_list|()
expr_stmt|;
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
name|token
init|=
operator|(
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
operator|)
name|tm1
operator|.
name|createToken
argument_list|(
name|UserGroupInformation
operator|.
name|getCurrentUser
argument_list|()
argument_list|,
literal|"foo"
argument_list|)
decl_stmt|;
name|Assert
operator|.
name|assertNotNull
argument_list|(
name|token
argument_list|)
expr_stmt|;
name|AbstractDelegationTokenSecretManager
name|sm
init|=
name|tm1
operator|.
name|getDelegationTokenSecretManager
argument_list|()
decl_stmt|;
name|ZKDelegationTokenSecretManager
name|zksm
init|=
operator|(
name|ZKDelegationTokenSecretManager
operator|)
name|sm
decl_stmt|;
name|ExecutorService
name|es
init|=
name|zksm
operator|.
name|getListenerThreadPool
argument_list|()
decl_stmt|;
name|es
operator|.
name|submit
argument_list|(
operator|new
name|Callable
argument_list|<
name|Void
argument_list|>
argument_list|()
block|{
specifier|public
name|Void
name|call
parameter_list|()
throws|throws
name|Exception
block|{
name|Thread
operator|.
name|sleep
argument_list|(
name|shutdownTimeoutMillis
operator|*
literal|2
argument_list|)
expr_stmt|;
comment|// force this to be shutdownNow
return|return
literal|null
return|;
block|}
block|}
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|destroy
argument_list|()
expr_stmt|;
block|}
annotation|@
name|Test
DECL|method|testACLs ()
specifier|public
name|void
name|testACLs
parameter_list|()
throws|throws
name|Exception
block|{
name|DelegationTokenManager
name|tm1
decl_stmt|;
name|String
name|connectString
init|=
name|zkServer
operator|.
name|getConnectString
argument_list|()
decl_stmt|;
name|Configuration
name|conf
init|=
name|getSecretConf
argument_list|(
name|connectString
argument_list|)
decl_stmt|;
name|RetryPolicy
name|retryPolicy
init|=
operator|new
name|ExponentialBackoffRetry
argument_list|(
literal|1000
argument_list|,
literal|3
argument_list|)
decl_stmt|;
name|String
name|userPass
init|=
literal|"<PASSWORD>"
decl_stmt|;
specifier|final
name|ACL
name|digestACL
init|=
operator|new
name|ACL
argument_list|(
name|ZooDefs
operator|.
name|Perms
operator|.
name|ALL
argument_list|,
operator|new
name|Id
argument_list|(
literal|"digest"
argument_list|,
name|DigestAuthenticationProvider
operator|.
name|generateDigest
argument_list|(
name|userPass
argument_list|)
argument_list|)
argument_list|)
decl_stmt|;
name|ACLProvider
name|digestAclProvider
init|=
operator|new
name|ACLProvider
argument_list|()
block|{
annotation|@
name|Override
specifier|public
name|List
argument_list|<
name|ACL
argument_list|>
name|getAclForPath
parameter_list|(
name|String
name|path
parameter_list|)
block|{
return|return
name|getDefaultAcl
argument_list|()
return|;
block|}
annotation|@
name|Override
specifier|public
name|List
argument_list|<
name|ACL
argument_list|>
name|getDefaultAcl
parameter_list|()
block|{
name|List
argument_list|<
name|ACL
argument_list|>
name|ret
init|=
operator|new
name|ArrayList
argument_list|<
name|ACL
argument_list|>
argument_list|()
decl_stmt|;
name|ret
operator|.
name|add
argument_list|(
name|digestACL
argument_list|)
expr_stmt|;
return|return
name|ret
return|;
block|}
block|}
decl_stmt|;
name|CuratorFramework
name|curatorFramework
init|=
name|CuratorFrameworkFactory
operator|.
name|builder
argument_list|()
operator|.
name|connectString
argument_list|(
name|connectString
argument_list|)
operator|.
name|retryPolicy
argument_list|(
name|retryPolicy
argument_list|)
operator|.
name|aclProvider
argument_list|(
name|digestAclProvider
argument_list|)
operator|.
name|authorization
argument_list|(
literal|"digest"
argument_list|,
name|userPass
operator|.
name|getBytes
argument_list|(
literal|"UTF-8"
argument_list|)
argument_list|)
operator|.
name|build
argument_list|()
decl_stmt|;
name|curatorFramework
operator|.
name|start
argument_list|()
expr_stmt|;
name|ZKDelegationTokenSecretManager
operator|.
name|setCurator
argument_list|(
name|curatorFramework
argument_list|)
expr_stmt|;
name|tm1
operator|=
operator|new
name|DelegationTokenManager
argument_list|(
name|conf
argument_list|,
operator|new
name|Text
argument_list|(
literal|"bla"
argument_list|)
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|init
argument_list|()
expr_stmt|;
comment|// check ACL
name|String
name|workingPath
init|=
name|conf
operator|.
name|get
argument_list|(
name|ZKDelegationTokenSecretManager
operator|.
name|ZK_DTSM_ZNODE_WORKING_PATH
argument_list|)
decl_stmt|;
name|verifyACL
argument_list|(
name|curatorFramework
argument_list|,
literal|"/"
operator|+
name|workingPath
argument_list|,
name|digestACL
argument_list|)
expr_stmt|;
name|tm1
operator|.
name|destroy
argument_list|()
expr_stmt|;
name|ZKDelegationTokenSecretManager
operator|.
name|setCurator
argument_list|(
literal|null
argument_list|)
expr_stmt|;
name|curatorFramework
operator|.
name|close
argument_list|()
expr_stmt|;
block|}
DECL|method|verifyACL (CuratorFramework curatorFramework, String path, ACL expectedACL)
specifier|private
name|void
name|verifyACL
parameter_list|(
name|CuratorFramework
name|curatorFramework
parameter_list|,
name|String
name|path
parameter_list|,
name|ACL
name|expectedACL
parameter_list|)
throws|throws
name|Exception
block|{
name|List
argument_list|<
name|ACL
argument_list|>
name|acls
init|=
name|curatorFramework
operator|.
name|getACL
argument_list|()
operator|.
name|forPath
argument_list|(
name|path
argument_list|)
decl_stmt|;
name|Assert
operator|.
name|assertEquals
argument_list|(
literal|1
argument_list|,
name|acls
operator|.
name|size
argument_list|()
argument_list|)
expr_stmt|;
name|Assert
operator|.
name|assertEquals
argument_list|(
name|expectedACL
argument_list|,
name|acls
operator|.
name|get
argument_list|(
literal|0
argument_list|)
argument_list|)
expr_stmt|;
block|}
comment|// Since it is possible that there can be a delay for the cancel token message
comment|// initiated by one node to reach another node.. The second node can ofcourse
comment|// verify with ZK directly if the token that needs verification has been
comment|// cancelled but.. that would mean having to make an RPC call for every
comment|// verification request.
comment|// Thus, the eventual consistency tradef-off should be acceptable here...
DECL|method|verifyTokenFail (DelegationTokenManager tm, Token<DelegationTokenIdentifier> token)
specifier|private
name|void
name|verifyTokenFail
parameter_list|(
name|DelegationTokenManager
name|tm
parameter_list|,
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
name|token
parameter_list|)
throws|throws
name|IOException
throws|,
name|InterruptedException
block|{
name|verifyTokenFailWithRetry
argument_list|(
name|tm
argument_list|,
name|token
argument_list|,
name|RETRY_COUNT
argument_list|)
expr_stmt|;
block|}
DECL|method|verifyTokenFailWithRetry (DelegationTokenManager tm, Token<DelegationTokenIdentifier> token, int retryCount)
specifier|private
name|void
name|verifyTokenFailWithRetry
parameter_list|(
name|DelegationTokenManager
name|tm
parameter_list|,
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
name|token
parameter_list|,
name|int
name|retryCount
parameter_list|)
throws|throws
name|IOException
throws|,
name|InterruptedException
block|{
try|try
block|{
name|tm
operator|.
name|verifyToken
argument_list|(
name|token
argument_list|)
expr_stmt|;
block|}
catch|catch
parameter_list|(
name|SecretManager
operator|.
name|InvalidToken
name|er
parameter_list|)
block|{
throw|throw
name|er
throw|;
block|}
if|if
condition|(
name|retryCount
operator|>
literal|0
condition|)
block|{
name|Thread
operator|.
name|sleep
argument_list|(
name|RETRY_WAIT
argument_list|)
expr_stmt|;
name|verifyTokenFailWithRetry
argument_list|(
name|tm
argument_list|,
name|token
argument_list|,
name|retryCount
operator|-
literal|1
argument_list|)
expr_stmt|;
block|}
block|}
annotation|@
name|SuppressWarnings
argument_list|(
block|{
literal|"unchecked"
block|}
argument_list|)
annotation|@
name|Test
DECL|method|testNodesLoadedAfterRestart ()
specifier|public
name|void
name|testNodesLoadedAfterRestart
parameter_list|()
throws|throws
name|Exception
block|{
specifier|final
name|String
name|connectString
init|=
name|zkServer
operator|.
name|getConnectString
argument_list|()
decl_stmt|;
specifier|final
name|Configuration
name|conf
init|=
name|getSecretConf
argument_list|(
name|connectString
argument_list|)
decl_stmt|;
specifier|final
name|int
name|removeScan
init|=
literal|1
decl_stmt|;
comment|// Set the remove scan interval to remove expired tokens
name|conf
operator|.
name|setLong
argument_list|(
name|DelegationTokenManager
operator|.
name|REMOVAL_SCAN_INTERVAL
argument_list|,
name|removeScan
argument_list|)
expr_stmt|;
comment|// Set the update interval to trigger background thread to run. The thread
comment|// is hard-coded to sleep at least 5 seconds.
name|conf
operator|.
name|setLong
argument_list|(
name|DelegationTokenManager
operator|.
name|UPDATE_INTERVAL
argument_list|,
literal|5
argument_list|)
expr_stmt|;
comment|// Set token expire time to 5 seconds.
name|conf
operator|.
name|setLong
argument_list|(
name|DelegationTokenManager
operator|.
name|RENEW_INTERVAL
argument_list|,
literal|5
argument_list|)
expr_stmt|;
name|DelegationTokenManager
name|tm
init|=
operator|new
name|DelegationTokenManager
argument_list|(
name|conf
argument_list|,
operator|new
name|Text
argument_list|(
literal|"bla"
argument_list|)
argument_list|)
decl_stmt|;
name|tm
operator|.
name|init
argument_list|()
expr_stmt|;
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
name|token
init|=
operator|(
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
operator|)
name|tm
operator|.
name|createToken
argument_list|(
name|UserGroupInformation
operator|.
name|getCurrentUser
argument_list|()
argument_list|,
literal|"good"
argument_list|)
decl_stmt|;
name|Assert
operator|.
name|assertNotNull
argument_list|(
name|token
argument_list|)
expr_stmt|;
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
name|cancelled
init|=
operator|(
name|Token
argument_list|<
name|DelegationTokenIdentifier
argument_list|>
operator|)
name|tm
operator|.
name|createToken
argument_list|(
name|UserGroupInformation
operator|.
name|getCurrentUser
argument_list|()
argument_list|,
literal|"cancelled"
argument_list|)
decl_stmt|;
name|Assert
operator|.
name|assertNotNull
argument_list|(
name|cancelled
argument_list|)
expr_stmt|;
name|tm
operator|.
name|verifyToken
argument_list|(
name|token
argument_list|)
expr_stmt|;
name|tm
operator|.
name|verifyToken
argument_list|(
name|cancelled
argument_list|)
expr_stmt|;
comment|// Cancel one token, verify it's gone
name|tm
operator|.
name|cancelToken
argument_list|(
name|cancelled
argument_list|,
literal|"cancelled"
argument_list|)
expr_stmt|;
specifier|final
name|AbstractDelegationTokenSecretManager
name|sm
init|=
name|tm
operator|.
name|getDelegationTokenSecretManager
argument_list|()
decl_stmt|;
specifier|final
name|ZKDelegationTokenSecretManager
name|zksm
init|=
operator|(
name|ZKDelegationTokenSecretManager
operator|)
name|sm
decl_stmt|;
specifier|final
name|AbstractDelegationTokenIdentifier
name|idCancelled
init|=
name|sm
operator|.
name|decodeTokenIdentifier
argument_list|(
name|cancelled
argument_list|)
decl_stmt|;
name|LOG
operator|.
name|info
argument_list|(
literal|"Waiting for the cancelled token to be removed"
argument_list|)
expr_stmt|;
name|GenericTestUtils
operator|.
name|waitFor
argument_list|(
operator|new
name|Supplier
argument_list|<
name|Boolean
argument_list|>
argument_list|()
block|{
annotation|@
name|Override
specifier|public
name|Boolean
name|get
parameter_list|()
block|{
name|AbstractDelegationTokenSecretManager
operator|.
name|DelegationTokenInformation
name|dtinfo
init|=
name|zksm
operator|.
name|getTokenInfo
argument_list|(
name|idCancelled
argument_list|)
decl_stmt|;
return|return
name|dtinfo
operator|==
literal|null
return|;
block|}
block|}
argument_list|,
literal|100
argument_list|,
literal|5000
argument_list|)
expr_stmt|;
comment|// Fake a restart which launches a new tm
name|tm
operator|.
name|destroy
argument_list|()
expr_stmt|;
name|tm
operator|=
operator|new
name|DelegationTokenManager
argument_list|(
name|conf
argument_list|,
operator|new
name|Text
argument_list|(
literal|"bla"
argument_list|)
argument_list|)
expr_stmt|;
name|tm
operator|.
name|init
argument_list|()
expr_stmt|;
specifier|final
name|AbstractDelegationTokenSecretManager
name|smNew
init|=
name|tm
operator|.
name|getDelegationTokenSecretManager
argument_list|()
decl_stmt|;
specifier|final
name|ZKDelegationTokenSecretManager
name|zksmNew
init|=
operator|(
name|ZKDelegationTokenSecretManager
operator|)
name|smNew
decl_stmt|;
comment|// The cancelled token should be gone, and not loaded.
name|AbstractDelegationTokenIdentifier
name|id
init|=
name|smNew
operator|.
name|decodeTokenIdentifier
argument_list|(
name|cancelled
argument_list|)
decl_stmt|;
name|AbstractDelegationTokenSecretManager
operator|.
name|DelegationTokenInformation
name|dtinfo
init|=
name|zksmNew
operator|.
name|getTokenInfo
argument_list|(
name|id
argument_list|)
decl_stmt|;
name|Assert
operator|.
name|assertNull
argument_list|(
literal|"canceled dt should be gone!"
argument_list|,
name|dtinfo
argument_list|)
expr_stmt|;
comment|// The good token should be loaded on startup, and removed after expiry.
name|id
operator|=
name|smNew
operator|.
name|decodeTokenIdentifier
argument_list|(
name|token
argument_list|)
expr_stmt|;
name|dtinfo
operator|=
name|zksmNew
operator|.
name|getTokenInfoFromMemory
argument_list|(
name|id
argument_list|)
expr_stmt|;
name|Assert
operator|.
name|assertNotNull
argument_list|(
literal|"good dt should be in memory!"
argument_list|,
name|dtinfo
argument_list|)
expr_stmt|;
comment|// Wait for the good token to expire.
name|Thread
operator|.
name|sleep
argument_list|(
literal|5000
argument_list|)
expr_stmt|;
specifier|final
name|ZKDelegationTokenSecretManager
name|zksm1
init|=
name|zksmNew
decl_stmt|;
specifier|final
name|AbstractDelegationTokenIdentifier
name|id1
init|=
name|id
decl_stmt|;
name|GenericTestUtils
operator|.
name|waitFor
argument_list|(
operator|new
name|Supplier
argument_list|<
name|Boolean
argument_list|>
argument_list|()
block|{
annotation|@
name|Override
specifier|public
name|Boolean
name|get
parameter_list|()
block|{
name|LOG
operator|.
name|info
argument_list|(
literal|"Waiting for the expired token to be removed..."
argument_list|)
expr_stmt|;
return|return
name|zksm1
operator|.
name|getTokenInfo
argument_list|(
name|id1
argument_list|)
operator|==
literal|null
return|;
block|}
block|}
argument_list|,
literal|1000
argument_list|,
literal|5000
argument_list|)
expr_stmt|;
block|}
block|}
end_class

end_unit

