Repository: apache/kafka
File: core/src/main/java/kafka/log/remote/RemoteLogManager.java (Java)

Code before:
            boolean canProcess = true;
            while (canProcess && epochIterator.hasNext()) {
                Integer epoch = epochIterator.next();
                Iterator<RemoteLogSegmentMetadata> segmentsIterator = remoteLogMetadataManager.listRemoteLogSegments(topicIdPartition, epoch);
                while (canProcess && segmentsIterator.hasNext()) {
                    if (isCancelled() || !isLeader()) {
                        logger.info("Returning from remote log segments cleanup for the remaining segments as the task state is changed.");
                        return;
                    }
                    RemoteLogSegmentMetadata metadata = segmentsIterator.next();

                    // When the log-start-offset is moved by the user, the leader-epoch-checkpoint file gets truncated
                    // as per the log-start-offset. Until the rlm-cleaner-thread runs in the next iteration, those
                    // remote log segments won't be removed. The `isRemoteSegmentWithinLeaderEpoch` validates whether
                    // the epochs present in the segment lies in the checkpoint file. It will always return false
                    // since the checkpoint file was already truncated.
                    boolean isSegmentDeleted = remoteLogRetentionHandler.deleteLogStartOffsetBreachedSegments(
                            metadata, logStartOffset, epochWithOffsets);
                    boolean isValidSegment = false;
                    if (!isSegmentDeleted) {
                        // check whether the segment contains the required epoch range with in the current leader epoch lineage.
                        isValidSegment = isRemoteSegmentWithinLeaderEpochs(metadata, logEndOffset, epochWithOffsets);
                        if (isValidSegment) {
                            isSegmentDeleted =
                                    remoteLogRetentionHandler.deleteRetentionTimeBreachedSegments(metadata) ||
                                            remoteLogRetentionHandler.deleteRetentionSizeBreachedSegments(metadata);
                        }
                    }
                    canProcess = isSegmentDeleted || !isValidSegment;
                }
            }

            // Remove the remote log segments whose segment-leader-epochs are less than the earliest-epoch known
            // to the leader. This will remove the unreferenced segments in the remote storage. This is needed for
            // unclean leader election scenarios as the remote storage can have epochs earlier to the current leader's
            // earliest leader epoch.
            Optional<EpochEntry> earliestEpochEntryOptional = leaderEpochCache.earliestEntry();
            if (earliestEpochEntryOptional.isPresent()) {
                EpochEntry earliestEpochEntry = earliestEpochEntryOptional.get();
                Iterator<Integer> epochsToClean = remoteLeaderEpochs.stream()
                        .filter(remoteEpoch -> remoteEpoch < earliestEpochEntry.epoch)
                        .iterator();
                while (epochsToClean.hasNext()) {
                    int epoch = epochsToClean.next();
                    Iterator<RemoteLogSegmentMetadata> segmentsToBeCleaned = remoteLogMetadataManager.listRemoteLogSegments(topicIdPartition, epoch);
                    while (segmentsToBeCleaned.hasNext()) {
                        if (isCancelled() || !isLeader()) {
                            return;
                        }
                        // No need to update the log-start-offset even though the segment is deleted as these epochs/offsets are earlier to that value.
                        remoteLogRetentionHandler.deleteLogSegmentsDueToLeaderEpochCacheTruncation(earliestEpochEntry, segmentsToBeCleaned.next());

Review comment (suggestion):
not an error since it's recoverable and expected when leadership has transitioned over. Perhaps a info.

Code after:
                    if (isCancelled() || !isLeader()) {
                        logger.info("Returning from remote log segments cleanup for the remaining segments as the task state is changed.");
                        return;
                    }
                    RemoteLogSegmentMetadata metadata = segmentsIterator.next();

                    // When the log-start-offset is moved by the user, the leader-epoch-checkpoint file gets truncated
                    // as per the log-start-offset. Until the rlm-cleaner-thread runs in the next iteration, those
                    // remote log segments won't be removed. The `isRemoteSegmentWithinLeaderEpoch` validates whether
                    // the epochs present in the segment lies in the checkpoint file. It will always return false
                    // since the checkpoint file was already truncated.
                    boolean shouldDeleteSegment = remoteLogRetentionHandler.deleteLogStartOffsetBreachedSegments(
                            metadata, logStartOffset, epochWithOffsets);
                    boolean isValidSegment = false;
                    if (!shouldDeleteSegment) {
                        // check whether the segment contains the required epoch range with in the current leader epoch lineage.
                        isValidSegment = isRemoteSegmentWithinLeaderEpochs(metadata, logEndOffset, epochWithOffsets);
                        if (isValidSegment) {
                            shouldDeleteSegment =
                                    remoteLogRetentionHandler.deleteRetentionTimeBreachedSegments(metadata) ||
                                            remoteLogRetentionHandler.deleteRetentionSizeBreachedSegments(metadata);
                        }
                    }
                    if (shouldDeleteSegment) {
                        segmentsToDelete.add(metadata);
                    }
                    canProcess = shouldDeleteSegment || !isValidSegment;
                }
            }

            // Remove the remote log segments whose segment-leader-epochs are less than the earliest-epoch known
            // to the leader. This will remove the unreferenced segments in the remote storage. This is needed for
            // unclean leader election scenarios as the remote storage can have epochs earlier to the current leader's
            // earliest leader epoch.
            Optional<EpochEntry> earliestEpochEntryOptional = leaderEpochCache.earliestEntry();
            if (earliestEpochEntryOptional.isPresent()) {
                EpochEntry earliestEpochEntry = earliestEpochEntryOptional.get();
                Iterator<Integer> epochsToClean = remoteLeaderEpochs.stream()
                        .filter(remoteEpoch -> remoteEpoch < earliestEpochEntry.epoch)
                        .iterator();
                while (epochsToClean.hasNext()) {
                    int epoch = epochsToClean.next();
                    Iterator<RemoteLogSegmentMetadata> segmentsToBeCleaned = remoteLogMetadataManager.listRemoteLogSegments(topicIdPartition, epoch);
                    while (segmentsToBeCleaned.hasNext()) {
                        if (isCancelled() || !isLeader()) {
                            return;
                        }
                        // No need to update the log-start-offset even though the segment is deleted as these epochs/offsets are earlier to that value.
                        remoteLogRetentionHandler.deleteLogSegmentsDueToLeaderEpochCacheTruncation(earliestEpochEntry, segmentsToBeCleaned.next());
                    }
                }