...

    public static final String REMOTE_COPY_LAG_MS_CONFIG = "remote.copy.lag.ms";
    public static final String REMOTE_COPY_LAG_MS_DOC = "Controls how long to delay uploading segments to remote storage. " +
            "When set to 0 (default), segments are uploaded as soon as they are eligible (no delay). " +
            "When set to -1, resolves to <code>local.retention.ms</code> (maximum delay). " +
            "When set to a positive value (ms), a segment is eligible for upload only after it has been closed for at least this many milliseconds. " +
            "The value should not exceed <code>local.retention.ms</code> (local retention time).";

    public static final String REMOTE_COPY_LAG_BYTES_CONFIG = "remote.copy.lag.bytes";
    public static final String REMOTE_COPY_LAG_BYTES_DOC = "Controls size-based delay for uploading segments to remote storage. " +
            "When set to 0 (default), segments are uploaded as soon as they are eligible (no size-based constraint). " +
            "When set to -1, resolves to <code>local.retention.bytes</code> (maximum delay). " +
            "When set to a positive value (bytes), a segment is eligible for upload only when at least this many bytes of log data exist after the segment. " +
            "The value should not exceed <code>local.retention.bytes</code> (local retention size).";

...

    public static final String LOG_REMOTE_COPY_LAZYLAG_ENABLEMS_CONFIGPROP = LOG_PREFIX + TopicConfig.REMOTE_COPY_LAZY_ENABLE_CONFIG"log.remote.copy.lag.ms";
    public static final String LOG_REMOTE_COPY_LAZYLAG_ENABLEBYTES_DOCPROP = TopicConfig.REMOTE_COPY_LAZY_ENABLE_DOC"log.remote.copy.lag.bytes";

The default value is false are 0 so that the whole remote storage module keeps the original behavior.

Proposed Changes

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

...

You can see the uploading will be delayed if the configure remote.copy.lazy.enable is trueitems set with non default value. 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

...

1. Local files won’t be deleted until they’ve been uploaded to the remote storage, so this change is very safe
   You don’t need to worry about files being cleaned up before they be upload to the remote.

2. When local and complete retention values are set to the same. What is the behaviour if the delay lag is set to local retention?
   This is corner case. So keep same as the existing behaviour; the segment will be uploaded to remote, then allowed for local-log deletion.
   

   Code Block
   language plain title Note
   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. when you also set the lag to max one (local rentention configure value) 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 configures' value (false) maintains current behavior
  Forward compatible: Older clients unaware of this config these configs will ignore it
  
  

• Deprecation
  N/A
• Migration for Existing Deployments
  N/A. The feature is optional

...

1. Test upload eligibility logic with delay enabled/disableddifferent configure values
2. Test configuration validation for topic

...