package org.apache.kafka.storage.internals.log

... public class LogSegment implements Closeable { ... private final FileRecords log; ... /** * Append the given messages starting with the given offset. Add * an entry to the index if needed. * * It is assumed this method is being called from within a lock, it is not thread-safe otherwise. * * @param largestOffset The last offset in the message set * @param largestTimestampMs The largest timestamp in the message set. * @param shallowOffsetOfMaxTimestamp The last offset of earliest batch with max timestamp in the messages to append. * @param records The log entries to append. * @throws LogSegmentOffsetOverflowException if the largest offset causes index offset overflow */
    public void append(long largestOffset, long largestTimestampMs, long shallowOffsetOfMaxTimestamp, MemoryRecords records) throws IOException { if (records.sizeInBytes() > 0) { LOGGER.trace("Inserting {} bytes at end offset {} at position {} with largest timestamp {} at offset {}", records.sizeInBytes(), largestOffset, log.sizeInBytes(), largestTimestampMs, shallowOffsetOfMaxTimestamp); int physicalPosition = log.sizeInBytes(); if (physicalPosition == 0) rollingBasedTimestamp = OptionalLong.of(largestTimestampMs); ensureOffsetInRange(largestOffset); // append the messages
            long appendedBytes = log.append(records); LOGGER.trace("Appended {} to {} at end offset {}", appendedBytes, log.file(), largestOffset); // Update the in memory max timestamp and corresponding offset.
            if (largestTimestampMs > maxTimestampSoFar()) { maxTimestampAndOffsetSoFar = new TimestampOffset(largestTimestampMs, shallowOffsetOfMaxTimestamp); } // append an entry to the index (if needed)
            // 稀疏索引，有一定的间隔。可以减少索引量
            if (bytesSinceLastIndexEntry > indexIntervalBytes) { offsetIndex().append(largestOffset, physicalPosition); timeIndex().maybeAppend(maxTimestampSoFar(), shallowOffsetOfMaxTimestampSoFar()); bytesSinceLastIndexEntry = 0; } bytesSinceLastIndexEntry += records.sizeInBytes(); } } ... }

package org.apache.kafka.common.record;

...

public class FileRecords extends AbstractRecords implements Closeable {
     ...  
     private final FileChannel channel;
     ....

    public int append(MemoryRecords records) throws IOException {
        if (records.sizeInBytes() > Integer.MAX_VALUE - size.get())
            throw new IllegalArgumentException("Append of size " + records.sizeInBytes() +
                    " bytes is too large for segment with current file position at " + size.get());

        int written = records.writeFullyTo(channel);
        size.getAndAdd(written);
        return written;
    }
  
    ...


}

package org.apache.kafka.common.record;
....

public class MemoryRecords extends AbstractRecords {
      ...
      private final ByteBuffer buffer;
      ...

          /**
     * Write all records to the given channel (including partial records).
     * @param channel The channel to write to
     * @return The number of bytes written
     * @throws IOException For any IO errors writing to the channel
     */
    public int writeFullyTo(GatheringByteChannel channel) throws IOException {
        buffer.mark();
        int written = 0;
        while (written < sizeInBytes())
            written += channel.write(buffer);
        buffer.reset();
        return written;
    }

    ....
}