You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "unknowntpo (via GitHub)" <gi...@apache.org> on 2024/07/22 07:26:59 UTC
[PR] KAFKA-17137: Feat admin client it acl configs [kafka]
unknowntpo opened a new pull request, #16648:
URL: https://github.com/apache/kafka/pull/16648
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on PR #16648:
URL: https://github.com/apache/kafka/pull/16648#issuecomment-2287731646
@chia7712 Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1686724598
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -2432,12 +2628,12 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
}
/**
- * 1. Assume ROOT logger == TRACE
- * 2. Change kafka.controller.KafkaController logger to INFO
- * 3. Unset kafka.controller.KafkaController via AlterConfigOp.OpType.DELETE (resets it to the root logger - TRACE)
- * 4. Change ROOT logger to ERROR
- * 5. Ensure the kafka.controller.KafkaController logger's level is ERROR (the current root logger level)
- */
+ * 1. Assume ROOT logger == TRACE
Review Comment:
please remove unrelated changes
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -2720,11 +2917,11 @@ object PlaintextAdminIntegrationTest {
@nowarn("cat=deprecation")
def checkValidAlterConfigs(
- admin: Admin,
- test: KafkaServerTestHarness,
- topicResource1: ConfigResource,
- topicResource2: ConfigResource
- ): Unit = {
+ admin: Admin,
Review Comment:
ditto
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -2790,9 +2987,9 @@ object PlaintextAdminIntegrationTest {
@nowarn("cat=deprecation")
def checkInvalidAlterConfigs(
- test: KafkaServerTestHarness,
- admin: Admin
- ): Unit = {
+ test: KafkaServerTestHarness,
Review Comment:
ditto
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -6,7 +6,7 @@
* (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
Review Comment:
ditto
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -2500,6 +2696,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
client = createAdminClient
val validLoggerName = "kafka.server.KafkaRequestHandler"
val expectedValidLoggerLogLevel = describeBrokerLoggers().get(validLoggerName)
+
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1723414106
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -127,6 +133,78 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxLifeTimeMs: Int, expiryTimePeriodMs: Int, expectedTokenNum: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxLifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == expectedTokenNum),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ try {
+ // Note that maxTimestamp = token created time + maxLifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1, 1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assertTrue(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
+
+ // expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
+ // expiryTimestamp will be set to min(expiryTimestamp, maxTimestamp),
Review Comment:
My point was there is a bit chaos caused by `expiryTimestamp`. the code used by calculated the `expiryTimestamp` is `min(now + expiryTimePeriodMs, maxTimestamp`)`, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1739520357
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -96,6 +96,50 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDescribeConfigWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ // Describe and broker
+ val brokerResource1 = new ConfigResource(ConfigResource.Type.BROKER, brokers(1).config.brokerId.toString)
+ val brokerResource2 = new ConfigResource(ConfigResource.Type.BROKER, brokers(2).config.brokerId.toString)
+ val configResources = Seq(brokerResource1, brokerResource2)
+
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.describeConfigs(configResources.asJava,new DescribeConfigsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAlterConfigsWithOptionTimeoutMs(quorum: String): Unit = {
+ client = createAdminClient
+ val config = createConfig
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val alterLogLevelsEntries = Seq(
+ new ConfigEntry("kafka.controller.KafkaController", LogLevelConfig.INFO_LOG_LEVEL)
+ ).asJavaCollection
+
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.alterConfigs(
+ Map(brokerLoggerConfigResource -> new Config(alterLogLevelsEntries)).asJava,
+ new AlterConfigsOptions().timeoutMs(0)).all()
Review Comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1717147297
##########
clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java:
##########
@@ -28,6 +28,14 @@
public class ExpireDelegationTokenOptions extends AbstractOptions<ExpireDelegationTokenOptions> {
private long expiryTimePeriodMs = -1L;
+ /**
+ * When {@link Admin#expireDelegationToken(byte[], ExpireDelegationTokenOptions)} method is called this option will control the time of token expiry,
Review Comment:
Could you please describe the behavior of "<0" and ">=0"? the description about `Admin` can be dropped since that is duplicate to the docs of `expireDelegationToken`
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -127,6 +133,79 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ // prepare normal client
+ client = createAdminClient
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+
+ client.createAcls(Collections.singleton(acl)).all().get()
+
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ client.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxLifeTimeMs: Int, expiryTimePeriodMs: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxLifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == 1),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ // Note that maxTimestamp = token created time + maxlifeTimeMs
Review Comment:
I know that is existent typo in our public APIs, but could you please fix the typo in the comment :)
`maxlifeTimeMs` -> `maxLifeTimeMs`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1723342017
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -127,6 +133,78 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxLifeTimeMs: Int, expiryTimePeriodMs: Int, expectedTokenNum: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxLifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == expectedTokenNum),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ try {
+ // Note that maxTimestamp = token created time + maxLifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1, 1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assertTrue(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
+
+ // expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
+ // expiryTimestamp will be set to min(expiryTimestamp, maxTimestamp),
Review Comment:
I think at here, we should have knowledge about `maxTimestamp` because we are doing IT tests, so it think this comment is okay, what do you think ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1703336803
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -94,6 +95,249 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(): Unit = {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ client = createAdminClient
+ val startTimeMs = Time.SYSTEM.milliseconds()
+ assertFutureExceptionTypeEquals(client.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(2)).all(), classOf[TimeoutException])
Review Comment:
this is unstable since it may pass when the test run fast enough. Maybe we can create a `Admin` with incorrect bootstrap and then send this request with timeout.
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -94,6 +95,249 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(): Unit = {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ client = createAdminClient
+ val startTimeMs = Time.SYSTEM.milliseconds()
+ assertFutureExceptionTypeEquals(client.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(2)).all(), classOf[TimeoutException])
+ val endTimeMs = Time.SYSTEM.milliseconds()
+ assertTrue(endTimeMs > startTimeMs, "Expected the timeout to take at least one millisecond.")
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDescribeAclsWithOptionTimeoutMs(): Unit = {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ client = createAdminClient
+ client.createAcls(Collections.singleton(acl)).all()
+ val startTimeMs = Time.SYSTEM.milliseconds()
+ assertFutureExceptionTypeEquals(client.describeAcls(AclBindingFilter.ANY, new DescribeAclsOptions().timeoutMs(2)).values(), classOf[TimeoutException])
+ val endTimeMs = Time.SYSTEM.milliseconds()
+ assertTrue(endTimeMs > startTimeMs, "Expected the timeout to take at least one millisecond.")
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(): Unit = {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ client = createAdminClient
+
+ client.createAcls(Collections.singleton(acl)).all()
+
+ val startTimeMs = Time.SYSTEM.milliseconds()
+ assertFutureExceptionTypeEquals(client.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(2)).all(), classOf[TimeoutException])
+ val endTimeMs = Time.SYSTEM.milliseconds()
+ assertTrue(endTimeMs > startTimeMs, "Expected the timeout to take at least one millisecond.")
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDescribeConfigWithOptionTimeoutMs(): Unit = {
+ client = createAdminClient
+
+ // Create topics
+ val topic1 = "describe-alter-configs-topic-1"
+ val topicResource1 = new ConfigResource(ConfigResource.Type.TOPIC, topic1)
+ val topicConfig1 = new Properties
+ topicConfig1.setProperty(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, "500000")
+ topicConfig1.setProperty(TopicConfig.RETENTION_MS_CONFIG, "60000000")
+ createTopic(topic1, numPartitions = 1, replicationFactor = 1, topicConfig1)
+
+ val topic2 = "describe-alter-configs-topic-2"
+ val topicResource2 = new ConfigResource(ConfigResource.Type.TOPIC, topic2)
+ createTopic(topic2)
+
+ // Describe topics and broker
+ val brokerResource1 = new ConfigResource(ConfigResource.Type.BROKER, brokers(1).config.brokerId.toString)
+ val brokerResource2 = new ConfigResource(ConfigResource.Type.BROKER, brokers(2).config.brokerId.toString)
+ val configResources = Seq(topicResource1, topicResource2, brokerResource1, brokerResource2)
+
+ val startTimeMs = Time.SYSTEM.milliseconds()
+ assertFutureExceptionTypeEquals(
+ client.describeConfigs(configResources.asJava,new DescribeConfigsOptions().timeoutMs(2)).all(),
Review Comment:
ditto
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -94,6 +95,249 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(): Unit = {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ client = createAdminClient
+ val startTimeMs = Time.SYSTEM.milliseconds()
+ assertFutureExceptionTypeEquals(client.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(2)).all(), classOf[TimeoutException])
+ val endTimeMs = Time.SYSTEM.milliseconds()
+ assertTrue(endTimeMs > startTimeMs, "Expected the timeout to take at least one millisecond.")
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDescribeAclsWithOptionTimeoutMs(): Unit = {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ client = createAdminClient
+ client.createAcls(Collections.singleton(acl)).all()
+ val startTimeMs = Time.SYSTEM.milliseconds()
+ assertFutureExceptionTypeEquals(client.describeAcls(AclBindingFilter.ANY, new DescribeAclsOptions().timeoutMs(2)).values(), classOf[TimeoutException])
+ val endTimeMs = Time.SYSTEM.milliseconds()
+ assertTrue(endTimeMs > startTimeMs, "Expected the timeout to take at least one millisecond.")
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(): Unit = {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ client = createAdminClient
+
+ client.createAcls(Collections.singleton(acl)).all()
+
+ val startTimeMs = Time.SYSTEM.milliseconds()
+ assertFutureExceptionTypeEquals(client.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(2)).all(), classOf[TimeoutException])
Review Comment:
ditto
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -94,6 +95,249 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(): Unit = {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ client = createAdminClient
+ val startTimeMs = Time.SYSTEM.milliseconds()
+ assertFutureExceptionTypeEquals(client.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(2)).all(), classOf[TimeoutException])
+ val endTimeMs = Time.SYSTEM.milliseconds()
+ assertTrue(endTimeMs > startTimeMs, "Expected the timeout to take at least one millisecond.")
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDescribeAclsWithOptionTimeoutMs(): Unit = {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ client = createAdminClient
+ client.createAcls(Collections.singleton(acl)).all()
+ val startTimeMs = Time.SYSTEM.milliseconds()
+ assertFutureExceptionTypeEquals(client.describeAcls(AclBindingFilter.ANY, new DescribeAclsOptions().timeoutMs(2)).values(), classOf[TimeoutException])
Review Comment:
ditto
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -94,6 +95,249 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(): Unit = {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ client = createAdminClient
+ val startTimeMs = Time.SYSTEM.milliseconds()
+ assertFutureExceptionTypeEquals(client.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(2)).all(), classOf[TimeoutException])
+ val endTimeMs = Time.SYSTEM.milliseconds()
+ assertTrue(endTimeMs > startTimeMs, "Expected the timeout to take at least one millisecond.")
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDescribeAclsWithOptionTimeoutMs(): Unit = {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ client = createAdminClient
+ client.createAcls(Collections.singleton(acl)).all()
+ val startTimeMs = Time.SYSTEM.milliseconds()
+ assertFutureExceptionTypeEquals(client.describeAcls(AclBindingFilter.ANY, new DescribeAclsOptions().timeoutMs(2)).values(), classOf[TimeoutException])
+ val endTimeMs = Time.SYSTEM.milliseconds()
+ assertTrue(endTimeMs > startTimeMs, "Expected the timeout to take at least one millisecond.")
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(): Unit = {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ client = createAdminClient
+
+ client.createAcls(Collections.singleton(acl)).all()
+
+ val startTimeMs = Time.SYSTEM.milliseconds()
+ assertFutureExceptionTypeEquals(client.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(2)).all(), classOf[TimeoutException])
+ val endTimeMs = Time.SYSTEM.milliseconds()
+ assertTrue(endTimeMs > startTimeMs, "Expected the timeout to take at least one millisecond.")
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDescribeConfigWithOptionTimeoutMs(): Unit = {
+ client = createAdminClient
+
+ // Create topics
+ val topic1 = "describe-alter-configs-topic-1"
+ val topicResource1 = new ConfigResource(ConfigResource.Type.TOPIC, topic1)
+ val topicConfig1 = new Properties
+ topicConfig1.setProperty(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, "500000")
+ topicConfig1.setProperty(TopicConfig.RETENTION_MS_CONFIG, "60000000")
+ createTopic(topic1, numPartitions = 1, replicationFactor = 1, topicConfig1)
+
+ val topic2 = "describe-alter-configs-topic-2"
+ val topicResource2 = new ConfigResource(ConfigResource.Type.TOPIC, topic2)
+ createTopic(topic2)
+
+ // Describe topics and broker
+ val brokerResource1 = new ConfigResource(ConfigResource.Type.BROKER, brokers(1).config.brokerId.toString)
+ val brokerResource2 = new ConfigResource(ConfigResource.Type.BROKER, brokers(2).config.brokerId.toString)
+ val configResources = Seq(topicResource1, topicResource2, brokerResource1, brokerResource2)
+
+ val startTimeMs = Time.SYSTEM.milliseconds()
+ assertFutureExceptionTypeEquals(
+ client.describeConfigs(configResources.asJava,new DescribeConfigsOptions().timeoutMs(2)).all(),
+ classOf[TimeoutException])
+ val endTimeMs = Time.SYSTEM.milliseconds()
+ assertTrue(endTimeMs > startTimeMs, "Expected the timeout to take at least one millisecond.")
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAlterConfigsWithOptionTimeoutMs(): Unit = {
+ val alterLogLevelsEntries = Seq(
+ new ConfigEntry("kafka.controller.KafkaController", LogLevelConfig.INFO_LOG_LEVEL)
+ ).asJavaCollection
+ client = createAdminClient
+
+ val startTimeMs = Time.SYSTEM.milliseconds()
+ assertFutureExceptionTypeEquals(client.alterConfigs(
+ Map(brokerLoggerConfigResource -> new Config(alterLogLevelsEntries)).asJava,
+ new AlterConfigsOptions().timeoutMs(2)).all(),
+ classOf[TimeoutException])
+ val endTimeMs = Time.SYSTEM.milliseconds()
+ assertTrue(endTimeMs > startTimeMs, "Expected the timeout to take at least one millisecond.")
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
+ // Override old configs
+ val props = TestUtils.createBrokerConfigs(brokers.size, zkConnectOrNull)
+ val finalProps = props.map{prop => {
+ prop.put(QuotaConfigs.NUM_QUOTA_SAMPLES_CONFIG, "5")
+ prop.put(QuotaConfigs.CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG, "1")
+ prop
+ }}
+ val conf = finalProps.map(prop => KafkaConfig.fromProps(prop))
+
+ instanceConfigs = conf
+
+ for (i <- 0 until brokerCount) {
+ killBroker(i)
+ }
+ restartDeadBrokers()
+
+ client = createAdminClient
+
+ val brokerIds = brokers.map(_.config.brokerId).toSet
+ brokerIds.foreach(id => TestUtils.waitForOnlineBroker(client, id))
+
+ println(brokers.head.config)
+
+ // fire createPartitions without retry, will fail
+ // add retry config, will suceess
+
+ // def partitions(topic: String, expectedNumPartitionsOpt: Option[Int]): util.List[TopicPartitionInfo] = {
Review Comment:
any updates?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1723483176
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -127,6 +133,78 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxLifeTimeMs: Int, expiryTimePeriodMs: Int, expectedTokenNum: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxLifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == expectedTokenNum),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ try {
+ // Note that maxTimestamp = token created time + maxLifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1, 1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assertTrue(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
+
+ // expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
+ // expiryTimestamp will be set to min(expiryTimestamp, maxTimestamp),
Review Comment:
Got it, `min(now + expiryTimePeriodMs, maxTimestamp)` seems more readable.
I'll update the comment to this:
```
// expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
// expiryTimestamp will be set to min(now + expiryTimePeriodMs, maxTimestamp),
// in this case, maxTimestamp is smaller, so expiryTimestamp will not be modified
```
Is that okay for you ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1724050335
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -127,6 +133,78 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxLifeTimeMs: Int, expiryTimePeriodMs: Int, expectedTokenNum: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxLifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == expectedTokenNum),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ try {
+ // Note that maxTimestamp = token created time + maxLifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1, 1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assertTrue(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
+
+ // expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
+ // expiryTimestamp will be set to min(expiryTimestamp, maxTimestamp),
Review Comment:
Fixed.
https://github.com/apache/kafka/commit/d8a8e5559e50d70584cccff5e780c70e31433237
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1723342017
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -127,6 +133,78 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxLifeTimeMs: Int, expiryTimePeriodMs: Int, expectedTokenNum: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxLifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == expectedTokenNum),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ try {
+ // Note that maxTimestamp = token created time + maxLifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1, 1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assertTrue(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
+
+ // expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
+ // expiryTimestamp will be set to min(expiryTimestamp, maxTimestamp),
Review Comment:
I think at here, we should have knowledge about `maxTimestamp` because we are doing IT tests, and we also need to test if expiryTimestamp is really set to `min(expiryTimestamp, maxTimestamp)`.
So it think this comment is okay, what do you think ?
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -127,6 +133,78 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxLifeTimeMs: Int, expiryTimePeriodMs: Int, expectedTokenNum: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxLifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == expectedTokenNum),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ try {
+ // Note that maxTimestamp = token created time + maxLifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1, 1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assertTrue(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
+
+ // expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
+ // expiryTimestamp will be set to min(expiryTimestamp, maxTimestamp),
Review Comment:
I think at here, we should have knowledge about `maxTimestamp` because we are doing IT tests, and we also need to test if `expiryTimestamp` is really set to `min(expiryTimestamp, maxTimestamp)`.
So it think this comment is okay, what do you think ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1713786391
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -49,12 +52,16 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val kraftAuthorizerClassName = classOf[StandardAuthorizer].getName
val kafkaPrincipal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, JaasTestUtils.KafkaServerPrincipalUnqualifiedName)
var superUserAdmin: Admin = _
+ val secretKey = "secretKey"
override protected def securityProtocol = SecurityProtocol.SASL_SSL
override protected lazy val trustStoreFile = Some(TestUtils.tempFile("truststore", ".jks"))
@BeforeEach
override def setUp(testInfo: TestInfo): Unit = {
+ // set this to use delegation token
+ this.serverConfig.setProperty(DelegationTokenManagerConfigs.DELEGATION_TOKEN_SECRET_KEY_CONFIG, secretKey)
if (TestInfoUtils.isKRaft(testInfo)) {
+ this.serverConfig.setProperty(ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG, kraftAuthorizerClassName)
Review Comment:
removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1721174263
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -126,6 +133,79 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ // prepare normal client
+ client = createAdminClient
Review Comment:
your right, fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1730562364
##########
clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java:
##########
@@ -25,10 +25,19 @@
* The API of this class is evolving, see {@link Admin} for details.
*/
@InterfaceStability.Evolving
-public class ExpireDelegationTokenOptions extends AbstractOptions<ExpireDelegationTokenOptions> {
+public class ExpireDelegationTokenOptions
Review Comment:
Could you please revert this unrelated change?
##########
clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java:
##########
@@ -25,10 +25,19 @@
* The API of this class is evolving, see {@link Admin} for details.
*/
@InterfaceStability.Evolving
-public class ExpireDelegationTokenOptions extends AbstractOptions<ExpireDelegationTokenOptions> {
+public class ExpireDelegationTokenOptions
+ extends AbstractOptions<ExpireDelegationTokenOptions> {
+
private long expiryTimePeriodMs = -1L;
- public ExpireDelegationTokenOptions expiryTimePeriodMs(long expiryTimePeriodMs) {
+ /**
+ * @param expiryTimePeriodMs the time period until we should expire this token.
+ * {@code expiryTimePeriodMs} >= 0: the token will update the `expiration timestamp` if the current expiration timestamp is small than (now + expiryTimePeriodMs).
Review Comment:
Sorry that could you please revise it to "{@code expiryTimePeriodMs} >= 0: the token will update the `expiration timestamp` to min(now + expiryTimePeriodMs, maxTimestamp)`
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -106,19 +150,22 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
config.put(AdminClientConfig.CLIENT_ID_CONFIG, clientId)
client = Admin.create(config)
- val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.CLIENT_ID -> clientId).asJava)
- val configEntries = Map(QuotaConfigs.CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG -> 1.0, QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG -> 3.0)
- client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, configEntries.map {case (k, v) =>
- new ClientQuotaAlteration.Op(k,v)}.asJavaCollection)).asJavaCollection).all.get
+ try {
+ val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.CLIENT_ID -> clientId).asJava)
+ val configEntries = Map(QuotaConfigs.CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG -> 1.0, QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG -> 3.0)
+ client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, configEntries.map { case (k, v) =>
+ new ClientQuotaAlteration.Op(k, v)
+ }.asJavaCollection)).asJavaCollection).all.get
- TestUtils.waitUntilTrue(() => {
- // wait for our ClientQuotaEntity to be set
- client.describeClientQuotas(ClientQuotaFilter.all()).entities().get().size == 1
- }, "Timed out waiting for quota config to be propagated to all servers")
+ TestUtils.waitUntilTrue(() => {
+ // wait for our ClientQuotaEntity to be set
+ client.describeClientQuotas(ClientQuotaFilter.all()).entities().get().size == 1
+ }, "Timed out waiting for quota config to be propagated to all servers")
- val quotaEntities = client.describeClientQuotas(ClientQuotaFilter.all()).entities().get()
+ val quotaEntities = client.describeClientQuotas(ClientQuotaFilter.all()).entities().get()
- assertEquals(configEntries,quotaEntities.get(entity).asScala)
+ assertEquals(configEntries, quotaEntities.get(entity).asScala)
+ } finally client.close(time.Duration.ZERO)
Review Comment:
this client is created normally (correct port), so it can be closed as usual. that means we don't need to add try-finally, as it will be closed by the `@AfterEach`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1713742230
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -126,6 +133,79 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ // prepare normal client
+ client = createAdminClient
Review Comment:
> The connection of brokenClient is never successful, right?
That's true, but this is intended, I need to use this brokenClient to test `deleteAcls` with timeout, since it never connected, so we will certainly get `TimeoutException`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1718663810
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -96,6 +96,62 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDescribeConfigWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
Review Comment:
please use try-finally to release this resource
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -96,6 +96,62 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDescribeConfigWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ client = createAdminClient
+
+ // Create topics
Review Comment:
The purpose of this test case is used to make sure timeout works well, so we can leverage existent configs (broker-level). Hence, could you please remove those topic creation?
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -126,6 +133,79 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ // prepare normal client
+ client = createAdminClient
Review Comment:
yes, so we don't need to have `client` and create ACLs, right?
##########
clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java:
##########
@@ -28,6 +28,13 @@
public class ExpireDelegationTokenOptions extends AbstractOptions<ExpireDelegationTokenOptions> {
private long expiryTimePeriodMs = -1L;
+ /**
+ * @param expiryTimePeriodMs the time period until we should expire this token.
+ * {@code expiryTimestamp} will be set to {@code min(expiryTimestamp, maxTimestamp)}
Review Comment:
line#33 seems to be duplicate. could you please remove it
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -127,6 +133,79 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
Review Comment:
ditto
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -96,6 +96,62 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDescribeConfigWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ client = createAdminClient
+
+ // Create topics
+ val topic1 = "describe-alter-configs-topic-1"
+ val topicResource1 = new ConfigResource(ConfigResource.Type.TOPIC, topic1)
+ val topicConfig1 = new Properties
+ topicConfig1.setProperty(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, "500000")
+ topicConfig1.setProperty(TopicConfig.RETENTION_MS_CONFIG, "60000000")
+ createTopic(topic1)
+
+ val topic2 = "describe-alter-configs-topic-2"
+ val topicResource2 = new ConfigResource(ConfigResource.Type.TOPIC, topic2)
+ createTopic(topic2)
+
+ // Describe topics and broker
+ val brokerResource1 = new ConfigResource(ConfigResource.Type.BROKER, brokers(1).config.brokerId.toString)
+ val brokerResource2 = new ConfigResource(ConfigResource.Type.BROKER, brokers(2).config.brokerId.toString)
+ val configResources = Seq(topicResource1, topicResource2, brokerResource1, brokerResource2)
+
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.describeConfigs(configResources.asJava,new DescribeConfigsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAlterConfigsWithOptionTimeoutMs(quorum: String): Unit = {
+ client = createAdminClient
+ val config = createConfig
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ val alterLogLevelsEntries = Seq(
+ new ConfigEntry("kafka.controller.KafkaController", LogLevelConfig.INFO_LOG_LEVEL)
+ ).asJavaCollection
+
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.alterConfigs(
+ Map(brokerLoggerConfigResource -> new Config(alterLogLevelsEntries)).asJava,
+ new AlterConfigsOptions().timeoutMs(0)).all()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
Review Comment:
ditto. try-finally
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1723862476
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -127,6 +133,78 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxLifeTimeMs: Int, expiryTimePeriodMs: Int, expectedTokenNum: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxLifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == expectedTokenNum),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ try {
+ // Note that maxTimestamp = token created time + maxLifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1, 1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assertTrue(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
+
+ // expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
+ // expiryTimestamp will be set to min(expiryTimestamp, maxTimestamp),
Review Comment:
that is good, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #16648:
URL: https://github.com/apache/kafka/pull/16648#issuecomment-2310293011
```
* What went wrong:
Execution failed for task ':clients:javadoc'.
> Javadoc generation failed. Generated Javadoc options file (useful for troubleshooting): '/home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-16648/clients/build/tmp/javadoc/javadoc.options'
```
@unknowntpo Could you please check the error?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1739046762
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -96,6 +96,50 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDescribeConfigWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ // Describe and broker
+ val brokerResource1 = new ConfigResource(ConfigResource.Type.BROKER, brokers(1).config.brokerId.toString)
+ val brokerResource2 = new ConfigResource(ConfigResource.Type.BROKER, brokers(2).config.brokerId.toString)
+ val configResources = Seq(brokerResource1, brokerResource2)
+
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.describeConfigs(configResources.asJava,new DescribeConfigsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAlterConfigsWithOptionTimeoutMs(quorum: String): Unit = {
+ client = createAdminClient
+ val config = createConfig
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val alterLogLevelsEntries = Seq(
+ new ConfigEntry("kafka.controller.KafkaController", LogLevelConfig.INFO_LOG_LEVEL)
+ ).asJavaCollection
+
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.alterConfigs(
+ Map(brokerLoggerConfigResource -> new Config(alterLogLevelsEntries)).asJava,
+ new AlterConfigsOptions().timeoutMs(0)).all()
Review Comment:
please call `.get()`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1713786866
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -126,6 +133,79 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ // prepare normal client
+ client = createAdminClient
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+
+ client.createAcls(Collections.singleton(acl)).all().get()
+
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ client.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxlifeTimeMs: Int, expiryTimePeriodMs: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxlifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == 1),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ // Note that maxTimestamp = token created time + maxlifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assert(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
+
+ // expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
Review Comment:
updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on PR #16648:
URL: https://github.com/apache/kafka/pull/16648#issuecomment-2295603023
@chia7712 Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on PR #16648:
URL: https://github.com/apache/kafka/pull/16648#issuecomment-2278926689
> @unknowntpo could you please fix the build error
Okay, fixed, but right now, lots of tests failed, I'll try to fix it.
https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-16648/11/testReport/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1709534075
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -6,7 +6,7 @@
* (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
Review Comment:
I think it's because I use the wrong formatter in `Intellij`, what is the recommanded way of formatting Kafka code in `Intellij` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1712682066
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -49,12 +52,16 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val kraftAuthorizerClassName = classOf[StandardAuthorizer].getName
val kafkaPrincipal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, JaasTestUtils.KafkaServerPrincipalUnqualifiedName)
var superUserAdmin: Admin = _
+ val secretKey = "secretKey"
override protected def securityProtocol = SecurityProtocol.SASL_SSL
override protected lazy val trustStoreFile = Some(TestUtils.tempFile("truststore", ".jks"))
@BeforeEach
override def setUp(testInfo: TestInfo): Unit = {
+ // set this to use delegation token
+ this.serverConfig.setProperty(DelegationTokenManagerConfigs.DELEGATION_TOKEN_SECRET_KEY_CONFIG, secretKey)
if (TestInfoUtils.isKRaft(testInfo)) {
+ this.serverConfig.setProperty(ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG, kraftAuthorizerClassName)
Review Comment:
duplicate?
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -126,6 +133,79 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ // prepare normal client
+ client = createAdminClient
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+
+ client.createAcls(Collections.singleton(acl)).all().get()
+
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ client.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxlifeTimeMs: Int, expiryTimePeriodMs: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxlifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == 1),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ // Note that maxTimestamp = token created time + maxlifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assert(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
+
+ // expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
Review Comment:
nice comment! Could you please update docs of APIs?
https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java#L31
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -126,6 +133,79 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ // prepare normal client
+ client = createAdminClient
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+
+ client.createAcls(Collections.singleton(acl)).all().get()
+
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ client.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxlifeTimeMs: Int, expiryTimePeriodMs: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
Review Comment:
`maxlifeTimeMs` -> `maxLifeTimeMs`
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -126,6 +133,79 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ // prepare normal client
+ client = createAdminClient
Review Comment:
The connection of `brokenClient` is never successful, right? so we don't use this `client`
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -126,6 +133,79 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ // prepare normal client
+ client = createAdminClient
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+
+ client.createAcls(Collections.singleton(acl)).all().get()
+
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ client.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxlifeTimeMs: Int, expiryTimePeriodMs: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxlifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == 1),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ // Note that maxTimestamp = token created time + maxlifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assert(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
Review Comment:
`assertTrue`
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -126,6 +133,79 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ // prepare normal client
+ client = createAdminClient
+
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+
+ client.createAcls(Collections.singleton(acl)).all().get()
+
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ brokenClient.close(time.Duration.ZERO)
+ client.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxlifeTimeMs: Int, expiryTimePeriodMs: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxlifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == 1),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ // Note that maxTimestamp = token created time + maxlifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assert(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
+
+ // expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
+ // expiryTimePeriodMs will be set to min(expiryTimestamp, maxTimestamp),
+ // in this case, maxTimestamp is smaller, so expiryTimestamp will not be modified
+ val (createResult2, expireResult2) = generateTokenResult(5000, 100000)
+ assert(createResult2.delegationToken().get().tokenInfo().expiryTimestamp() == expireResult2.expiryTimestamp().get())
Review Comment:
`assertTrue`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #16648:
URL: https://github.com/apache/kafka/pull/16648
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1723342017
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -127,6 +133,78 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxLifeTimeMs: Int, expiryTimePeriodMs: Int, expectedTokenNum: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxLifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == expectedTokenNum),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ try {
+ // Note that maxTimestamp = token created time + maxLifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1, 1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assertTrue(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
+
+ // expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
+ // expiryTimestamp will be set to min(expiryTimestamp, maxTimestamp),
Review Comment:
I think at here, we should have knowledge about `maxTimestamp` because we are doing IT tests, and we needs to test if expiryTimestamp is really set to `min(expiryTimestamp, maxTimestamp)`.
So it think this comment is okay, what do you think ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1723339500
##########
clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java:
##########
@@ -28,6 +28,13 @@
public class ExpireDelegationTokenOptions extends AbstractOptions<ExpireDelegationTokenOptions> {
private long expiryTimePeriodMs = -1L;
+ /**
+ * @param expiryTimePeriodMs the time period until we should expire this token.
+ * {@code expiryTimestamp} is the time we actually expire the token.
+ * If {@code expiryTimePeriodMs} < 0, token will be expired immediately.
+ * If {@code expiryTimePeriodMs} >= 0 and {@code expiryTimestamp} is not due,
+ * {@code expiryTimestamp} will be set to {@code min(expiryTimestamp, maxTimestamp)}.
Review Comment:
okay, I'll change it to
```
/**
* @param expiryTimePeriodMs the time period until we should expire this token.
* {@code expiryTimePeriodMs} >= 0: the token will update the `expiration timestamp` if the current expiration timestamp is small than (now + expiryTimePeriodMs).
* {@code expiryTimePeriodMs} < 0: token will be expired immediately.
*/
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1722203370
##########
clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java:
##########
@@ -28,6 +28,13 @@
public class ExpireDelegationTokenOptions extends AbstractOptions<ExpireDelegationTokenOptions> {
private long expiryTimePeriodMs = -1L;
+ /**
+ * @param expiryTimePeriodMs the time period until we should expire this token.
+ * {@code expiryTimestamp} is the time we actually expire the token.
+ * If {@code expiryTimePeriodMs} < 0, token will be expired immediately.
+ * If {@code expiryTimePeriodMs} >= 0 and {@code expiryTimestamp} is not due,
+ * {@code expiryTimestamp} will be set to {@code min(expiryTimestamp, maxTimestamp)}.
Review Comment:
I feel `min(expiryTimestamp, maxTimestamp)` is a bit puzzling, since the public APIs don't mention the `maxTimestamp`. How about:
```
{@code expiryTimePeriodMs} >= 0: the token will update the `expiration timestamp` if the current expiration timestamp is small than (now + expiryTimePeriodMs)
```
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -505,7 +583,7 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
assertNotEquals(Uuid.ZERO_UUID, createResult.topicId(topic1).get())
assertEquals(topicIds(topic1), createResult.topicId(topic1).get())
assertFutureExceptionTypeEquals(createResult.topicId(topic2), classOf[TopicAuthorizationException])
-
+
Review Comment:
please remove this unrelated change
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -127,6 +133,78 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxLifeTimeMs: Int, expiryTimePeriodMs: Int, expectedTokenNum: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxLifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == expectedTokenNum),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ try {
+ // Note that maxTimestamp = token created time + maxLifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1, 1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assertTrue(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
+
+ // expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
+ // expiryTimestamp will be set to min(expiryTimestamp, maxTimestamp),
Review Comment:
ditto. please check the bove comment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1733726177
##########
clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java:
##########
@@ -28,7 +28,14 @@
public class ExpireDelegationTokenOptions extends AbstractOptions<ExpireDelegationTokenOptions> {
private long expiryTimePeriodMs = -1L;
- public ExpireDelegationTokenOptions expiryTimePeriodMs(long expiryTimePeriodMs) {
+ /**
+ * @param expiryTimePeriodMs the time period until we should expire this token.
+ * {@code expiryTimePeriodMs} >= 0: the token will update the expiration timestamp to min(now + expiryTimePeriodMs, maxTimestamp)
Review Comment:
fixed.
##########
clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java:
##########
@@ -28,7 +28,14 @@
public class ExpireDelegationTokenOptions extends AbstractOptions<ExpireDelegationTokenOptions> {
private long expiryTimePeriodMs = -1L;
- public ExpireDelegationTokenOptions expiryTimePeriodMs(long expiryTimePeriodMs) {
+ /**
+ * @param expiryTimePeriodMs the time period until we should expire this token.
+ * {@code expiryTimePeriodMs} >= 0: the token will update the expiration timestamp to min(now + expiryTimePeriodMs, maxTimestamp)
+ * {@code expiryTimePeriodMs} < 0: token will be expired immediately.
Review Comment:
fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1731316019
##########
clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java:
##########
@@ -28,7 +28,14 @@
public class ExpireDelegationTokenOptions extends AbstractOptions<ExpireDelegationTokenOptions> {
private long expiryTimePeriodMs = -1L;
- public ExpireDelegationTokenOptions expiryTimePeriodMs(long expiryTimePeriodMs) {
+ /**
+ * @param expiryTimePeriodMs the time period until we should expire this token.
+ * {@code expiryTimePeriodMs} >= 0: the token will update the expiration timestamp to min(now + expiryTimePeriodMs, maxTimestamp)
Review Comment:
please replace `>` by `>`
##########
clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java:
##########
@@ -28,7 +28,14 @@
public class ExpireDelegationTokenOptions extends AbstractOptions<ExpireDelegationTokenOptions> {
private long expiryTimePeriodMs = -1L;
- public ExpireDelegationTokenOptions expiryTimePeriodMs(long expiryTimePeriodMs) {
+ /**
+ * @param expiryTimePeriodMs the time period until we should expire this token.
+ * {@code expiryTimePeriodMs} >= 0: the token will update the expiration timestamp to min(now + expiryTimePeriodMs, maxTimestamp)
+ * {@code expiryTimePeriodMs} < 0: token will be expired immediately.
Review Comment:
please replace `<` by `<`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #16648:
URL: https://github.com/apache/kafka/pull/16648#issuecomment-2276085484
@unknowntpo could you please fix the build error
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1717638246
##########
clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java:
##########
@@ -28,6 +28,14 @@
public class ExpireDelegationTokenOptions extends AbstractOptions<ExpireDelegationTokenOptions> {
private long expiryTimePeriodMs = -1L;
+ /**
+ * When {@link Admin#expireDelegationToken(byte[], ExpireDelegationTokenOptions)} method is called this option will control the time of token expiry,
Review Comment:
I put the description of this behavior inside `@param`, is that ok ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1721197831
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -127,6 +133,74 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxLifeTimeMs: Int, expiryTimePeriodMs: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxLifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == 1),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ try {
+ // Note that maxTimestamp = token created time + maxLifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assertTrue(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
+
+ // expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
+ // expiryTimestamp will be set to min(expiryTimestamp, maxTimestamp),
+ // in this case, maxTimestamp is smaller, so expiryTimestamp will not be modified
+ val (createResult2, expireResult2) = generateTokenResult(5000, 100000)
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1721184098
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -127,6 +133,74 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxLifeTimeMs: Int, expiryTimePeriodMs: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxLifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == 1),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ try {
+ // Note that maxTimestamp = token created time + maxLifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assertTrue(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
+
+ // expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
+ // expiryTimestamp will be set to min(expiryTimestamp, maxTimestamp),
+ // in this case, maxTimestamp is smaller, so expiryTimestamp will not be modified
+ val (createResult2, expireResult2) = generateTokenResult(5000, 100000)
Review Comment:
Could you please add IT to test `generateTokenResult(5000, 2000)` to prove `min(expiryTimestamp, maxTimestamp)`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #16648:
URL: https://github.com/apache/kafka/pull/16648#issuecomment-2287099565
@unknowntpo could you please fix the conflicts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1730699172
##########
clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java:
##########
@@ -25,10 +25,19 @@
* The API of this class is evolving, see {@link Admin} for details.
*/
@InterfaceStability.Evolving
-public class ExpireDelegationTokenOptions extends AbstractOptions<ExpireDelegationTokenOptions> {
+public class ExpireDelegationTokenOptions
Review Comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1730698924
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -106,19 +150,22 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
config.put(AdminClientConfig.CLIENT_ID_CONFIG, clientId)
client = Admin.create(config)
- val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.CLIENT_ID -> clientId).asJava)
- val configEntries = Map(QuotaConfigs.CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG -> 1.0, QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG -> 3.0)
- client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, configEntries.map {case (k, v) =>
- new ClientQuotaAlteration.Op(k,v)}.asJavaCollection)).asJavaCollection).all.get
+ try {
+ val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.CLIENT_ID -> clientId).asJava)
+ val configEntries = Map(QuotaConfigs.CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG -> 1.0, QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG -> 3.0)
+ client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, configEntries.map { case (k, v) =>
+ new ClientQuotaAlteration.Op(k, v)
+ }.asJavaCollection)).asJavaCollection).all.get
- TestUtils.waitUntilTrue(() => {
- // wait for our ClientQuotaEntity to be set
- client.describeClientQuotas(ClientQuotaFilter.all()).entities().get().size == 1
- }, "Timed out waiting for quota config to be propagated to all servers")
+ TestUtils.waitUntilTrue(() => {
+ // wait for our ClientQuotaEntity to be set
+ client.describeClientQuotas(ClientQuotaFilter.all()).entities().get().size == 1
+ }, "Timed out waiting for quota config to be propagated to all servers")
- val quotaEntities = client.describeClientQuotas(ClientQuotaFilter.all()).entities().get()
+ val quotaEntities = client.describeClientQuotas(ClientQuotaFilter.all()).entities().get()
- assertEquals(configEntries,quotaEntities.get(entity).asScala)
+ assertEquals(configEntries, quotaEntities.get(entity).asScala)
+ } finally client.close(time.Duration.ZERO)
Review Comment:
ok, done.
##########
clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java:
##########
@@ -25,10 +25,19 @@
* The API of this class is evolving, see {@link Admin} for details.
*/
@InterfaceStability.Evolving
-public class ExpireDelegationTokenOptions extends AbstractOptions<ExpireDelegationTokenOptions> {
+public class ExpireDelegationTokenOptions
+ extends AbstractOptions<ExpireDelegationTokenOptions> {
+
private long expiryTimePeriodMs = -1L;
- public ExpireDelegationTokenOptions expiryTimePeriodMs(long expiryTimePeriodMs) {
+ /**
+ * @param expiryTimePeriodMs the time period until we should expire this token.
+ * {@code expiryTimePeriodMs} >= 0: the token will update the `expiration timestamp` if the current expiration timestamp is small than (now + expiryTimePeriodMs).
Review Comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17137: Feat admin client it acl configs [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16648:
URL: https://github.com/apache/kafka/pull/16648#discussion_r1723483176
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -127,6 +133,78 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testAclOperationsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val acl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", PatternType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.createAcls(Collections.singleton(acl), new CreateAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @Timeout(30)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testDeleteAclsWithOptionTimeoutMs(quorum: String): Unit = {
+ val config = createConfig
+ // this will cause timeout connecting to broker
+ config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, s"localhost:${TestUtils.IncorrectBrokerPort}")
+ val brokenClient = Admin.create(config)
+
+ try {
+ val exception = assertThrows(classOf[ExecutionException], () => {
+ brokenClient.deleteAcls(Collections.singleton(AclBindingFilter.ANY), new DeleteAclsOptions().timeoutMs(0)).all().get()
+ })
+ assertInstanceOf(classOf[TimeoutException], exception.getCause)
+ } finally brokenClient.close(time.Duration.ZERO)
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk","kraft"))
+ def testExpireDelegationTokenWithOptionExpireTimePeriodMs(quorum: String): Unit = {
+ client = createAdminClient
+ val renewer = List(SecurityUtils.parseKafkaPrincipal("User:renewer"))
+
+ def generateTokenResult(maxLifeTimeMs: Int, expiryTimePeriodMs: Int, expectedTokenNum: Int): (CreateDelegationTokenResult, ExpireDelegationTokenResult) = {
+ val createResult = client.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer.asJava).maxlifeTimeMs(maxLifeTimeMs))
+ val tokenCreated = createResult.delegationToken.get
+ TestUtils.waitUntilTrue(() => brokers.forall(server => server.tokenCache.tokens().size() == expectedTokenNum),
+ "Timed out waiting for token to propagate to all servers")
+ val expireResult = client.expireDelegationToken(
+ tokenCreated.hmac(),
+ new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)
+ )
+ (createResult, expireResult)
+ }
+
+ try {
+ // Note that maxTimestamp = token created time + maxLifeTimeMs
+ val (createResult1, expireResult1) = generateTokenResult(10000, -1, 1)
+ // if expiryTimePeriodMs < 0, token will be expired immediately.
+ assertTrue(createResult1.delegationToken().get().tokenInfo().maxTimestamp() > expireResult1.expiryTimestamp().get())
+
+ // expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
+ // expiryTimestamp will be set to min(expiryTimestamp, maxTimestamp),
Review Comment:
Got it, I'll update the comment to this:
```
// expireDelegationToken will decrease the value of expiryTimestamp, since this token is not expired,
// expiryTimestamp will be set to min(now + expiryTimePeriodMs, maxTimestamp),
// in this case, maxTimestamp is smaller, so expiryTimestamp will not be modified
```
Is that okay for you ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org