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/25 00:06:29 UTC
[PR] KAFKA-17137: feat(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
unknowntpo opened a new pull request, #16687:
URL: https://github.com/apache/kafka/pull/16687
We set `controller_mutation_rate` to a very low value, so when calling `createPartition` for about 30 times simultaneously, lots of `ThrottlingQuotaExceededException` should be returned.
And after that, we set `retryOnQuotaViolation` to `true` on `CreatePartitionsOptions`, and try again, this time, we should get no `ThrottlingQuotaExceededException` exceptions.
### Committer Checklist (excluded from commit message)
- [x] 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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16687:
URL: https://github.com/apache/kafka/pull/16687#discussion_r1703505096
##########
core/src/test/scala/integration/kafka/api/BaseAdminIntegrationTest.scala:
##########
@@ -230,6 +232,7 @@ abstract class BaseAdminIntegrationTest extends IntegrationTestHarness with Logg
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers())
config.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "20000")
config.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "40000")
+ config.put(AdminClientConfig.CLIENT_ID_CONFIG, clientId)
Review Comment:
ok, 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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16687:
URL: https://github.com/apache/kafka/pull/16687#discussion_r1697602834
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -94,6 +94,59 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
+ client = createAdminClient
+
+ val brokerIds = brokers.map(_.config.brokerId).toSet
+ brokerIds.foreach(id => TestUtils.waitForOnlineBroker(client, id))
+
+ val topicNums = 30
+
+ def makeTopics(): java.util.List[String] = {
+ val topics = for (i <- 0 until topicNums) yield {
+ val topicName = s"topic-$i"
+ createTopic(topicName)
+ topicName
+ }
+ topics.toList.asJava
+ }
+
+ val topics = makeTopics()
+
+ waitForTopics(client, topics.asScala.toSeq, List())
+ ensureConsistentKRaftMetadata()
+
+ val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.CLIENT_ID -> "test-client-id").asJava)
+
+ val quotas = Seq(new ClientQuotaAlteration.Op(QuotaConfigs.CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG, 1))
+ val alterResult = client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, quotas.asJava)).asJava)
+ alterResult.all.get
+
+ def makeCreatePartitionCalls(topics: java.util.List[String], numPartitions: Int, retry: Boolean): java.util.Collection[KafkaFuture[Void]] = {
+ val futures = topics.asScala.map{
+ topic => client.createPartitions(Map(topic -> NewPartitions.increaseTo(numPartitions)).asJava, new CreatePartitionsOptions().timeoutMs(30000).retryOnQuotaViolation(retry)).all()
+ }
+ futures.toList.asJavaCollection
+ }
+
+ def countFailedFutures(futures: java.util.Collection[KafkaFuture[Void]]): Int = {
+ futures.asScala.count {future =>
+ Try(future.get()) match {
+ case Failure(e: ExecutionException) if e.getCause.isInstanceOf[ThrottlingQuotaExceededException] => true
+ case _ => false
+ }
+ }
+ }
+
+ val failedFuture = makeCreatePartitionCalls(topics, 3, retry = false)
Review Comment:
The quota tests are almost time-based, and that is not stable in our CI. Maybe we can just verify the "settings". For example:
1. add quota
2. describe quota to make sure the quota are updated
3. delete quota
4. describe quota to make sure the quota are deleted
WDYT?
--
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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16687:
URL: https://github.com/apache/kafka/pull/16687#discussion_r1690608224
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -94,6 +94,59 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
+ client = createAdminClient
+
+ val brokerIds = brokers.map(_.config.brokerId).toSet
+ brokerIds.foreach(id => TestUtils.waitForOnlineBroker(client, id))
+
+ val topicNums = 30
+
+ def makeTopics(): java.util.List[String] = {
+ val topics = for (i <- 0 until topicNums) yield {
+ val topicName = s"topic-$i"
+ createTopic(topicName)
+ topicName
+ }
+ topics.toList.asJava
+ }
+
+ val topics = makeTopics()
+
+ waitForTopics(client, topics.asScala.toSeq, List())
+ ensureConsistentKRaftMetadata()
+
+ val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.CLIENT_ID -> "test-client-id").asJava)
+
+ val quotas = Seq(new ClientQuotaAlteration.Op(QuotaConfigs.CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG, 1))
+ val alterResult = client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, quotas.asJava)).asJava)
+ alterResult.all.get
+
+ def makeCreatePartitionCalls(topics: java.util.List[String], numPartitions: Int, retry: Boolean): java.util.Collection[KafkaFuture[Void]] = {
+ val futures = topics.asScala.map{
+ topic => client.createPartitions(Map(topic -> NewPartitions.increaseTo(numPartitions)).asJava, new CreatePartitionsOptions().timeoutMs(30000).retryOnQuotaViolation(retry)).all()
+ }
+ futures.toList.asJavaCollection
+ }
+
+ def countFailedFutures(futures: java.util.Collection[KafkaFuture[Void]]): Int = {
+ futures.asScala.count {future =>
+ Try(future.get()) match {
+ case Failure(e: ExecutionException) if e.getCause.isInstanceOf[ThrottlingQuotaExceededException] => true
+ case _ => false
+ }
+ }
+ }
+
+ val failedFuture = makeCreatePartitionCalls(topics, 3, retry = false)
Review Comment:
Actually, I got lots of `TimeoutException` here.
--
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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on PR #16687:
URL: https://github.com/apache/kafka/pull/16687#issuecomment-2249095033
@chia7712 Would you like to review this PR ?
--
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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on PR #16687:
URL: https://github.com/apache/kafka/pull/16687#issuecomment-2276959593
> @unknowntpo could you please fix the conflicts?
Ok, 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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #16687:
URL: https://github.com/apache/kafka/pull/16687
--
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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #16687:
URL: https://github.com/apache/kafka/pull/16687#issuecomment-2276326743
@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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16687:
URL: https://github.com/apache/kafka/pull/16687#discussion_r1712922851
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -95,6 +96,35 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
+ // Since it's hard to stably reach quota limit in integration test, we only verify quota configs are set correctly
+ val config = createConfig
+ val clientId = "test-client-id"
+
+ config.put(AdminClientConfig.CLIENT_ID_CONFIG, clientId)
+ client = AdminClientTestUtils.create(config, null)
+
+ 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)
+ val quotas = configEntries.map {case (k, v) => new ClientQuotaAlteration.Op(k,v)}
Review Comment:
Elegant!
--
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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16687:
URL: https://github.com/apache/kafka/pull/16687#discussion_r1703045072
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -94,6 +94,59 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
+ client = createAdminClient
+
+ val brokerIds = brokers.map(_.config.brokerId).toSet
+ brokerIds.foreach(id => TestUtils.waitForOnlineBroker(client, id))
+
+ val topicNums = 30
+
+ def makeTopics(): java.util.List[String] = {
+ val topics = for (i <- 0 until topicNums) yield {
+ val topicName = s"topic-$i"
+ createTopic(topicName)
+ topicName
+ }
+ topics.toList.asJava
+ }
+
+ val topics = makeTopics()
+
+ waitForTopics(client, topics.asScala.toSeq, List())
+ ensureConsistentKRaftMetadata()
+
+ val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.CLIENT_ID -> "test-client-id").asJava)
+
+ val quotas = Seq(new ClientQuotaAlteration.Op(QuotaConfigs.CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG, 1))
+ val alterResult = client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, quotas.asJava)).asJava)
+ alterResult.all.get
+
+ def makeCreatePartitionCalls(topics: java.util.List[String], numPartitions: Int, retry: Boolean): java.util.Collection[KafkaFuture[Void]] = {
+ val futures = topics.asScala.map{
+ topic => client.createPartitions(Map(topic -> NewPartitions.increaseTo(numPartitions)).asJava, new CreatePartitionsOptions().timeoutMs(30000).retryOnQuotaViolation(retry)).all()
+ }
+ futures.toList.asJavaCollection
+ }
+
+ def countFailedFutures(futures: java.util.Collection[KafkaFuture[Void]]): Int = {
+ futures.asScala.count {future =>
+ Try(future.get()) match {
+ case Failure(e: ExecutionException) if e.getCause.isInstanceOf[ThrottlingQuotaExceededException] => true
+ case _ => false
+ }
+ }
+ }
+
+ val failedFuture = makeCreatePartitionCalls(topics, 3, retry = false)
Review Comment:
@chia7712 done, please take a look.
--
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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16687:
URL: https://github.com/apache/kafka/pull/16687#discussion_r1712922948
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -95,6 +96,35 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
+ // Since it's hard to stably reach quota limit in integration test, we only verify quota configs are set correctly
+ val config = createConfig
+ val clientId = "test-client-id"
+
+ config.put(AdminClientConfig.CLIENT_ID_CONFIG, clientId)
+ client = AdminClientTestUtils.create(config, null)
+
+ 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)
+ val quotas = configEntries.map {case (k, v) => new ClientQuotaAlteration.Op(k,v)}
+ val alterResult = client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, quotas.asJavaCollection)).asJavaCollection)
+ alterResult.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")
+
+ val describeResult = client.describeClientQuotas(ClientQuotaFilter.all())
+ val quotaEntities = describeResult.entities().get()
+ assertTrue(quotaEntities.containsKey(entity))
+ assertEquals(configEntries,quotaEntities.get(entity).asScala)
Review Comment:
Seems assertTrue() is redundant.
--
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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16687:
URL: https://github.com/apache/kafka/pull/16687#discussion_r1690609481
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -94,6 +94,59 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
+ client = createAdminClient
+
+ val brokerIds = brokers.map(_.config.brokerId).toSet
+ brokerIds.foreach(id => TestUtils.waitForOnlineBroker(client, id))
+
+ val topicNums = 30
+
+ def makeTopics(): java.util.List[String] = {
+ val topics = for (i <- 0 until topicNums) yield {
+ val topicName = s"topic-$i"
+ createTopic(topicName)
+ topicName
+ }
+ topics.toList.asJava
+ }
+
+ val topics = makeTopics()
+
+ waitForTopics(client, topics.asScala.toSeq, List())
+ ensureConsistentKRaftMetadata()
+
+ val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.CLIENT_ID -> "test-client-id").asJava)
+
+ val quotas = Seq(new ClientQuotaAlteration.Op(QuotaConfigs.CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG, 1))
+ val alterResult = client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, quotas.asJava)).asJava)
+ alterResult.all.get
+
+ def makeCreatePartitionCalls(topics: java.util.List[String], numPartitions: Int, retry: Boolean): java.util.Collection[KafkaFuture[Void]] = {
+ val futures = topics.asScala.map{
+ topic => client.createPartitions(Map(topic -> NewPartitions.increaseTo(numPartitions)).asJava, new CreatePartitionsOptions().timeoutMs(30000).retryOnQuotaViolation(retry)).all()
+ }
+ futures.toList.asJavaCollection
+ }
+
+ def countFailedFutures(futures: java.util.Collection[KafkaFuture[Void]]): Int = {
+ futures.asScala.count {future =>
+ Try(future.get()) match {
+ case Failure(e: ExecutionException) if e.getCause.isInstanceOf[ThrottlingQuotaExceededException] => true
+ case _ => false
+ }
+ }
+ }
+
+ val failedFuture = makeCreatePartitionCalls(topics, 3, retry = false)
+ assert(countFailedFutures(failedFuture) > 0, s"Expected at least one future to fail, but none failed")
+
+ val successFutures = makeCreatePartitionCalls(topics, 3, retry = true)
Review Comment:
Actually, I got lots of `TimeoutException` here, and when call `makeCreatePartitionCalls` with `numPartitions = 6` and `retry = true`, I still got some `ThrottlingQuotaExceededException`s.
--
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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16687:
URL: https://github.com/apache/kafka/pull/16687#discussion_r1703061733
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -94,6 +95,26 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
+ // Since it's hard to stably reach quota limit in integration test, we only verify quota configs are set correctly
+ client = createAdminClient
+ 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)
+ val quotas = configEntries.map {case (k, v) => new ClientQuotaAlteration.Op(k,v)}
+ val alterResult = client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, quotas.asJavaCollection)).asJavaCollection)
+ alterResult.all.get
+
+ val describeResult = client.describeClientQuotas(ClientQuotaFilter.all())
+ val quotaEntries = describeResult.entities().get()
+
+ assertTrue(quotaEntries.size() == 1)
Review Comment:
`assertEquals(1, quotaEntries.size())`
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -94,6 +95,26 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
+ // Since it's hard to stably reach quota limit in integration test, we only verify quota configs are set correctly
+ client = createAdminClient
+ 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)
+ val quotas = configEntries.map {case (k, v) => new ClientQuotaAlteration.Op(k,v)}
+ val alterResult = client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, quotas.asJavaCollection)).asJavaCollection)
+ alterResult.all.get
+
+ val describeResult = client.describeClientQuotas(ClientQuotaFilter.all())
Review Comment:
kafka metadata may be not synced so quick, and hence we need to use `TestUtils.waitUntilTrue` to loop it to avoid flaky test
--
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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on PR #16687:
URL: https://github.com/apache/kafka/pull/16687#issuecomment-2282650513
@chia7712 Done, please take a look.
--
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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16687:
URL: https://github.com/apache/kafka/pull/16687#discussion_r1703335810
##########
core/src/test/scala/integration/kafka/api/BaseAdminIntegrationTest.scala:
##########
@@ -230,6 +232,7 @@ abstract class BaseAdminIntegrationTest extends IntegrationTestHarness with Logg
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers())
config.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "20000")
config.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "40000")
+ config.put(AdminClientConfig.CLIENT_ID_CONFIG, clientId)
Review Comment:
This will impact all admins from this test. Maybe you can add it to `testCreatePartitionWithOptionRetryOnQuotaViolation` only
--
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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #16687:
URL: https://github.com/apache/kafka/pull/16687#discussion_r1697688555
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -94,6 +94,59 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
+ client = createAdminClient
+
+ val brokerIds = brokers.map(_.config.brokerId).toSet
+ brokerIds.foreach(id => TestUtils.waitForOnlineBroker(client, id))
+
+ val topicNums = 30
+
+ def makeTopics(): java.util.List[String] = {
+ val topics = for (i <- 0 until topicNums) yield {
+ val topicName = s"topic-$i"
+ createTopic(topicName)
+ topicName
+ }
+ topics.toList.asJava
+ }
+
+ val topics = makeTopics()
+
+ waitForTopics(client, topics.asScala.toSeq, List())
+ ensureConsistentKRaftMetadata()
+
+ val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.CLIENT_ID -> "test-client-id").asJava)
+
+ val quotas = Seq(new ClientQuotaAlteration.Op(QuotaConfigs.CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG, 1))
+ val alterResult = client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, quotas.asJava)).asJava)
+ alterResult.all.get
+
+ def makeCreatePartitionCalls(topics: java.util.List[String], numPartitions: Int, retry: Boolean): java.util.Collection[KafkaFuture[Void]] = {
+ val futures = topics.asScala.map{
+ topic => client.createPartitions(Map(topic -> NewPartitions.increaseTo(numPartitions)).asJava, new CreatePartitionsOptions().timeoutMs(30000).retryOnQuotaViolation(retry)).all()
+ }
+ futures.toList.asJavaCollection
+ }
+
+ def countFailedFutures(futures: java.util.Collection[KafkaFuture[Void]]): Int = {
+ futures.asScala.count {future =>
+ Try(future.get()) match {
+ case Failure(e: ExecutionException) if e.getCause.isInstanceOf[ThrottlingQuotaExceededException] => true
+ case _ => false
+ }
+ }
+ }
+
+ val failedFuture = makeCreatePartitionCalls(topics, 3, retry = false)
Review Comment:
Sounds great! I'll change my code to test the quota settings.
--
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(PlaintextAdminIntegrationTest): add test for createPartition wit… [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16687:
URL: https://github.com/apache/kafka/pull/16687#discussion_r1712681399
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -95,6 +96,35 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
+ // Since it's hard to stably reach quota limit in integration test, we only verify quota configs are set correctly
+ val config = createConfig
+ val clientId = "test-client-id"
+
+ config.put(AdminClientConfig.CLIENT_ID_CONFIG, clientId)
+ client = AdminClientTestUtils.create(config, null)
+
+ 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)
+ val quotas = configEntries.map {case (k, v) => new ClientQuotaAlteration.Op(k,v)}
+ val alterResult = client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, quotas.asJavaCollection)).asJavaCollection)
+ alterResult.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")
+
+ val describeResult = client.describeClientQuotas(ClientQuotaFilter.all())
+ val quotaEntities = describeResult.entities().get()
Review Comment:
`val quotaEntities = client.describeClientQuotas(ClientQuotaFilter.all()).entities().get()`
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -95,6 +96,35 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
+ // Since it's hard to stably reach quota limit in integration test, we only verify quota configs are set correctly
+ val config = createConfig
+ val clientId = "test-client-id"
+
+ config.put(AdminClientConfig.CLIENT_ID_CONFIG, clientId)
+ client = AdminClientTestUtils.create(config, null)
+
+ 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)
+ val quotas = configEntries.map {case (k, v) => new ClientQuotaAlteration.Op(k,v)}
+ val alterResult = client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, quotas.asJavaCollection)).asJavaCollection)
+ alterResult.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")
+
+ val describeResult = client.describeClientQuotas(ClientQuotaFilter.all())
+ val quotaEntities = describeResult.entities().get()
+ assertTrue(quotaEntities.containsKey(entity))
+ assertEquals(configEntries,quotaEntities.get(entity).asScala)
Review Comment:
`assertEquals(configEntries, quotaEntities.get(entity).asScala)`
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -95,6 +96,35 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
+ // Since it's hard to stably reach quota limit in integration test, we only verify quota configs are set correctly
+ val config = createConfig
+ val clientId = "test-client-id"
+
+ config.put(AdminClientConfig.CLIENT_ID_CONFIG, clientId)
+ client = AdminClientTestUtils.create(config, null)
+
+ 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)
+ val quotas = configEntries.map {case (k, v) => new ClientQuotaAlteration.Op(k,v)}
Review Comment:
```java
client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, configEntries.map {case (k, v) =>
new ClientQuotaAlteration.Op(k,v)}.asJavaCollection)).asJavaCollection).all.get
```
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -95,6 +96,35 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
+ // Since it's hard to stably reach quota limit in integration test, we only verify quota configs are set correctly
+ val config = createConfig
+ val clientId = "test-client-id"
+
+ config.put(AdminClientConfig.CLIENT_ID_CONFIG, clientId)
+ client = AdminClientTestUtils.create(config, null)
Review Comment:
`client = Admin.create(config)`
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -95,6 +96,35 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
+ // Since it's hard to stably reach quota limit in integration test, we only verify quota configs are set correctly
+ val config = createConfig
+ val clientId = "test-client-id"
+
+ config.put(AdminClientConfig.CLIENT_ID_CONFIG, clientId)
+ client = AdminClientTestUtils.create(config, null)
+
+ 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)
+ val quotas = configEntries.map {case (k, v) => new ClientQuotaAlteration.Op(k,v)}
+ val alterResult = client.alterClientQuotas(Seq(new ClientQuotaAlteration(entity, quotas.asJavaCollection)).asJavaCollection)
+ alterResult.all.get
+
+ TestUtils.waitUntilTrue(() => {
Review Comment:
```java
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")
```
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -95,6 +96,35 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
super.tearDown()
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testCreatePartitionWithOptionRetryOnQuotaViolation(): Unit = {
Review Comment:
please add `quorum: String`
--
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