Current state: Discussion
Discussion thread: here or here (sometimes. the mail thread can't show all the mails. so I provided two links here)
Vote thread: here
JIRA: KAFKA-19893
PR: https://github.com/apache/kafka/pull/20913
This KIP proposes an optional, topic-level feature that provides an opportunity for cost savings on remote storage when the scenario is suitable to use.
Currently, Kafka's tiered storage implementation uploads all non-active local log segments to remote storage immediately, even when they are still within the local retention period.
This results in redundant storage of the same data in both local and remote tiers.
We can see the one real redundancy example in the following picture. (I just highlight only one segment.)
When there is no requirement for real-time analytics based on remote storage directly. It has the following drawbacks:
1. Wastes storage capacity and costs: The same data is stored twice during the local retention window
2. No immediate benefit: During the local retention period, reads prioritize local data, making the remote copy unnecessary
Example scenario and this KIP's goal:
Consider a topic with remote storage enable:
- Local retention: 1 day (24 hours)
- Remote retention: 3 days (72 hours)
Data is stored in both tiers for the first day, resulting in about 16 hours of redundant storage. This can leads to cost waste.
So we can reduce this storage redundancy. In this case, it will save about 25% cost payment for total size of disk.
We can take an AWS S3 billing example (last 3 months) to show the cost saving detail:
AWS S3 has two main cost items for Kafka usage (normally, Kafka deployed machines and the remote storage/S3 bucket are placed in the same region so that we can ignore the network transfer cost).
The cost item TimedStorage-ByteHrs represents the storage part which costs are based on the stored object’s lifetime.
In above case, each remote segment lives for about 64 hours. thus, with the new optional feature, each will live for about 48 hours, resulting in approximately 25% cost savings.
In this billing example, the cost would decrease about 67K per quarter. BTW: if our topic is local:1 day + remote: 7 days . the saving part will be about 10% （27K）
so it mean that if you set 1 day local + 3-7 days remote. the cost-saving for TimedStorage-ByteHrs will range from 25% to 10%..
Note: You can also refer to Amazon S3 Cost to know more information.
However, this optimization is offered as a topic level optional configuration rather than the default behavior based on followed scenarios:
(1) Some users/topics rely on remote storage for real-time analytics based on remote storage directly and need the latest data to be available as soon as possible (In fact, it only tries to stay as up-to-date as possible,
because it still can’t include the latest data because the active segment always hasn’t been uploaded yet.).
(2) Some topics may set a very high ratio for remote-to-local retention time. The cost savings amount will be small, so it is mainly to avoid waste. Users may think it is not worth enabling the feature for the topics
Considering the latency or risk of remote storage, the local retention period won’t be set too short for me. For example, in our production environment, we keep one day of local data alongside 3-7 days in remote storage, Keeping one day of data in local storage ensures that Kafka users have enough time to handle unexpected consumption issues while maintaining good latency, and it also reduces the risk of relying on remote storage. so there’s still one day of redundancy for my case. What’s more, even if you configure a very short local retention time, you may need to extend it when certain issues occur. The more local retention time you keep, the more cost savings you achieve. |
(3) Kafka admin want to reduce the expansion times for local disk when remote storage down for a long time. after all. The already uploaded segments are eligible for deletion from broker when not enable the feature for topic.
You can check the follow picture ( Configure: 1 day local + 3 day remote) to understand the logic:
(a) If the remote storage outage for a short time: no matter if you enable the feature. it don't have difference.
(b) If the remote storage outage for a medium-term time
You will need one extra expansion: the max size is the your saving cost's part. You can think you should return back the save part/redundancy by expansion.
(c) If the remote storage outage for a long time (check the 48 hours' outage example in picture)）:
No matter if you enable the feature. you should keep doing expansion due to remote upload failed.
This KIP introduces:
(1) one new topic configuration item: remote.copy.lazy.enable
BTW: topic's remote storage feature already had some others configure items such as remote.log.delete.on.disable, remote.log.copy.disable, etc.
public static final String REMOTE_COPY_LAZY_ENABLE_CONFIG = "remote.copy.lazy.enable";
public static final String REMOTE_COPY_LAZY_ENABLE_DOC = "Determines whether to delay uploading segments to remote storage. " +
"When set to false (default), all non-active segments will be uploaded immediately without checking local retention constraints. " +
"When set to true, segments within local retention will not be uploaded, the upload is delayed until they expire.";
|
(2) one new server configuration item: log.remote.copy.lazy.enable
If a user wants to enable the lazy copy behaviour for all the topics (including the new ones), then they can set this broker level config to true. Otherwise,
it will be hard for the user to create the new topics with this config set when remote storage is enabled.
public static final String LOG_REMOTE_COPY_LAZY_ENABLE_CONFIG = LOG_PREFIX + TopicConfig.REMOTE_COPY_LAZY_ENABLE_CONFIG;
public static final String LOG_REMOTE_COPY_LAZY_ENABLE_DOC = TopicConfig.REMOTE_COPY_LAZY_ENABLE_DOC; |
The default value is false so that the whole remote storage module keeps the original behavior.
You can refer to https://github.com/apache/kafka/pull/20913 for the detailed changes.
We change the RemoteLogManager.RLMCopyTask#candidateLogSegments's logic for decide one segment if need to upload to remote:
You can see the uploading will be delayed if the configure remote.copy.lazy.enable is true. And After the change, the remote tiered storage redundancy will be reduced with delayed upload.
You can refer to the test case and result: https://github.com/apache/kafka/pull/20913#issuecomment-3547156286
The storage compare in my test: Partition 2 (host on the machine with the delay upload feature) is about 50% of Partition 0 or Partition 1.
BTW: Here are some additional thoughts/considerations.
When local and complete retention values are set to the same. What is the behaviour?
This is corner case. So keep same as the existing behaviour; the segment will be uploaded to remote, then allowed for local-log deletion.
Note: Actually, this is a valid case because local retention only not allow to > remote retention in current code. If they are equal, it is better to skip the update, since the segment would be immediately deleted after being uploaded to remote storage. However, if we do not upload it to remote storage, the local segment will not be deleted because it waits for the highest offset in remote storage to be updated after the upload. Moreover, if we skip the upload but directly update the highest offset in remote storage, it becomes ambiguous whether the segment has already been uploaded or not. Therefore, We can skip the upload and update the LogStartOffset. The demo PR is: https://github.com/apache/kafka/pull/21361, Considering this is a corner case and this solution also helps address another issue: if the remote storage service is unavailable for a long time, local segments may never get deleted forever even it over the retention time. It means that it isn't special case for this KIP. So I just list this thought here. |
Backward Compatibility
This change is fully compatible:
Backward compatible: Default value (true) maintains current behavior
Forward compatible: Older clients unaware of this config will ignore it
We can use follow tests to cover the change:
Unit Tests:
Integration Tests:
Alternative 1: Make this the default behavior
Reason for rejection: Some users require real-time remote analytics and need data uploaded as soon as possible. Breaking their use case would be unacceptable. after all it is the default behavior before this change.