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/20811 for the detailed changes.

• Configuration Loading:

  TopicBasedRemoteLogMetadataManagerConfig

  public static final String REMOTE_LOG_METADATA_TOPIC_MIN_ISR_PROP =  "remote.log.metadata.topic.min.isr";
  public static final String REMOTE_LOG_METADATA_TOPIC_MIN_ISR_DOC =  "The minimum number of replicas that must acknowledge a write to remote log metadata topic";
  public static final short DEFAULT_REMOTE_LOG_METADATA_TOPIC_MIN_ISR = 2;

  The new configuration will be added to TopicBasedRemoteLogMetadataManagerConfig, the configuration will be validated to ensure it does not exceed the replication factor.

• 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

This another approach is maintaining the current default of min.isr=1, It was rejected for the following reasons:

1. Security by Default Principle
   Kafka should provide secure defaults for critical internal topics. The current default of min.isr=1 creates an unacceptable data loss risk that most users will not proactively address. 

2. Inconsistency with other critical metadata topics
   The __transaction_state topic explicitly sets min.isr=2 via transaction.state.log.min.isr, There is no justifiable reason for treating remote log metadata differently.