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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
