KAFKA-15265: Add Remote Log Manager quota manager #15625

abhijeetk88 · 2024-03-29T12:11:35Z

In this PR, I have added the implementation of the quota manager that will be used to throttle copy and fetch requests from the remote storage. Reference KIP-956

Added unit-tests for the Quota Manager implementation

In follow-up PRs, the integration of the copy/fetch quota managers with copy and fetch paths will be added. Also the quota configs will be made dynamically configurable.

Committer Checklist (excluded from commit message)

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

core/src/main/java/kafka/log/remote/quota/RLMQuotaManager.java

funky-eyes · 2024-04-19T09:09:43Z

core/src/main/java/kafka/log/remote/quota/RLMQuotaManager.java

I have a question, in fact, what this PR does is to provide a standard configuration and generate corresponding rate limiters and related monitoring indicators, right? Then it needs to be used in the corresponding RemoteStorageManager, correct?

Yes, the integration of the quota manager will come in the follow-up PRs. I have mentioned it in the description of the PR.

kamalcph

Overall LGTM, left minor comments to address. Thanks for the patch!

core/src/main/java/kafka/log/remote/quota/RLMQuotaManagerConfig.java

core/src/main/java/kafka/log/remote/quota/RLMQuotaManager.java

core/src/test/java/kafka/log/remote/quota/RLMQuotaManagerTest.java

kamalcph

LGTM

showuon

Overall LGTM. Left some comments.

showuon · 2024-05-03T09:06:02Z

storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java

@@ -143,6 +143,38 @@ public final class RemoteLogManagerConfig {
            "less than or equal to `log.retention.bytes` value.";
    public static final Long DEFAULT_LOG_LOCAL_RETENTION_BYTES = -2L;

+    public static final String REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP = "remote.log.manager.copy.max.bytes.per.second";
+    public static final String REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_DOC = "The maximum number of bytes that can be copied from local storage to remote storage per second. " +
+            "This is a global limit for all the partitions that are being copied from remote storage to local storage. " +


This is a global limit for all the partitions that are being copied from remote storage to local storage. <-- is it right? Copied from local storage to remote storage?

My bad. Thanks for catching this.

showuon · 2024-05-03T09:23:17Z

core/src/main/java/kafka/log/remote/quota/RLMQuotaManager.java

+            MetricName quotaMetricName = metricName();
+            KafkaMetric metric = allMetrics.get(quotaMetricName);
+            if (metric != null) {
+                LOGGER.warn("Sensor for quota-id {} already exists. Setting quota to {} in MetricConfig", quotaMetricName, newQuota);


I'd like to know why we set WARN logs here. It looks to me if we want to update quota dynamically, it is expected the metric is already existed, right? If so, I don't think this is unexpected. So maybe INFO or DEBUG level, WDYT?

On second thought, INFO should be the right level, because quota update is a significant change in the application state and will decide how fast copies/fetches from remote storage can happen. Also, quota updates are infrequent, hence it will not cause excessive logging.

core/src/main/java/kafka/log/remote/quota/RLMQuotaManager.java

junrao

@abhijeetk88 : Thanks for the PR. Made a pass of non-testing files. Left a few comments.

junrao · 2024-05-03T19:51:44Z

core/src/main/java/kafka/log/remote/quota/RLMQuotaManagerConfig.java

+    private final int numQuotaSamples;
+    private final int quotaWindowSizeSeconds;
+
+    public long getQuotaBytesPerSecond() {


For consistency, we don't typically use getters. So this can just be quotaBytesPerSecond. Ditto below.

Sure, will change these.

junrao · 2024-05-03T20:13:28Z

core/src/main/java/kafka/log/remote/quota/RLMQuotaManager.java

+    }
+
+    public void record(double value) {
+        sensor().record(value, time.milliseconds(), false);


Why do we turn the quota checking off during record()? In other implementations like ClientQuotaManager.recordAndGetThrottleTimeMs(), we call record() by turning on quota checking and get back the amount of time to throttle.

In KIP-956, we do not utilize the throttle time provided by the quota manager to regulate fetches and copies. For fetch operations, we initially verify quota availability before initiating the retrieval of remote data. If the quota is unavailable, our priority is to serve partitions requiring local data, rather than throttling the client. Therefore, we focus on fulfilling data requests for other partitions in the queue, eliminating the need for throttle time in fetch operations.

Similarly, when a RLM Task attempts to copy a segment, it first checks if the write quota is available. If the quota is not available, the thread waits until the quota becomes available. As a result, we do not require throttle time for copies either.

core/src/main/java/kafka/log/remote/RemoteLogManager.java

junrao · 2024-05-03T20:21:19Z

storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java

+    public static final String REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM_PROP = "remote.log.manager.fetch.quota.window.num";
+    public static final String REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM_DOC = "The number of samples to retain in memory for remote fetch quota management. " +
+            "The default value is 11, which means there are 10 whole windows + 1 current window.";
+    public static final int DEFAULT_REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM = 11;


Why do we choose the default fetch window different from copy?

For fetches, the default window size was chosen to match the default window size used for other quotas, such as ClientQuota and ReplicationQuota.

Using an 11-second (10 whole + 1 current) window size for copies, similar to other quotas, does seem to be a better option. Consider this:

The broker-level quota for copying may be set to 250 MBps. The RLM task records the log segment size with the quota manager when uploading a log segment. The typical log segment size is 500 MB, meaning only one log segment can be uploaded every 2 seconds without breaching the quota. If uploads occur faster, the quota will be exceeded. Therefore, as long as the window size is greater than 2 seconds, either a 10-second or 60-second (whole) window should work.

However, a shorter window (10 seconds) has advantages. It tracks data uploads more precisely and prevents large spikes in data upload more effectively. For example:

With a 10-second window:

Buckets: b1, b2, ..., b10
In the 10th second, 5 segments can be uploaded without breaching the average quota (5 * 500 MB / 10 seconds = 250 MBps), though the spike will be 2.5 GB in that second.
With a 60-second window:

Buckets: b1, b2, ..., b60
In the 60th second, 30 segments can be uploaded without breaching the average quota (30 * 500 MB / 60 seconds = 250 MBps), but the spike will be 15 GB in that second.
Given the need to avoid quota breaches, a 10-second window is preferable to a 60-second window.

Let me know if it makes sense. I can change the default copy window to be the same as the default fetch window.

If there is no good reach, perhaps it's better to use the same default window number for copy too.

showuon · 2024-05-23T09:58:37Z

@abhijeetk88 , do we have any update on this PR?

abhijeetk88 · 2024-05-27T04:03:29Z

@showuon @junrao I have addressed your comments. Please take a look.

showuon

LGTM! Thanks.

satishd

Thanks @abhijeetk88 for addressing the review comments. LGTM.

kamalcph

Do we plan to change the default copy quota window num samples from 61 to 11?

core/src/main/java/kafka/log/remote/quota/RLMQuotaManagerConfig.java

showuon · 2024-05-29T02:31:50Z

@abhijeetk88 , there is a merge conflict. Please help resolve it. Thanks.

abhijeetk88 · 2024-05-29T05:11:10Z

@abhijeetk88 , there is a merge conflict. Please help resolve it. Thanks.

done

Do we plan to change the default copy quota window num samples from 61 to 11?

Waiting for a confirmation from Jun.

showuon · 2024-05-29T12:02:37Z

@junrao , since this PR blocks other follow-up PRs and v3.8.0 release date is approaching, I'd like to merge it tomorrow if you don't have any other comments. Thanks.

junrao

@abhijeetk88 : Sorry for the delay. Just a minor followup comment. Otherwise, the PR LGTM.

junrao · 2024-05-29T20:36:34Z

storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java

+    public static final String REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM_PROP = "remote.log.manager.fetch.quota.window.num";
+    public static final String REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM_DOC = "The number of samples to retain in memory for remote fetch quota management. " +
+            "The default value is 11, which means there are 10 whole windows + 1 current window.";
+    public static final int DEFAULT_REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM = 11;


If there is no good reach, perhaps it's better to use the same default window number for copy too.

junrao

@abhijeetk88 : Thanks for the updated PR. LGTM

jolshan · 2024-05-30T20:21:57Z

Can we look at testCopyQuotaManagerConfig() – kafka.log.remote.RemoteLogManagerTest? It seems like it is failing pretty consistently. https://issues.apache.org/jira/browse/KAFKA-16866

chia7712 · 2024-05-30T21:07:21Z

@jolshan I file #16146 to fix it.

junrao · 2024-05-30T21:53:11Z

@jolshan : Thanks for pointing this out. Sorry that I didn't look at the test results carefully before merging.

abhijeetk88 · 2024-05-31T04:01:16Z

Thanks @chia7712 @jolshan . Apologies for the miss.

Added the implementation of the quota manager that will be used to throttle copy and fetch requests from the remote storage. Reference KIP-956 Reviewers: Luke Chen <[email protected]>, Kamal Chandraprakash <[email protected]>, Jun Rao <[email protected]>

abhijeetk88 marked this pull request as draft March 29, 2024 12:11

abhijeetk88 force-pushed the KAFKA-15434 branch 3 times, most recently from 577b442 to 9ecd20d Compare March 30, 2024 17:45

abhijeetk88 force-pushed the KAFKA-15434 branch from 9ecd20d to 0d29c4c Compare April 13, 2024 19:17

abhijeetk88 changed the title ~~[WIP] Add Remote Log Manager quota manager~~ Add Remote Log Manager quota manager Apr 13, 2024

abhijeetk88 marked this pull request as ready for review April 13, 2024 19:31

abhijeetk88 changed the title ~~Add Remote Log Manager quota manager~~ KAFKA-15265: Add Remote Log Manager quota manager Apr 15, 2024

HenryCaiHaiying reviewed Apr 16, 2024

View reviewed changes

core/src/main/java/kafka/log/remote/quota/RLMQuotaManager.java Outdated Show resolved Hide resolved

satishd requested review from junrao, showuon, kamalcph and satishd April 18, 2024 10:29

funky-eyes reviewed Apr 19, 2024

View reviewed changes

funky-eyes mentioned this pull request Apr 22, 2024

How to limit the speed of tiered storage Aiven-Open/tiered-storage-for-apache-kafka#520

Closed

kamalcph reviewed Apr 24, 2024

View reviewed changes

kamalcph approved these changes Apr 26, 2024

View reviewed changes

showuon reviewed May 3, 2024

View reviewed changes

junrao reviewed May 3, 2024

View reviewed changes

abhijeetk88 mentioned this pull request May 24, 2024

KAFKA-15265: Integrate RLMQuotaManager for throttling copies to remote storage #15820

Merged

3 tasks

abhijeetk88 force-pushed the KAFKA-15434 branch from 56549ea to 48f6b51 Compare May 24, 2024 13:25

This was referenced May 25, 2024

KAFKA-15265: Integrate RLMQuotaManager for throttling fetches from remote storage #16071

Merged

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

Merged

showuon approved these changes May 27, 2024

View reviewed changes

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

satishd approved these changes May 28, 2024

View reviewed changes

kamalcph reviewed May 28, 2024

View reviewed changes

core/src/main/java/kafka/log/remote/quota/RLMQuotaManagerConfig.java Show resolved Hide resolved

abhijeetk88 force-pushed the KAFKA-15434 branch from 48f6b51 to e3c4adf Compare May 29, 2024 05:11

junrao reviewed May 29, 2024

View reviewed changes

abhijeetk88 added 6 commits May 30, 2024 11:10

Add Remote Log Manager quota manager

0f20308

Address review comments

7ec240e

Address review comments

21f5363

Rebase

a2048f2

Address review comments

a4f3d63

Change default window size to 11

d57b8c3

abhijeetk88 force-pushed the KAFKA-15434 branch from e3c4adf to d57b8c3 Compare May 30, 2024 05:52

junrao approved these changes May 30, 2024

View reviewed changes

junrao merged commit bb7db87 into apache:trunk May 30, 2024
1 check failed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-15265: Add Remote Log Manager quota manager #15625

KAFKA-15265: Add Remote Log Manager quota manager #15625

abhijeetk88 commented Mar 29, 2024 •

edited

Loading

funky-eyes Apr 19, 2024

abhijeetk88 Apr 23, 2024

kamalcph left a comment

kamalcph left a comment

showuon left a comment

showuon May 3, 2024

abhijeetk88 May 12, 2024

showuon May 3, 2024

abhijeetk88 May 12, 2024

junrao left a comment

junrao May 3, 2024

abhijeetk88 May 12, 2024

junrao May 3, 2024

abhijeetk88 May 24, 2024

junrao May 3, 2024

abhijeetk88 May 24, 2024

junrao May 29, 2024

abhijeetk88 May 30, 2024

showuon commented May 23, 2024

abhijeetk88 commented May 27, 2024

showuon left a comment

satishd left a comment

kamalcph left a comment

showuon commented May 29, 2024

abhijeetk88 commented May 29, 2024

showuon commented May 29, 2024

junrao left a comment

junrao May 29, 2024

junrao left a comment

jolshan commented May 30, 2024 •

edited

Loading

chia7712 commented May 30, 2024

junrao commented May 30, 2024

abhijeetk88 commented May 31, 2024

KAFKA-15265: Add Remote Log Manager quota manager #15625

KAFKA-15265: Add Remote Log Manager quota manager #15625

Conversation

abhijeetk88 commented Mar 29, 2024 • edited Loading

Committer Checklist (excluded from commit message)

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

kamalcph left a comment

Choose a reason for hiding this comment

showuon 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

junrao 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

showuon commented May 23, 2024

abhijeetk88 commented May 27, 2024

showuon left a comment

Choose a reason for hiding this comment

satishd left a comment

Choose a reason for hiding this comment

kamalcph left a comment

Choose a reason for hiding this comment

showuon commented May 29, 2024

abhijeetk88 commented May 29, 2024

showuon commented May 29, 2024

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

junrao left a comment

Choose a reason for hiding this comment

jolshan commented May 30, 2024 • edited Loading

chia7712 commented May 30, 2024

junrao commented May 30, 2024

abhijeetk88 commented May 31, 2024

abhijeetk88 commented Mar 29, 2024 •

edited

Loading

jolshan commented May 30, 2024 •

edited

Loading