Status

Current state:  Discussion

Discussion thread: here

Vote thread:  here

JIRA: KAFKA-19893

PR: https://github.com/apache/kafka/pull/20913

Motivation

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.

 When there is no requirement for real-time analytics or immediate consumption from remote storage. It has the following drawbacks:

1. Wastes storage capacity and costs: The same data is stored twice during the local retention window
2. Provides no immediate benefit: During the local retention period, reads prioritize local data, making the remote copy unnecessary

Example Scenari and The goal:

Consider a topic with remote stroage enable:
- Local retention: 1 day (24 hours)
- Remote retention: 3 days (72 hours)

Currently, data is stored in both tiers for the first day, resulting in about 16 hours of redundant storage. This can leads to cost increases.

So we can reduce the tiered storage redundancy for cost saving. In this case, it will save about 25% cost pay for total size of disk.

However, some users/topics rely on remote storage for real-time analytics 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, but it still can’t include the latest data because the active segment hasn’t been uploaded yet.). Therefore, this optimization is offered as a topic's optional configuration rather than the default behavior.

Public Interfaces

This KIP introduces one new topic configuration property:  remote.log.keep.latest

public static final String REMOTE_LOG_KEEP_LATEST_CONFIG = "remote.log.keep.latest";
public static final String REMOTE_LOG_KEEP_LATEST_DOC = "Determines whether to upload all segments to remote storage including the latest ones within local retention. " +
            "When set to true (default), all committed segments will be uploaded without checking local retention constraints. " +
            "When set to false, only segments beyond local retention period will be uploaded to remote storage.";

The default value is true so that the whole remote stroage module keeps the orginial behavior when topic don't set it to false.

Proposed Changes

You can refer to https://github.com/apache/kafka/pull/20913 for the detailed changes.

• Topic Creation:

  TopicBasedRemoteLogMetadataManager

  topicConfigs.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE);
  topicConfigs.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "false");
  topicConfigs.put(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, Short.toString(rlmmConfig.metadataTopicMinIsr())); //add the configure
  return new NewTopic(rlmmConfig.remoteLogMetadataTopicName(), rlmmConfig.metadataTopicPartitionsCount(), rlmmConfig.metadataTopicReplicationFactor()).configs(topicConfigs);

  When TopicBasedRemoteLogMetadataManager creates the __remote_log_metadata topic, it will include the min.isr configuration:          

Compatibility, Deprecation, and Migration Plan 

• Backward Compatibility
  This change is fully backward compatible:
  1. Existing Deployments: Clusters with an existing __remote_log_metadata topic will continue to operate unchanged. The topic's current min.insync.replicas setting (typically inherited from the broker default) will remain in effect.
  2. New Deployments: Only clusters enabling Tiered Storage for the first time, after upgrading to the release containing this change will automatically receive min.isr=2 for the topic.
  3. Protocol Compatibility: No changes
  4. Rolling Upgrade: The cluster can be upgraded using standard rolling upgrade procedures without special considerations due to the topic only be created once.

• Deprecation
  N/A
• Migration for Existing Deployments
  Users with existing __remote_log_metadata topics can evaluate their current configuration and consider updating it. This operation is not mandatory.

           Step 1: Check Current Configuration

                       kafka-configs.sh --bootstrap-server localhost:9092 --describe --topic __remote_log_metadata

           Step 2: Update Configuration if need.  The change takes effect immediately and requires no restart.

                     If the current min.isr is 1 and replication.factor is 3 or higher:

                     kafka-configs.sh --bootstrap-server localhost:9092  --alter --topic __remote_log_metadata  --add-config min.insync.replicas=2 

Test Plan

We can use follow tests to cover the change:

Unit Tests:

 * Verify that NewTopic includes min.insync.replicas configuration

Integration Tests: 

 * Create __remote_log_metadata topic with default configuration
 * Verify topic can be created with custom replication.factor and min.isr

Rejected Alternatives

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.

Alternative 2: Global broker-level configuration only

Reason for rejection: Different topics have different requirements. Topic-levelgranularity is essential.