KAFKA-15265: Dynamic broker configs for remote fetch/copy quotas #16078

abhijeetk88 · 2024-05-24T19:46:52Z

In this PR, I have added dynamic config support for remote fetch/copy quotas. Reference KIP-956

Added unit-tests for the new dynamic broker configs

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

kamalcph

LGTM (reviewed the 4th commit in the PR)

showuon · 2024-05-31T10:10:46Z

@abhijeetk88 , Please help resolve the merge conflict. Thanks.

abhijeetk88 · 2024-05-31T12:30:49Z

@showuon I have updated the PR after resolving conflicts. Please take a look.

satishd

Thanks @abhijeetk88 for the PR, left a few comments.

satishd · 2024-06-07T04:48:23Z

core/src/main/scala/kafka/server/DynamicBrokerConfig.scala

@@ -1166,7 +1166,9 @@ class DynamicRemoteLogConfig(server: KafkaBroker) extends BrokerReconfigurable w
  override def validateReconfiguration(newConfig: KafkaConfig): Unit = {
    newConfig.values.forEach { (k, v) =>
      if (reconfigurableConfigs.contains(k)) {
-        if (k.equals(RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP)) {
+        if (k.equals(RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP) ||


I see that it is the existing logic for the existing config key RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROPfor the respective Long values.

Although all the valid keys are already checked by the earlier line if (reconfigurableConfigs.contains(k)), they are getting repeated here to check and cast the value as Long. Can we avoid double checking by removing the earlier check reconfigurableConfigs.contains(k)?

I wasn't sure, why there was double-checking initially, so did not want to change it and just added the new configs check. But I think it should be ok to remove double-checks. Will fix this.

satishd · 2024-06-07T05:09:09Z

core/src/main/scala/kafka/server/DynamicBrokerConfig.scala

        remoteLogManager.get.resizeCacheSize(newValue)
        info(s"Dynamic remote log manager config: ${RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP} updated, " +
          s"old value: $oldValue, new value: $newValue")
+      } else if (isChangedLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP)) {


This method should take care of updating any of the respective configs. That means, it should individually check for each config by replacing else if with if.

It is also good to add a test that updates multiple configs and check all of them are getting updated.

Thanks for catching this. Fixed this and also added a test.

satishd · 2024-06-07T05:13:34Z

core/src/main/scala/kafka/server/DynamicBrokerConfig.scala

@@ -1195,13 +1214,19 @@ class DynamicRemoteLogConfig(server: KafkaBroker) extends BrokerReconfigurable w
    name match {
      case RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP =>
        config.getLong(RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP)
+      case RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP =>


This match case can be simplified like the below code:

name match { case RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP | RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP | RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP => config.getLong(name) case n => throw new IllegalStateException(s"Unexpected dynamic remote log manager config $n")

Makes sense.

abhijeetk88 · 2024-06-08T23:43:34Z

@satishd I have addressed your comments. Please take a look.

kamalcph · 2024-06-09T09:37:26Z

core/src/main/scala/kafka/server/DynamicBrokerConfig.scala

-            val errorMsg = s"Dynamic remote log manager config update validation failed for $k=$v"
-            throw new ConfigException(s"$errorMsg, value should be at least 1")
-          }
+      if (k.equals(RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP) ||


can we do an inverse check since we removed the contains check? What if the k is null?

if (RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP.equals(k)) || ...

Not completely sure if k can be null, but I changed it since it provides safety.

kamalcph

Please rebase the branch with trunk

kamalcph · 2024-06-09T14:47:54Z

core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala

+    // Update default config
+    props.put(copyQuotaProp, "100")
+    config.dynamicConfig.updateDefaultConfig(props)
+    assertEquals(100, config.getLong(copyQuotaProp))


should we move the getter method from RemoteLogManagerConfig to KafkaConfig class?

remoteLogManagerCopyMaxBytesPerSecond

remoteLogManagerFetchMaxBytesPerSecond and

remoteLogIndexFileCacheTotalSizeBytes

Should we take this as part of a follow-up PR? I want to avoid unrelated changes here.

kamalcph · 2024-06-09T14:49:46Z

core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala

+  }
+
+  @Test
+  def testRemoteLogManagerFetchQuotaUpdates(): Unit = {


this test is similar to testRemoteLogManagerCopyQuotaUpdates. can we refactor the test to extract the method? Or, use parameterized test.

kamalcph

LGTM, thanks for the patch!

satishd

Thanks @abhijeetk88 for addressing the review comments. LGTM.

satishd · 2024-06-12T14:16:55Z

There were a few unrelated test failures, merging it to trunk and 3.8 branches.

) Reviewers: Kamal Chandraprakash<[email protected]>, Satish Duggana <[email protected]>

commit 9368ef8 Author: Gantigmaa Selenge <[email protected]> Date: Wed Jun 12 16:04:24 2024 +0100 KAFKA-16865: Add IncludeTopicAuthorizedOperations option for DescribeTopicPartitionsRequest (apache#16136) Reviewers: Mickael Maison <[email protected]>, Chia-Ping Tsai <[email protected]>, Calvin Liu <[email protected]>, Andrew Schofield <[email protected]>, Apoorv Mittal <[email protected]> commit 46eb081 Author: gongxuanzhang <[email protected]> Date: Wed Jun 12 22:23:39 2024 +0800 KAFKA-10787 Apply spotless to log4j-appender, trogdor, jmh-benchmarks, examples, shell and generator (apache#16296) Reviewers: Chia-Ping Tsai <[email protected]> commit 79b9c44 Author: gongxuanzhang <[email protected]> Date: Wed Jun 12 22:19:47 2024 +0800 KAFKA-10787 Apply spotless to connect module (apache#16299) Reviewers: Chia-Ping Tsai <[email protected]> commit b5fb654 Author: Abhijeet Kumar <[email protected]> Date: Wed Jun 12 19:47:46 2024 +0530 KAFKA-15265: Dynamic broker configs for remote fetch/copy quotas (apache#16078) Reviewers: Kamal Chandraprakash<[email protected]>, Satish Duggana <[email protected]> commit faee6a4 Author: Dmitry Werner <[email protected]> Date: Wed Jun 12 15:44:11 2024 +0500 MINOR: Use predetermined dir IDs in ReplicationQuotasTest Use predetermined directory IDs instead of Uuid.randomUuid() in ReplicationQuotasTest. Reviewers: Igor Soarez <[email protected]> commit 638844f Author: David Jacot <[email protected]> Date: Wed Jun 12 08:29:50 2024 +0200 KAFKA-16770; [2/2] Coalesce records into bigger batches (apache#16215) This patch is the continuation of apache#15964. It introduces the records coalescing to the CoordinatorRuntime. It also introduces a new configuration `group.coordinator.append.linger.ms` which allows administrators to chose the linger time or disable it with zero. The new configuration defaults to 10ms. Reviewers: Jeff Kim <[email protected]>, Justine Olshan <[email protected]> commit 39ffdea Author: Bruno Cadonna <[email protected]> Date: Wed Jun 12 07:51:38 2024 +0200 KAFKA-10199: Enable state updater by default (apache#16107) We have already enabled the state updater by default once. However, we ran into issues that forced us to disable it again. We think that we fixed those issues. So we want to enable the state updater again by default. Reviewers: Lucas Brutschy <[email protected]>, Matthias J. Sax <[email protected]> commit 0782232 Author: Antoine Pourchet <[email protected]> Date: Tue Jun 11 22:31:43 2024 -0600 KAFKA-15045: (KIP-924 pt. 22) Add RackAwareOptimizationParams and other minor TaskAssignmentUtils changes (apache#16294) We now provide a way to more easily customize the rack aware optimizations that we provide by way of a configuration class called RackAwareOptimizationParams. We also simplified the APIs for the optimizeXYZ utility functions since they were mutating the inputs anyway. Reviewers: Anna Sophie Blee-Goldman <[email protected]> commit 226ac5e Author: Murali Basani <[email protected]> Date: Wed Jun 12 05:38:50 2024 +0200 KAFKA-16922 Adding unit tests for NewTopic (apache#16255) Reviewers: Chia-Ping Tsai <[email protected]> commit 23fe71d Author: Abhijeet Kumar <[email protected]> Date: Wed Jun 12 06:27:02 2024 +0530 KAFKA-15265: Integrate RLMQuotaManager for throttling copies to remote storage (apache#15820) - Added the integration of the quota manager to throttle copy requests to the remote storage. Reference KIP-956 - Added unit-tests for the copy throttling logic. Reviewers: Satish Duggana <[email protected]>, Luke Chen <[email protected]>, Kamal Chandraprakash<[email protected]> commit 2fa2c72 Author: Chris Egerton <[email protected]> Date: Tue Jun 11 23:15:07 2024 +0200 MINOR: Wait for embedded clusters to start before using them in Connect OffsetsApiIntegrationTest (apache#16286) Reviewers: Greg Harris <[email protected]>

abhijeetk88 marked this pull request as draft May 24, 2024 19:47

abhijeetk88 changed the title ~~[WIP] KAFKA-15265: Dynamic broker configs for remote fetch/copy quotas~~ KAFKA-15265: Dynamic broker configs for remote fetch/copy quotas May 25, 2024

abhijeetk88 marked this pull request as ready for review May 25, 2024 08:17

satishd requested review from satishd, hudeqi, junrao, showuon and kamalcph and removed request for hudeqi May 27, 2024 04:46

satishd added the tiered-storage Related to the Tiered Storage feature label May 27, 2024

kamalcph approved these changes May 28, 2024

View reviewed changes

abhijeetk88 force-pushed the KAFKA-15434_dynamic_config branch from 0e34986 to f226aae Compare May 31, 2024 12:19

satishd reviewed Jun 7, 2024

View reviewed changes

kamalcph reviewed Jun 9, 2024

View reviewed changes

abhijeetk88 force-pushed the KAFKA-15434_dynamic_config branch from 5173c4f to f72c000 Compare June 12, 2024 05:26

abhijeetk88 added 2 commits June 12, 2024 11:07

KAFKA-15265: Dynamic broker configs for remote fetch/copy quotas

82d0413

Address review comments

189e859

abhijeetk88 force-pushed the KAFKA-15434_dynamic_config branch from f72c000 to 189e859 Compare June 12, 2024 05:42

kamalcph approved these changes Jun 12, 2024

View reviewed changes

satishd approved these changes Jun 12, 2024

View reviewed changes

satishd merged commit b5fb654 into apache:trunk Jun 12, 2024
1 check failed

satishd pushed a commit that referenced this pull request Jun 12, 2024

KAFKA-15265: Dynamic broker configs for remote fetch/copy quotas (#16078

5fd9bd1

) Reviewers: Kamal Chandraprakash<[email protected]>, Satish Duggana <[email protected]>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-15265: Dynamic broker configs for remote fetch/copy quotas #16078

KAFKA-15265: Dynamic broker configs for remote fetch/copy quotas #16078

abhijeetk88 commented May 24, 2024 •

edited

Loading

kamalcph left a comment •

edited

Loading

showuon commented May 31, 2024

abhijeetk88 commented May 31, 2024

satishd left a comment

satishd Jun 7, 2024

abhijeetk88 Jun 8, 2024

satishd Jun 7, 2024

abhijeetk88 Jun 8, 2024

satishd Jun 7, 2024

abhijeetk88 Jun 8, 2024

abhijeetk88 commented Jun 8, 2024

kamalcph Jun 9, 2024 •

edited

Loading

abhijeetk88 Jun 12, 2024

kamalcph left a comment

kamalcph Jun 9, 2024

abhijeetk88 Jun 12, 2024

kamalcph Jun 9, 2024

abhijeetk88 Jun 12, 2024

kamalcph left a comment

satishd left a comment

satishd commented Jun 12, 2024

KAFKA-15265: Dynamic broker configs for remote fetch/copy quotas #16078

KAFKA-15265: Dynamic broker configs for remote fetch/copy quotas #16078

Conversation

abhijeetk88 commented May 24, 2024 • edited Loading

Committer Checklist (excluded from commit message)

kamalcph left a comment • edited Loading

Choose a reason for hiding this comment

showuon commented May 31, 2024

abhijeetk88 commented May 31, 2024

satishd left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

abhijeetk88 commented Jun 8, 2024

kamalcph Jun 9, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kamalcph left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kamalcph left a comment

Choose a reason for hiding this comment

satishd left a comment

Choose a reason for hiding this comment

satishd commented Jun 12, 2024

abhijeetk88 commented May 24, 2024 •

edited

Loading

kamalcph left a comment •

edited

Loading

kamalcph Jun 9, 2024 •

edited

Loading