...

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

• Topic Creation:

  Code Block
  language java title 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.Backward compatible: Default value (true) maintains current behavior
  Forward compatible: Older clients unaware of this config will ignore it
  
  

• 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 

• N/A. The feature is optional with topic level.

Test Plan

We can use follow tests to cover the change:

Unit Tests: * Verify that NewTopic includes min.insync.replicas configuration

1. Test upload eligibility logic with delay enabled/disabled
2. Test configuration validation for topic

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

1. Verify segments are uploaded before local segment deleted
2. Test the remote storage reduced after topic enable the feature.

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.

...