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