Manikumar Reddy O
7 years ago
committed by
Jun Rao
41 changed files with 995 additions and 178 deletions
@ -0,0 +1,53 @@
@@ -0,0 +1,53 @@
|
||||
/* |
||||
* 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. |
||||
*/ |
||||
|
||||
package org.apache.kafka.clients.admin; |
||||
|
||||
import java.util.LinkedList; |
||||
import java.util.List; |
||||
|
||||
import org.apache.kafka.common.annotation.InterfaceStability; |
||||
import org.apache.kafka.common.security.auth.KafkaPrincipal; |
||||
|
||||
/** |
||||
* Options for {@link AdminClient#createDelegationToken(CreateDelegationTokenOptions)}. |
||||
* |
||||
* The API of this class is evolving, see {@link AdminClient} for details. |
||||
*/ |
||||
@InterfaceStability.Evolving |
||||
public class CreateDelegationTokenOptions extends AbstractOptions<CreateDelegationTokenOptions> { |
||||
private long maxLifeTimeMs = -1; |
||||
private List<KafkaPrincipal> renewers = new LinkedList<>(); |
||||
|
||||
public CreateDelegationTokenOptions renewers(List<KafkaPrincipal> renewers) { |
||||
this.renewers = renewers; |
||||
return this; |
||||
} |
||||
|
||||
public List<KafkaPrincipal> renewers() { |
||||
return renewers; |
||||
} |
||||
|
||||
public CreateDelegationTokenOptions maxlifeTimeMs(long maxLifeTimeMs) { |
||||
this.maxLifeTimeMs = maxLifeTimeMs; |
||||
return this; |
||||
} |
||||
|
||||
public long maxlifeTimeMs() { |
||||
return maxLifeTimeMs; |
||||
} |
||||
} |
||||
@ -0,0 +1,43 @@
@@ -0,0 +1,43 @@
|
||||
/* |
||||
* 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. |
||||
*/ |
||||
|
||||
package org.apache.kafka.clients.admin; |
||||
|
||||
import org.apache.kafka.common.KafkaFuture; |
||||
import org.apache.kafka.common.annotation.InterfaceStability; |
||||
import org.apache.kafka.common.security.token.delegation.DelegationToken; |
||||
|
||||
/** |
||||
* The result of the {@link KafkaAdminClient#createDelegationToken(CreateDelegationTokenOptions)} call. |
||||
* |
||||
* The API of this class is evolving, see {@link AdminClient} for details. |
||||
*/ |
||||
@InterfaceStability.Evolving |
||||
public class CreateDelegationTokenResult { |
||||
private final KafkaFuture<DelegationToken> delegationToken; |
||||
|
||||
CreateDelegationTokenResult(KafkaFuture<DelegationToken> delegationToken) { |
||||
this.delegationToken = delegationToken; |
||||
} |
||||
|
||||
/** |
||||
* Returns a future which yields a delegation token |
||||
*/ |
||||
public KafkaFuture<DelegationToken> delegationToken() { |
||||
return delegationToken; |
||||
} |
||||
} |
||||
@ -0,0 +1,48 @@
@@ -0,0 +1,48 @@
|
||||
/* |
||||
* 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. |
||||
*/ |
||||
|
||||
package org.apache.kafka.clients.admin; |
||||
|
||||
import java.util.List; |
||||
|
||||
import org.apache.kafka.common.annotation.InterfaceStability; |
||||
import org.apache.kafka.common.security.auth.KafkaPrincipal; |
||||
|
||||
/** |
||||
* Options for {@link AdminClient#describeDelegationToken(DescribeDelegationTokenOptions)}. |
||||
* |
||||
* The API of this class is evolving, see {@link AdminClient} for details. |
||||
*/ |
||||
@InterfaceStability.Evolving |
||||
public class DescribeDelegationTokenOptions extends AbstractOptions<DescribeDelegationTokenOptions> { |
||||
private List<KafkaPrincipal> owners; |
||||
|
||||
/** |
||||
* if owners is null, all the user owned tokens and tokens where user have Describe permission |
||||
* will be returned. |
||||
* @param owners |
||||
* @return this instance |
||||
*/ |
||||
public DescribeDelegationTokenOptions owners(List<KafkaPrincipal> owners) { |
||||
this.owners = owners; |
||||
return this; |
||||
} |
||||
|
||||
public List<KafkaPrincipal> owners() { |
||||
return owners; |
||||
} |
||||
} |
||||
@ -0,0 +1,45 @@
@@ -0,0 +1,45 @@
|
||||
/* |
||||
* 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. |
||||
*/ |
||||
|
||||
package org.apache.kafka.clients.admin; |
||||
|
||||
import java.util.List; |
||||
|
||||
import org.apache.kafka.common.KafkaFuture; |
||||
import org.apache.kafka.common.annotation.InterfaceStability; |
||||
import org.apache.kafka.common.security.token.delegation.DelegationToken; |
||||
|
||||
/** |
||||
* The result of the {@link KafkaAdminClient#describeDelegationToken(DescribeDelegationTokenOptions)} call. |
||||
* |
||||
* The API of this class is evolving, see {@link AdminClient} for details. |
||||
*/ |
||||
@InterfaceStability.Evolving |
||||
public class DescribeDelegationTokenResult { |
||||
private final KafkaFuture<List<DelegationToken>> delegationTokens; |
||||
|
||||
DescribeDelegationTokenResult(KafkaFuture<List<DelegationToken>> delegationTokens) { |
||||
this.delegationTokens = delegationTokens; |
||||
} |
||||
|
||||
/** |
||||
* Returns a future which yields list of delegation tokens |
||||
*/ |
||||
public KafkaFuture<List<DelegationToken>> delegationTokens() { |
||||
return delegationTokens; |
||||
} |
||||
} |
||||
@ -0,0 +1,39 @@
@@ -0,0 +1,39 @@
|
||||
/* |
||||
* 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. |
||||
*/ |
||||
|
||||
package org.apache.kafka.clients.admin; |
||||
|
||||
import org.apache.kafka.common.annotation.InterfaceStability; |
||||
|
||||
/** |
||||
* Options for {@link AdminClient#expireDelegationToken(byte[], ExpireDelegationTokenOptions)}. |
||||
* |
||||
* The API of this class is evolving, see {@link AdminClient} for details. |
||||
*/ |
||||
@InterfaceStability.Evolving |
||||
public class ExpireDelegationTokenOptions extends AbstractOptions<ExpireDelegationTokenOptions> { |
||||
private long expiryTimePeriodMs = -1L; |
||||
|
||||
public ExpireDelegationTokenOptions expiryTimePeriodMs(long expiryTimePeriodMs) { |
||||
this.expiryTimePeriodMs = expiryTimePeriodMs; |
||||
return this; |
||||
} |
||||
|
||||
public long expiryTimePeriodMs() { |
||||
return expiryTimePeriodMs; |
||||
} |
||||
} |
||||
@ -0,0 +1,42 @@
@@ -0,0 +1,42 @@
|
||||
/* |
||||
* 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. |
||||
*/ |
||||
|
||||
package org.apache.kafka.clients.admin; |
||||
|
||||
import org.apache.kafka.common.KafkaFuture; |
||||
import org.apache.kafka.common.annotation.InterfaceStability; |
||||
|
||||
/** |
||||
* The result of the {@link KafkaAdminClient#expireDelegationToken(byte[], ExpireDelegationTokenOptions)} call. |
||||
* |
||||
* The API of this class is evolving, see {@link AdminClient} for details. |
||||
*/ |
||||
@InterfaceStability.Evolving |
||||
public class ExpireDelegationTokenResult { |
||||
private final KafkaFuture<Long> expiryTimestamp; |
||||
|
||||
ExpireDelegationTokenResult(KafkaFuture<Long> expiryTimestamp) { |
||||
this.expiryTimestamp = expiryTimestamp; |
||||
} |
||||
|
||||
/** |
||||
* Returns a future which yields expiry timestamp |
||||
*/ |
||||
public KafkaFuture<Long> expiryTimestamp() { |
||||
return expiryTimestamp; |
||||
} |
||||
} |
||||
@ -0,0 +1,39 @@
@@ -0,0 +1,39 @@
|
||||
/* |
||||
* 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. |
||||
*/ |
||||
|
||||
package org.apache.kafka.clients.admin; |
||||
|
||||
import org.apache.kafka.common.annotation.InterfaceStability; |
||||
|
||||
/** |
||||
* Options for {@link AdminClient#renewDelegationToken(byte[], RenewDelegationTokenOptions)}. |
||||
* |
||||
* The API of this class is evolving, see {@link AdminClient} for details. |
||||
*/ |
||||
@InterfaceStability.Evolving |
||||
public class RenewDelegationTokenOptions extends AbstractOptions<RenewDelegationTokenOptions> { |
||||
private long renewTimePeriodMs = -1; |
||||
|
||||
public RenewDelegationTokenOptions renewTimePeriodMs(long renewTimePeriodMs) { |
||||
this.renewTimePeriodMs = renewTimePeriodMs; |
||||
return this; |
||||
} |
||||
|
||||
public long renewTimePeriodMs() { |
||||
return renewTimePeriodMs; |
||||
} |
||||
} |
||||
@ -0,0 +1,42 @@
@@ -0,0 +1,42 @@
|
||||
/* |
||||
* 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. |
||||
*/ |
||||
|
||||
package org.apache.kafka.clients.admin; |
||||
|
||||
import org.apache.kafka.common.KafkaFuture; |
||||
import org.apache.kafka.common.annotation.InterfaceStability; |
||||
|
||||
/** |
||||
* The result of the {@link KafkaAdminClient#expireDelegationToken(byte[], ExpireDelegationTokenOptions)} call. |
||||
* |
||||
* The API of this class is evolving, see {@link AdminClient} for details. |
||||
*/ |
||||
@InterfaceStability.Evolving |
||||
public class RenewDelegationTokenResult { |
||||
private final KafkaFuture<Long> expiryTimestamp; |
||||
|
||||
RenewDelegationTokenResult(KafkaFuture<Long> expiryTimestamp) { |
||||
this.expiryTimestamp = expiryTimestamp; |
||||
} |
||||
|
||||
/** |
||||
* Returns a future which yields expiry timestamp |
||||
*/ |
||||
public KafkaFuture<Long> expiryTimestamp() { |
||||
return expiryTimestamp; |
||||
} |
||||
} |
||||
@ -0,0 +1,147 @@
@@ -0,0 +1,147 @@
|
||||
/** |
||||
* 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. |
||||
*/ |
||||
package kafka.admin |
||||
|
||||
import java.util |
||||
|
||||
import kafka.admin.DelegationTokenCommand.DelegationTokenCommandOptions |
||||
import kafka.api.{KafkaSasl, SaslSetup} |
||||
import kafka.server.{BaseRequestTest, KafkaConfig} |
||||
import kafka.utils.{JaasTestUtils, TestUtils} |
||||
import org.apache.kafka.clients.admin.AdminClientConfig |
||||
import org.apache.kafka.common.security.auth.SecurityProtocol |
||||
import org.junit.Assert._ |
||||
import org.junit.{After, Before, Test} |
||||
|
||||
import scala.collection.JavaConverters._ |
||||
import scala.collection.mutable.ListBuffer |
||||
import scala.concurrent.ExecutionException |
||||
|
||||
class DelegationTokenCommandTest extends BaseRequestTest with SaslSetup { |
||||
override protected def securityProtocol = SecurityProtocol.SASL_PLAINTEXT |
||||
private val kafkaClientSaslMechanism = "PLAIN" |
||||
private val kafkaServerSaslMechanisms = List("PLAIN") |
||||
protected override val serverSaslProperties = Some(kafkaServerSaslProperties(kafkaServerSaslMechanisms, kafkaClientSaslMechanism)) |
||||
protected override val clientSaslProperties = Some(kafkaClientSaslProperties(kafkaClientSaslMechanism)) |
||||
var adminClient: org.apache.kafka.clients.admin.AdminClient = null |
||||
|
||||
override def numBrokers = 1 |
||||
|
||||
@Before |
||||
override def setUp(): Unit = { |
||||
startSasl(jaasSections(kafkaServerSaslMechanisms, Some(kafkaClientSaslMechanism), KafkaSasl, JaasTestUtils.KafkaServerContextName)) |
||||
super.setUp() |
||||
} |
||||
|
||||
override def generateConfigs = { |
||||
val props = TestUtils.createBrokerConfigs(numBrokers, zkConnect, |
||||
enableControlledShutdown = false, enableDeleteTopic = true, |
||||
interBrokerSecurityProtocol = Some(securityProtocol), |
||||
trustStoreFile = trustStoreFile, saslProperties = serverSaslProperties, enableToken = true) |
||||
props.foreach(propertyOverrides) |
||||
props.map(KafkaConfig.fromProps) |
||||
} |
||||
|
||||
private def createAdminConfig():util.Map[String, Object] = { |
||||
val config = new util.HashMap[String, Object] |
||||
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList) |
||||
val securityProps: util.Map[Object, Object] = |
||||
TestUtils.adminClientSecurityConfigs(securityProtocol, trustStoreFile, clientSaslProperties) |
||||
securityProps.asScala.foreach { case (key, value) => config.put(key.asInstanceOf[String], value) } |
||||
config |
||||
} |
||||
|
||||
@Test |
||||
def testDelegationTokenRequests(): Unit = { |
||||
adminClient = org.apache.kafka.clients.admin.AdminClient.create(createAdminConfig) |
||||
val renewer1 = "User:renewer1" |
||||
val renewer2 = "User:renewer2" |
||||
|
||||
// create token1 with renewer1 |
||||
val tokenCreated = DelegationTokenCommand.createToken(adminClient, getCreateOpts(List(renewer1))) |
||||
|
||||
var tokens = DelegationTokenCommand.describeToken(adminClient, getDescribeOpts(List())) |
||||
assertTrue(tokens.size == 1) |
||||
val token1 = tokens.head |
||||
assertEquals(token1, tokenCreated) |
||||
|
||||
// create token2 with renewer2 |
||||
val token2 = DelegationTokenCommand.createToken(adminClient, getCreateOpts(List(renewer2))) |
||||
|
||||
tokens = DelegationTokenCommand.describeToken(adminClient, getDescribeOpts(List())) |
||||
assertTrue(tokens.size == 2) |
||||
assertEquals(Set(token1, token2), tokens.toSet) |
||||
|
||||
//get tokens for renewer2 |
||||
tokens = DelegationTokenCommand.describeToken(adminClient, getDescribeOpts(List(renewer2))) |
||||
assertTrue(tokens.size == 1) |
||||
assertEquals(Set(token2), tokens.toSet) |
||||
|
||||
//test renewing tokens |
||||
val expiryTimestamp = DelegationTokenCommand.renewToken(adminClient, getRenewOpts(token1.hmacAsBase64String())) |
||||
val renewedToken = DelegationTokenCommand.describeToken(adminClient, getDescribeOpts(List(renewer1))).head |
||||
assertEquals(expiryTimestamp, renewedToken.tokenInfo().expiryTimestamp()) |
||||
|
||||
//test expire tokens |
||||
DelegationTokenCommand.expireToken(adminClient, getExpireOpts(token1.hmacAsBase64String())) |
||||
DelegationTokenCommand.expireToken(adminClient, getExpireOpts(token2.hmacAsBase64String())) |
||||
|
||||
tokens = DelegationTokenCommand.describeToken(adminClient, getDescribeOpts(List())) |
||||
assertTrue(tokens.size == 0) |
||||
|
||||
//create token with invalid renewer principal type |
||||
intercept[ExecutionException](DelegationTokenCommand.createToken(adminClient, getCreateOpts(List("Group:Renewer3")))) |
||||
|
||||
// try describing tokens for unknown owner |
||||
assertTrue(DelegationTokenCommand.describeToken(adminClient, getDescribeOpts(List("User:Unknown"))).isEmpty) |
||||
} |
||||
|
||||
private def getCreateOpts(renewers: List[String]): DelegationTokenCommandOptions = { |
||||
val opts = ListBuffer("--bootstrap-server", brokerList, "--max-life-time-period", "-1", |
||||
"--command-config", "testfile", "--create") |
||||
renewers.foreach(renewer => opts ++= ListBuffer("--renewer-principal", renewer)) |
||||
new DelegationTokenCommandOptions(opts.toArray) |
||||
} |
||||
|
||||
private def getDescribeOpts(owners: List[String]): DelegationTokenCommandOptions = { |
||||
val opts = ListBuffer("--bootstrap-server", brokerList, "--command-config", "testfile", "--describe") |
||||
owners.foreach(owner => opts ++= ListBuffer("--owner-principal", owner)) |
||||
new DelegationTokenCommandOptions(opts.toArray) |
||||
} |
||||
|
||||
private def getRenewOpts(hmac: String): DelegationTokenCommandOptions = { |
||||
val opts = Array("--bootstrap-server", brokerList, "--command-config", "testfile", "--renew", |
||||
"--renew-time-period", "-1", |
||||
"--hmac", hmac) |
||||
new DelegationTokenCommandOptions(opts) |
||||
} |
||||
|
||||
private def getExpireOpts(hmac: String): DelegationTokenCommandOptions = { |
||||
val opts = Array("--bootstrap-server", brokerList, "--command-config", "testfile", "--expire", |
||||
"--expiry-time-period", "-1", |
||||
"--hmac", hmac) |
||||
new DelegationTokenCommandOptions(opts) |
||||
} |
||||
|
||||
@After |
||||
override def tearDown(): Unit = { |
||||
if (adminClient != null) |
||||
adminClient.close() |
||||
super.tearDown() |
||||
closeSasl() |
||||
} |
||||
} |
||||
Loading…
Reference in new issue