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

Code before:
     *
     * This method only returns Empty when 1) all messages' offset < startOffing or 2) the log is not empty, but we did not
     * see any message when scanning the log from the indexed position. The latter could happen if the log is truncated
     * after we get the indexed position but before we scan the log from there. In this case we simply return Empty and the
     * caller will need to check on the truncated log and maybe retry or even do the search on another log segment.
     *
     * @param timestampMs The timestamp to search for.
     * @param startingOffset The starting offset to search.
     * @return the timestamp and offset of the first message that meets the requirements. Empty will be returned if there is no such message.
     */
    public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long timestampMs, long startingOffset) throws IOException {
        // Get the index entry with a timestamp less than or equal to the target timestamp
        TimestampOffset timestampOffset = timeIndex().lookup(timestampMs);
        int position = offsetIndex().lookup(Math.max(timestampOffset.offset, startingOffset)).position;

        // Search the timestamp
        return Optional.ofNullable(log.searchForTimestamp(timestampMs, position, startingOffset));
    }

    /**
     * Close this log segment
     */
    @Override
    public void close() throws IOException {
        if (maxTimestampAndOffsetSoFar != TimestampOffset.UNKNOWN)
            Utils.swallow(LOGGER, Level.WARN, "maybeAppend", () -> timeIndex().maybeAppend(maxTimestampSoFar(), shallowOffsetOfMaxTimestampSoFar(), true));
        Utils.closeQuietly(lazyOffsetIndex, "offsetIndex", LOGGER);
        Utils.closeQuietly(lazyTimeIndex, "timeIndex", LOGGER);
        Utils.closeQuietly(log, "log", LOGGER);
        Utils.closeQuietly(txnIndex, "txnIndex", LOGGER);
    }

    /**
     * Close file handlers used by the log segment but don't write to disk. This is used when the disk may have failed
     */
    void closeHandlers() {
        Utils.swallow(LOGGER, Level.WARN, "offsetIndex", () -> lazyOffsetIndex.closeHandler());
        Utils.swallow(LOGGER, Level.WARN, "timeIndex", () -> lazyTimeIndex.closeHandler());
        Utils.swallow(LOGGER, Level.WARN, "log", () -> log.closeHandlers());
        Utils.closeQuietly(txnIndex, "txnIndex", LOGGER);
    }

    /**
     * Delete this log segment from the filesystem.
     */
    public void deleteIfExists() throws IOException {
        try {
            Utils.tryAll(List.of(
                () -> deleteTypeIfExists(() -> log.deleteIfExists(), "log", log.file(), true),
                () -> deleteTypeIfExists(() -> lazyOffsetIndex.deleteIfExists(), "offset index", offsetIndexFile(), true),
                () -> deleteTypeIfExists(() -> lazyTimeIndex.deleteIfExists(), "time index", timeIndexFile(), true),

Review comment (suggestion):
I think writing the max timestamp has a similar issue. the last entry is assumed to be the max timestamp after restarting. Hence, it needs to rebuild the index if the "true" max timestamp is not stored correctly

I open KAFKA-19428 to trace it

Code after:
     *
     * This method only returns Empty when 1) all messages' offset < startOffing or 2) the log is not empty, but we did not
     * see any message when scanning the log from the indexed position. The latter could happen if the log is truncated
     * after we get the indexed position but before we scan the log from there. In this case we simply return Empty and the
     * caller will need to check on the truncated log and maybe retry or even do the search on another log segment.
     *
     * @param timestampMs The timestamp to search for.
     * @param startingOffset The starting offset to search.
     * @return the timestamp and offset of the first message that meets the requirements. Empty will be returned if there is no such message.
     */
    public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long timestampMs, long startingOffset) throws IOException {
        // Get the index entry with a timestamp less than or equal to the target timestamp
        TimestampOffset timestampOffset = timeIndex().lookup(timestampMs);
        int position = offsetIndex().lookup(Math.max(timestampOffset.offset, startingOffset)).position;

        // Search the timestamp
        return Optional.ofNullable(log.searchForTimestamp(timestampMs, position, startingOffset));
    }

    /**
     * Close this log segment
     */
    @Override
    public void close() throws IOException {
        if (maxTimestampAndOffsetSoFar != TimestampOffset.UNKNOWN)
            Utils.swallow(LOGGER, Level.WARN, "maybeAppend", () -> timeIndex().maybeAppend(maxTimestampSoFar(), shallowOffsetOfMaxTimestampSoFar(), true));
        Utils.closeAll(lazyOffsetIndex, lazyTimeIndex, log, txnIndex);
    }

    /**
     * Close file handlers used by the log segment but don't write to disk. This is used when the disk may have failed
     */
    void closeHandlers() {
        Utils.swallow(LOGGER, Level.WARN, "offsetIndex", () -> lazyOffsetIndex.closeHandler());
        Utils.swallow(LOGGER, Level.WARN, "timeIndex", () -> lazyTimeIndex.closeHandler());
        Utils.swallow(LOGGER, Level.WARN, "log", () -> log.closeHandlers());
        Utils.closeQuietly(txnIndex, "txnIndex", LOGGER);
    }

    /**
     * Delete this log segment from the filesystem.
     */
    public void deleteIfExists() throws IOException {
        try {
            Utils.tryAll(List.of(
                () -> deleteTypeIfExists(() -> log.deleteIfExists(), "log", log.file(), true),
                () -> deleteTypeIfExists(() -> lazyOffsetIndex.deleteIfExists(), "offset index", offsetIndexFile(), true),
                () -> deleteTypeIfExists(() -> lazyTimeIndex.deleteIfExists(), "time index", timeIndexFile(), true),
                () -> deleteTypeIfExists(() -> txnIndex.deleteIfExists(), "transaction index", txnIndex.file(), false)));
        } catch (Throwable t) {
            if (t instanceof IOException)