Repository: apache/kafka
File: storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java (Java)

Code before:
                            segment.deleteIfExists();
                        }
                        return null;
                    });
        };
        if (asyncDelete) {
            scheduler.scheduleOnce("delete-file", deleteSegments, config.fileDeleteDelayMs);
        } else {
            deleteSegments.run();
        }
    }

    public static LogSegment createNewCleanedSegment(File dir, LogConfig logConfig, long baseOffset) throws IOException {
        LogSegment.deleteIfExists(dir, baseOffset, CLEANED_FILE_SUFFIX);
        return LogSegment.open(dir, baseOffset, logConfig, Time.SYSTEM, false, logConfig.initFileSize(), logConfig.preallocate, CLEANED_FILE_SUFFIX);
    }

    /**
     * Split a segment into one or more segments such that there is no offset overflow in any of them. The
     * resulting segments will contain the exact same messages that are present in the input segment. On successful
     * completion of this method, the input segment will be deleted and will be replaced by the resulting new segments.
     * See replaceSegments for recovery logic, in case the broker dies in the middle of this operation.
     * <br/>
     * Note that this method assumes we have already determined that the segment passed in contains records that cause
     * offset overflow.
     * <br/>
     * The split logic overloads the use of .clean files that LogCleaner typically uses to make the process of replacing
     * the input segment with multiple new segments atomic and recoverable in the event of a crash. See replaceSegments
     * and completeSwapOperations for the implementation to make this operation recoverable on crashes.
     *
     * @param segment Segment to split
     * @param existingSegments The existing segments of the log
     * @param dir The directory in which the log will reside
     * @param topicPartition The topic
     * @param config The log configuration settings
     * @param scheduler The thread pool scheduler used for background actions
     * @param logDirFailureChannel The LogDirFailureChannel to asynchronously handle log dir failure
     * @param logPrefix The logging prefix
     * @return List of new segments that replace the input segment
     */
    public static SplitSegmentResult splitOverflowedSegment(LogSegment segment,
                                                     LogSegments existingSegments,
                                                     File dir,
                                                     TopicPartition topicPartition,
                                                     LogConfig config,
                                                     Scheduler scheduler,
                                                     LogDirFailureChannel logDirFailureChannel,
                                                     String logPrefix) throws IOException {
        require(isLogFile(segment.log().file()), "Cannot split file " + segment.log().file().getAbsoluteFile());
        require(segment.hasOverflow(), "Split operation is only permitted for segments with overflow, and the problem path is " + segment.log().file().getAbsoluteFile());


Review comment (suggestion):
This would read better with a `<p>` rather than `<br/>` I think. 

Code after:
                            segment.deleteIfExists();
                        }
                        return null;
                    });
        };
        if (asyncDelete) {
            scheduler.scheduleOnce("delete-file", deleteSegments, config.fileDeleteDelayMs);
        } else {
            deleteSegments.run();
        }
    }

    public static LogSegment createNewCleanedSegment(File dir, LogConfig logConfig, long baseOffset) throws IOException {
        LogSegment.deleteIfExists(dir, baseOffset, CLEANED_FILE_SUFFIX);
        return LogSegment.open(dir, baseOffset, logConfig, Time.SYSTEM, false, logConfig.initFileSize(), logConfig.preallocate, CLEANED_FILE_SUFFIX);
    }

    /**
     * Split a segment into one or more segments such that there is no offset overflow in any of them. The
     * resulting segments will contain the exact same messages that are present in the input segment. On successful
     * completion of this method, the input segment will be deleted and will be replaced by the resulting new segments.
     * See replaceSegments for recovery logic, in case the broker dies in the middle of this operation.
     * <p>
     * Note that this method assumes we have already determined that the segment passed in contains records that cause
     * offset overflow.
     * <p>
     * The split logic overloads the use of .clean files that LogCleaner typically uses to make the process of replacing
     * the input segment with multiple new segments atomic and recoverable in the event of a crash. See replaceSegments
     * and completeSwapOperations for the implementation to make this operation recoverable on crashes.
     *
     * @param segment Segment to split
     * @param existingSegments The existing segments of the log
     * @param dir The directory in which the log will reside
     * @param topicPartition The topic
     * @param config The log configuration settings
     * @param scheduler The thread pool scheduler used for background actions
     * @param logDirFailureChannel The LogDirFailureChannel to asynchronously handle log dir failure
     * @param logPrefix The logging prefix
     * @return List of new segments that replace the input segment
     */
    public static SplitSegmentResult splitOverflowedSegment(LogSegment segment,
                                                     LogSegments existingSegments,
                                                     File dir,
                                                     TopicPartition topicPartition,
                                                     LogConfig config,
                                                     Scheduler scheduler,
                                                     LogDirFailureChannel logDirFailureChannel,
                                                     String logPrefix) throws IOException {
        require(isLogFile(segment.log().file()), "Cannot split file " + segment.log().file().getAbsoluteFile());
        require(segment.hasOverflow(), "Split operation is only permitted for segments with overflow, and the problem path is " + segment.log().file().getAbsoluteFile());