KAFKA-19893: Reduce tiered storage redundancy with delayed upload (KIP-1241)

[jiafu1115]

JIRA:19893
KIP:1241

Currently, Kafka uploads all non-active local log segments to remote
storage even when they are still within the local retention period,
resulting in redundant storage of the same data in both tiers.

This wastes storage capacity (cost) without providing immediate
benefits,since reads during the retention window prioritize local data.

However, some users/topics do 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, 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.

Here are some additional thoughts/considerations.

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. Considering the latency of remote storage, the local retention period
   won’t be set too short. For example, in our production environment, we
   keep 1 day of local data alongside 3-7 days in remote storage, so
   there’s still 1 day of redundancy.

Example for the goal:

Reviewers: Kamal Chandraprakash kamal.chandraprakash@gmail.com

[jiafu1115]

Attach test result:
[Precondition]
Create one topic enable remote stroage in Kafka (3 brokers + 3 controller)

local storage time: 20 minutes
remote stroage time: 40 minutes
partition:  3
segement.bytes: 10M

[Steps]

1. Deploy this code patch into one broker only and restart the broker
2. Keep sending a lot of messages to the topic
3. Check the disk sizes on both local and remote storage at two points in time: 20 minutes before and 1 hour after.

[Result]

Before 20 minutes:

1. only 2 partition upload the local to remote.

After 1 hour:

1. The remote storage size for one partition (on the broker with the code change) is much smaller than the other two.

2. The sizes of the local disks are similar.

[github-actions]

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

[jiafu1115]

Hi, @kamalcph
Sorry to bother you. I know you’ve been deeply involved in the remote storage area, and I was wondering if you might be interested — when you have some free time — in taking a look at this cost-saving topic and providing some guidance. Thank you very much!

[github-actions]

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

[jiafu1115]

cc @kamalcph here due to community's email don't allow to attach the image. We can discuss the content in email about the KIP. Thanks

[kamalcph]

@jiafu1115

The already uploaded segments are eligible for deletion from broker. So, when remote storage is down, then those segments can be deleted as per the local retention settings and new segments can occupy those space. This provides more time for the Admin to act when remote storage is down for a longer time.

[jiafu1115]

@kamalcph I think I understand what you mean now. I’ve updated the picture above. Could you help double-check whether we’ve reached the same understanding?
The drawback of this KIP is that, during a long time remote storage outage. it will occupied more disk so that admin may need one extra disk expansion. The max value is the redundant part we saving.
Thus. After the outage recovered. It will come back to the beginning.
Right?

[github-actions]

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

[github-actions]

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

[jiafu1115]

@kamalcph

Hi Kamal:
Sorry to trouble you. I refactored the code based on your suggestion.
0 is the default value only mean upload at once only. -1 is the max lag value as local retention time/size.

It seems the logic is a bit simpler now and the concepts are more clearer, and it’s not as complicated as I originally thought. so I decided to go with your approach. Could you help review it again?

I pasted the test matrix here for your reference to help check the logic.

[kamalcph]

Thanks @jiafu1115 for addressing the review comments. We may have to further simplify the logic.

1. The current logic has multiple negative conditions, which makes it harder to follow. The delayCopy method internally calls notExceededCopyLagTime and notExceededCopyLagSize. Both of these not... methods check whether the threshold is exceeded and then invert the result again. Can we update this logic to use a positive flow instead?

(eg)

List<EnrichedLogSegment> candidateLogSegments(UnifiedLog log, Long fromOffset, Long lastStableOffset) {
...
...
 if (isEligibleForUpload(log.config(), previousSeg, currentTimeMs, totalLogSize, cumulativeSize)) {
      candidateLogSegments.add(new EnrichedLogSegment(previousSeg, currentSeg.baseOffset()));
  } else {
      break;
  }

and update the delay logic to:

private boolean isEligibleForUpload(LogConfig logConfig,
                                LogSegment previousSeg,
                                long currentTimeMs,
                                long totalLogSize,
                                long cumulativeSize) {
  long effectiveCopyLagMs = logConfig.remoteCopyLagMs();
  // derive the value from local retention time when copyLagMs configured to -1
  if (effectiveCopyLagMs == -1L) {
      effectiveCopyLagMs = logConfig.localRetentionMs();
      // if the local retention time is configured to infinite, then configure copyLagMs as infinite
      if (effectiveCopyLagMs == -1L) {
          effectiveCopyLagMs = Long.MAX_VALUE;
      }
  }
  long effectiveCopyLagBytes = logConfig.remoteCopyLagBytes();
  // derive the value from local retention size when copyLagBytes configured to -1
  if (effectiveCopyLagBytes == -1L) {
      effectiveCopyLagBytes = logConfig.localRetentionBytes();
      // if the local retention size is configured to infinite, then configure copyLagBytes as infinite
      if (effectiveCopyLagBytes == -1L) {
          effectiveCopyLagBytes = Long.MAX_VALUE;
      }
  }
  
  try {
      long segmentAgeMs = currentTimeMs - previousSeg.largestTimestamp();
      // If the segment's largestTimestamp is higher than the current time, then allow the segment to upload.
      boolean shouldUploadNow = segmentAgeMs < 0;
      // Upload when either of remote-copy lag time/size breaches the threshold.
      if (!shouldUploadNow) {
          shouldUploadNow = segmentAgeMs >= effectiveCopyLagMs;
      }
      if (!shouldUploadNow) {
          long sizeLagBytes = totalLogSize - cumulativeSize;
          shouldUploadNow = sizeLagBytes >= effectiveCopyLagBytes;
      }
      return shouldUploadNow;
  } catch (IOException e) {
      // in case of any error, allow the segment to upload. Should not block the upload that might hinder the
      // deletion logic
      LOGGER.warn("Failed to get largest timestamp for segment {}, marking it as eligible for upload based on time", previousSeg, e);
      return true;
  }
}

I ran the newly added unit tests and it is passing.

2. Also, move the unit tests from RemoteLogManagerTest to a new RemoteLagCopyTest since the RemoteLogManagerTest is huge with 4k+ lines.

[kamalcph]

LGTM, thanks for addressing the review comments! Left few minor comments.

[kamalcph]

could you also add one test for valid dynamic broker config change? Thanks!

[kamalcph]

Also, move the testDynamicRemoteCopyLagThrowsOnIncorrectConfig test to DynamicBrokerConfigTest.java instead of DynamicBrokerConfigTest.scala.

[kamalcph]

This can be taken up later, we already have one dynamic broker config change test in KafkaConfigTest.

[jiafu1115]

Got it. thanks @kamalcph
Thank you very much for your patient and thorough review.

[chia7712]

@jiafu1115 thanks for this patch. It is cooool

[chia7712]

This is a lingering issue. The local segment with a future timestamp is still NOT deleted, right? Should we allow the deletion of these local files once they have been successfully uploaded to remote storage, even if they contain future timestamps?

[kamalcph]

Should we allow the deletion of these local files once they have been successfully uploaded to remote storage, even if they contain future timestamps?

Currently, it gets deleted based on the local retention bytes limit since the local retention deletion logic follow the similar methods of full deletion logic.

https://sourcegraph.com/r/github.com/apache/kafka/-/blob/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java?L2008

[chia7712]

The current behavior looks like a bug to me, as local segments shouldn't be blocked from deletion once they are already uploaded. Perhaps we could fall back to checking the lastModifiedTime when a segment contains future records. Alternatively, we could introduce a new configuration flag to provide alternative logic for handling future records.

[kamalcph]

shouldn't be blocked from deletion once they are already uploaded. Perhaps we could fall back to checking the lastModifiedTime when a segment contains future records.

Yeah, we can allow this behavior by introducing a config. The segment exist in the remote storage but the user might face slowness in reading the data from remote if they don't have prefetching feature implemented in the Remote Storage Manager. So, better to gate the change in behavior via config and the change applies only when remote storage is enabled on the topic.

[chia7712]

If we all hate -2, there is another approach: reverse the configuration logic.

For example, we could introduce remote.copy.before.deletion.ms instead of remote.copy.lag.ms. If a user sets remote.copy.before.deletion.ms=2hr, it means they want to upload log segments 2 hours before they are scheduled to be deleted from local storage.

The default value would be -1, which dynamically resolves to local.retention.ms. This fully maintains backward compatibility (resulting in immediate upload). Setting it to 0 means the user wants to delay the upload as much as possible, triggering it right before local deletion.

The only side effect is that this kind of "reverse" or "countdown" time calculation is quite rare in the existing Kafka codebase.

[kamalcph]

The only side effect is that this kind of "reverse"

yeah, this is not intuitive when compared with the other configs like retention time/bytes. My suggestion is to:

1. Improve the config documentation about configuring both the values and
2. Add validation to throw an error when remote-copy-lag-time = 0 and remote-copy-lag-bytes != 0 and vice-versa.

[chia7712]

I just realized that I replied to the wrong thread earlier, sorry about that 😢

How about this solution? if we found it contain future record. we use LogSegment#lastModified to compare the time?

Yes, that is an acceptable approach. I've opened https://issues.apache.org/jira/browse/KAFKA-20609 so we can keep discussing there.

yeah, this is not intuitive when compared with the other configs like retention time/bytes. My suggestion is to:

While documentation is the final line of defense for users, it's better to have an intuitive design out of the box. After all, me proposed reverse configuration was rejected precisely because it wasn't intuitive enough

Another way is to align the logic with retention.ms and retention.bytes. Since most users care more about time than size, we could set the default value of the time lag to 0, and the size lag to -1. This way, most users can just adjust the time setting without having to touch the size configuration. WDYT?

[kamalcph]

• Thanks for filing KAFKA-20609 ticket. Fixing this is out of scope of the KIP-1241 as the bug exist before too.

Since most users care more about time than size, we could set the default value of the time lag to 0, and the size lag to -1.

I like the idea to provide out-of-box default values that works in majority of use-cases and retain the eager upload logic. The new default values for remote-copy lag align with the log.retention.hours = 7 days and log.retention.bytes = infinite (-1). It lgtm.

[jiafu1115]

@kamalcph @chia7712 ack
Let me take time to think and handle it. Thanks

[kamalcph]

Opened #22394 to address this comment. PTAL.

[chia7712]

Map<?, ?> ->Map<String, ?>

[jiafu1115]

duplicated with another one. propose the fix in #22363 . thanks.

[chia7712]

Map<?, ?> ->Map<String, ?>

[jiafu1115]

@chia7712 thanks for review. Fix it on #22363

[chia7712]

Should we log a warning or info message when a user configures one lag property but leaves the other at its default value of 0?

[jiafu1115]

@chia7712 Thanks for your view.
How about describing this case in the documentation? If we only log it as a warning, users may ignore it. WDYT?

[chia7712]

Sounds good. We can expand upgrade.md as well.