Kafka存储引擎深度解析：日志段管理与高性能存储技术剖析

graph TB
    subgraph "Kafka存储引擎分层架构"
        subgraph "逻辑层 Logical Layer"
            T[Topic 主题]
            P1[Partition 0]
            P2[Partition 1]
            P3[Partition N]
        end
        
        subgraph "管理层 Management Layer"
            LM[LogManager 日志管理器]
            UL[UnifiedLog 统一日志]
            PM[PartitionManager 分区管理器]
        end
        
        subgraph "存储层 Storage Layer"
            LS1[LogSegment 日志段1]
            LS2[LogSegment 日志段2]
            LSN[LogSegment 日志段N]
        end
        
        subgraph "文件层 File Layer"
            subgraph "日志段1文件组"
                LOG1[00000000000000000000.log]
                IDX1[00000000000000000000.index]
                TIM1[00000000000000000000.timeindex]
                TXN1[00000000000000000000.txnindex]
                SNAP1[00000000000000000000.snapshot]
            end
            
            subgraph "日志段2文件组"
                LOG2[00000000000001000000.log]
                IDX2[00000000000001000000.index]
                TIM2[00000000000001000000.timeindex]
                TXN2[00000000000001000000.txnindex]
                SNAP2[00000000000001000000.snapshot]
            end
        end
        
        subgraph "系统层 System Layer"
            PC[Page Cache 页缓存]
            FS[File System 文件系统]
            DIO[Direct I/O 直接IO]
        end
        
        %% 连接关系
        T --> P1
        T --> P2
        T --> P3
        
        LM --> UL
        PM --> UL
        UL --> LS1
        UL --> LS2
        UL --> LSN
        
        LS1 --> LOG1
        LS1 --> IDX1
        LS1 --> TIM1
        LS1 --> TXN1
        LS1 --> SNAP1
        
        LS2 --> LOG2
        LS2 --> IDX2
        LS2 --> TIM2
        LS2 --> TXN2
        LS2 --> SNAP2
        
        LOG1 --> PC
        LOG2 --> PC
        PC --> FS
        FS --> DIO
    end
    
    style LM fill:#e1f5fe
    style LS1 fill:#e8f5e8
    style LS2 fill:#e8f5e8
    style PC fill:#f3e5f5
    style FS fill:#fff3e0

# 日志段文件组织示例（基础偏移量为0）
00000000000000000000.log        # 消息数据文件（主文件）
00000000000000000000.index      # 偏移量索引文件
00000000000000000000.timeindex  # 时间戳索引文件
00000000000000000000.txnindex   # 事务索引文件
00000000000000000000.snapshot   # Producer状态快照文件

# 下一个日志段（基础偏移量为1000000）
00000000000001000000.log        # 新的消息数据文件
00000000000001000000.index      # 对应的偏移量索引
00000000000001000000.timeindex  # 对应的时间戳索引
# ... 其他相关文件

/**
 * UnifiedLog - Kafka的统一日志实现
 * 管理单个分区的完整日志，包括活跃段和历史段
 */
public class UnifiedLog implements Closeable {
    
    // 核心组件
    private final File dir;                        // 日志目录
    private final LogConfig config;                // 日志配置
    private final Scheduler scheduler;             // 调度器
    private final Time time;                       // 时间提供器
    private final TopicPartition topicPartition;   // 主题分区
    
    // 日志段管理
    private final LogSegments segments;            // 日志段集合
    private volatile LogSegment activeSegment;     // 当前活跃段
    
    // 关键位移管理
    private volatile Long logStartOffset;          // 日志起始偏移量（LSO）
    private volatile Long logEndOffset;            // 日志结束偏移量（LEO）
    private volatile Long highWatermark;           // 高水位（HW）
    private volatile Long lastStableOffset;        // 最后稳定偏移量（LSO）
    
    // 生产者状态管理
    private final ProducerStateManager producerStateManager;  // 生产者状态管理器
    private final LeaderEpochFileCache leaderEpochCache;      // Leader纪元缓存

    /**
     * 向日志追加记录批次
     * 这是Kafka写入性能的核心实现
     * 
     * @param records 要追加的记录批次
     * @param origin 写入来源（客户端、副本同步等）
     * @param interBrokerProtocolVersion 代理间协议版本
     * @param assignOffsets 是否分配偏移量
     * @param requestLocal 请求本地缓存
     * @return 追加结果信息
     */
    public LogAppendInfo append(MemoryRecords records,
                               AppendOrigin origin,
                               Integer interBrokerProtocolVersion,
                               boolean assignOffsets,
                               int leaderEpoch,
                               Optional<BufferSupplier> bufferSupplier,
                               RequestLocal requestLocal,
                               boolean ignoreRecordSize) throws IOException {
        
        long startTimeMs = time.milliseconds();
        
        // 1. 验证记录批次的有效性
        analyzeAndValidateRecords(records, origin);
        
        // 2. 检查日志段是否需要滚动
        maybeRoll(activeSegment, records, config.segmentMs - (time.milliseconds() - activeSegment.created),
                 config.segmentBytes - activeSegment.size(), leaderEpoch);
        
        // 3. 为记录分配偏移量（如果需要）
        long offset = nextOffsetMetadata().messageOffset;
        if (assignOffsets) {
            validateMessages(records, origin, offset, leaderEpoch);
            offset = assignOffsetsToRecords(records, offset);
        }
        
        // 4. 更新生产者状态（用于幂等性和事务）
        for (RecordBatch batch : records.batches()) {
            if (batch.hasProducerId()) {
                producerStateManager.update(batch);
            }
        }
        
        // 5. 实际写入文件
        activeSegment.append(offset, records);
        
        // 6. 更新日志结束偏移量（LEO）
        updateLogEndOffset(offset + 1);
        
        // 7. 更新高水位（如果是Leader）
        if (origin != AppendOrigin.Replication) {
            maybeIncrementHighWatermark(activeSegment.readNextOffset());
        }
        
        // 8. 构建追加信息
        LogAppendInfo appendInfo = new LogAppendInfo(
            firstOffset,           // 第一条消息偏移量
            offset,               // 最后一条消息偏移量
            time.milliseconds(),  // 追加时间
            logStartOffset,       // 日志起始偏移量
            records.validBytes(), // 有效字节数
            records.records().size(), // 记录数量
            MonotonicClock.INSTANCE.hiResClockMs() - startTimeMs, // 追加耗时
            RecordBatch.NO_TIMESTAMP, // 分片时间戳
            Collections.emptyList()   // 压缩记录
        );
        
        trace("追加了 {} 字节到日志 {}，新的LEO: {}", records.sizeInBytes(), topicPartition, offset + 1);
        
        return appendInfo;
    }

    /**
     * 从日志读取消息
     * 实现高效的消息检索，支持零拷贝优化
     * 
     * @param startOffset 起始偏移量
     * @param maxLength 最大读取长度
     * @param isolation 隔离级别
     * @param minOneMessage 是否至少返回一条消息
     * @return 读取到的数据信息
     */
    public FetchDataInfo read(long startOffset,
                             int maxLength,
                             FetchIsolation isolation,
                             boolean minOneMessage) throws IOException {
        
        trace("从偏移量 {} 读取最多 {} 字节数据", startOffset, maxLength);
        
        // 1. 检查偏移量有效性
        if (startOffset > logEndOffset) {
            return FetchDataInfo.EMPTY;
        }
        
        // 2. 查找包含起始偏移量的日志段
        LogSegment segment = segments.floorSegment(startOffset);
        if (segment == null) {
            // 偏移量在日志起始偏移量之前
            throw new OffsetOutOfRangeException("请求的偏移量 " + startOffset + 
                " 小于日志起始偏移量 " + logStartOffset);
        }
        
        // 3. 计算最大可读偏移量（根据隔离级别）
        long maxOffsetMetadata = isolation == FetchIsolation.TXN_COMMITTED 
            ? lastStableOffset 
            : logEndOffset;
        
        // 4. 从段中读取数据
        FetchDataInfo fetchDataInfo = segment.read(startOffset, maxLength, 
            maxOffsetMetadata, minOneMessage);
        
        // 5. 如果当前段读取的数据不足且不是最后一个段，尝试从下一个段继续读取
        if (fetchDataInfo.records.sizeInBytes() < maxLength && segment != activeSegment) {
            LogSegment nextSegment = segments.higherSegment(segment.baseOffset());
            if (nextSegment != null) {
                int remainingLength = maxLength - fetchDataInfo.records.sizeInBytes();
                FetchDataInfo nextFetchInfo = nextSegment.read(nextSegment.baseOffset(), 
                    remainingLength, maxOffsetMetadata, false);
                
                // 合并两个段的数据
                if (nextFetchInfo.records.sizeInBytes() > 0) {
                    List<FileRecords> combinedRecords = Arrays.asList(
                        fetchDataInfo.records, nextFetchInfo.records);
                    fetchDataInfo = new FetchDataInfo(
                        fetchDataInfo.fetchOffsetMetadata,
                        new MultiRecords(combinedRecords)
                    );
                }
            }
        }
        
        trace("读取完成：起始偏移量={}, 数据大小={}", startOffset, fetchDataInfo.records.sizeInBytes());
        return fetchDataInfo;
    }

    /**
     * 检查是否需要滚动到新的日志段
     * 基于时间、大小、偏移量等多种条件判断
     */
    private boolean maybeRoll(LogSegment segment, MemoryRecords records, 
                             long timeUntilRoll, long sizeUntilRoll, int leaderEpoch) {
        
        int recordSize = records.sizeInBytes();
        
        // 检查各种滚动条件
        boolean shouldRoll = segment.shouldRoll(new RollParams(
            config.segmentMs,                    // 最大段时间
            config.segmentBytes,                 // 最大段大小
            timeUntilRoll,                      // 距离时间滚动的剩余时间
            sizeUntilRoll,                      // 距离大小滚动的剩余空间
            recordSize,                         // 当前记录大小
            time.milliseconds(),                // 当前时间
            records.batches().iterator().next().maxTimestamp() // 最大时间戳
        ));
        
        if (shouldRoll) {
            debug("基于滚动条件创建新的日志段，当前段: {}, 记录大小: {}", 
                 segment.baseOffset(), recordSize);
            
            // 创建新的活跃段
            LogSegment newSegment = roll(Some(leaderEpoch));
            
            // 异步刷新旧段到磁盘
            scheduler.schedule("flush-log-segment", () -> {
                segment.flush();
            }, 0);
            
            return true;
        }
        
        return false;
    }
    
    /**
     * 创建新的日志段
     * 实现日志的滚动机制
     */
    private LogSegment roll(Optional<Integer> expectedNextOffset) throws IOException {
        long startMs = time.milliseconds();
        
        // 1. 计算新段的基础偏移量
        long newSegmentBaseOffset = expectedNextOffset.orElse(logEndOffset);
        
        // 2. 关闭当前活跃段的写入
        activeSegment.closeForRecordAppends();
        
        // 3. 创建新的日志段文件
        LogSegment newSegment = LogSegment.open(
            dir,                                 // 日志目录
            newSegmentBaseOffset,               // 基础偏移量
            config,                             // 日志配置
            time,                               // 时间提供器
            config.initFileSize(),              // 初始文件大小
            config.preallocate()                // 是否预分配
        );
        
        // 4. 将新段添加到段集合中
        segments.add(newSegment);
        
        // 5. 更新活跃段引用
        activeSegment = newSegment;
        
        // 6. 更新监控指标
        info("创建新的日志段 {} 于目录 {}，耗时 {} ms", 
             newSegmentBaseOffset, dir.getAbsolutePath(), time.milliseconds() - startMs);
        
        return newSegment;
    }
}

/**
 * FileRecords - 通过...实现
 * 利用零拷贝技术实现高效的数据传输
 */
public class FileRecords implements Records {
    
    private final File file;                       // 底层文件
    private final FileChannel channel;             // 文件通道
    private final int start;                       // 起始位置
    private final int end;                         // 结束位置
    private final boolean isSlice;                 // 是否为切片
    
    /**
     * 使用零拷贝技术传输数据到GatheringByteChannel
     * 这是Kafka高性能网络传输的核心实现
     * 
     * @param destChannel 目标通道（通常是SocketChannel）
     * @param position 起始位置
     * @param count 传输字节数
     * @return 实际传输的字节数
     */
    @Override
    public long transferTo(GatheringByteChannel destChannel, long position, long count) throws IOException {
        if (position < 0 || count < 0) {
            throw new IllegalArgumentException("位置和计数必须非负");
        }
        
        if (position > sizeInBytes()) {
            return 0;
        }
        
        // 计算实际传输大小
        long actualCount = Math.min(count, sizeInBytes() - position);
        long startPosition = start + position;
        
        try {
            // 使用FileChannel.transferTo实现零拷贝传输
            // 数据直接从文件的内核缓冲区传输到网络Socket的内核缓冲区
            // 避免了用户空间的内存拷贝，显著提升性能
            return channel.transferTo(startPosition, actualCount, destChannel);
        } catch (IOException e) {
            // 在某些操作系统上，transferTo可能有bug或限制
            // 降级到传统的读写方式
            warn("零拷贝传输失败，降级到传统传输方式: {}", e.getMessage());
            return transferToFallback(destChannel, startPosition, actualCount);
        }
    }
    
    /**
     * 传统方式的数据传输（当零拷贝失败时的降级方案）
     */
    private long transferToFallback(GatheringByteChannel destChannel, 
                                   long position, long count) throws IOException {
        ByteBuffer buffer = ByteBuffer.allocate(Math.min(8192, (int) count)); // 8KB缓冲区
        long transferred = 0;
        
        while (transferred < count) {
            buffer.clear();
            long toRead = Math.min(buffer.remaining(), count - transferred);
            
            // 从文件读取数据到缓冲区
            int bytesRead = channel.read(buffer, position + transferred);
            if (bytesRead <= 0) {
                break;
            }
            
            buffer.flip();
            
            // 将缓冲区数据写入目标通道
            while (buffer.hasRemaining()) {
                int bytesWritten = destChannel.write(buffer);
                if (bytesWritten <= 0) {
                    throw new IOException("无法写入数据到目标通道");
                }
                transferred += bytesWritten;
            }
        }
        
        return transferred;
    }
    
    /**
     * 使用内存映射读取文件数据
     * 利用操作系统的页缓存机制提高读取性能
     */
    public ByteBuffer readInto(ByteBuffer buffer, int position) throws IOException {
        if (position < 0 || position >= sizeInBytes()) {
            throw new IllegalArgumentException("读取位置超出范围");
        }
        
        // 使用内存映射读取文件内容
        MappedByteBuffer mappedBuffer = channel.map(
            FileChannel.MapMode.READ_ONLY,      // 只读映射
            start + position,                   // 映射起始位置
            Math.min(buffer.remaining(), sizeInBytes() - position) // 映射大小
        );
        
        // 将映射的数据复制到目标缓冲区
        int originalLimit = mappedBuffer.limit();
        try {
            if (buffer.remaining() < mappedBuffer.remaining()) {
                mappedBuffer.limit(mappedBuffer.position() + buffer.remaining());
            }
            buffer.put(mappedBuffer);
        } finally {
            mappedBuffer.limit(originalLimit);
        }
        
        return buffer;
    }
}

/**
 * 零拷贝技术的底层实现原理
 * 
 * 传统数据传输路径：
 * 磁盘 -> 内核缓冲区 -> 用户空间缓冲区 -> Socket内核缓冲区 -> 网卡
 * (涉及4次拷贝：2次DMA拷贝 + 2次CPU拷贝)
 * 
 * 零拷贝优化路径：
 * 磁盘 -> 内核缓冲区 -> Socket内核缓冲区 -> 网卡
 * (仅涉及2次DMA拷贝，消除了CPU拷贝)
 * 
 * 关键系统调用：
 * - sendfile()：Linux系统的零拷贝实现
 * - transferTo()：Java NIO的零拷贝API封装
 */

/**
 * PageCache优化策略实现
 * 充分利用操作系统的页缓存机制提升I/O性能
 */
public class PageCacheOptimization {
    
    /**
     * 页缓存友好的读取策略
     * 通过顺序访问模式最大化页缓存命中率
     */
    public static class SequentialReadOptimizer {
        
        private static final int READ_AHEAD_SIZE = 65536; // 64KB预读大小
        
        /**
         * 实现页缓存友好的顺序读取
         * 利用操作系统的预读机制提高性能
         */
        public ByteBuffer readWithPageCacheOptimization(FileChannel channel, 
                                                        long position, 
                                                        int size) throws IOException {
            // 1. 对齐到页缓存边界（通常为4KB）
            long alignedPosition = (position / 4096) * 4096;
            int alignedSize = (int) Math.min(
                ((size + 4095) / 4096) * 4096,  // 向上对齐到4KB边界
                channel.size() - alignedPosition
            );
            
            // 2. 使用内存映射进行读取
            MappedByteBuffer mappedBuffer = channel.map(
                FileChannel.MapMode.READ_ONLY,
                alignedPosition,
                alignedSize
            );
            
            // 3. 触发预读（建议操作系统预加载后续页）
            mappedBuffer.load(); // 将映射的页加载到内存
            
            // 4. 定位到实际需要的数据
            int offset = (int) (position - alignedPosition);
            mappedBuffer.position(offset);
            mappedBuffer.limit(offset + size);
            
            return mappedBuffer.slice();
        }
        
        /**
         * 批量预读策略
         * 一次性读取多个消息批次，减少系统调用开销
         */
        public List<ByteBuffer> batchRead(FileChannel channel, 
                                         List<Long> positions, 
                                         List<Integer> sizes) throws IOException {
            List<ByteBuffer> results = new ArrayList<>();
            
            // 合并连续的读取请求
            List<ReadRequest> mergedRequests = mergeContiguousReads(positions, sizes);
            
            for (ReadRequest request : mergedRequests) {
                ByteBuffer buffer = readWithPageCacheOptimization(
                    channel, request.position, request.size);
                
                // 根据原始请求拆分数据
                List<ByteBuffer> splitBuffers = splitBuffer(buffer, request.originalSizes);
                results.addAll(splitBuffers);
            }
            
            return results;
        }
    }
    
    /**
     * 页缓存监控和调优
     */
    public static class PageCacheMonitor {
        
        /**
         * 监控页缓存命中率
         * 通过/proc/vmstat等系统接口获取页缓存统计信息
         */
        public PageCacheStats getPageCacheStats() {
            try {
                // 读取系统页缓存统计信息
                List<String> vmstatLines = Files.readAllLines(Paths.get("/proc/vmstat"));
                
                long cacheHits = 0;
                long cacheMisses = 0;
                
                for (String line : vmstatLines) {
                    if (line.startsWith("pgmajfault")) {
                        // 主页错误（从磁盘读取）
                        cacheMisses = Long.parseLong(line.split("\\s+")[1]);
                    } else if (line.startsWith("pgpgin")) {
                        // 页面换入
                        cacheHits = Long.parseLong(line.split("\\s+")[1]);
                    }
                }
                
                double hitRate = (double) cacheHits / (cacheHits + cacheMisses);
                
                return new PageCacheStats(cacheHits, cacheMisses, hitRate);
                
            } catch (IOException e) {
                warn("无法读取页缓存统计信息", e);
                return PageCacheStats.UNKNOWN;
            }
        }
        
        /**
         * 页缓存预热策略
         * 在服务启动时预先加载热点数据到页缓存
         */
        public void warmupPageCache(List<File> logFiles) {
            info("开始页缓存预热，文件数量: {}", logFiles.size());
            
            ExecutorService executor = Executors.newFixedThreadPool(
                Runtime.getRuntime().availableProcessors());
            
            try {
                List<CompletableFuture<Void>> futures = logFiles.stream()
                    .map(file -> CompletableFuture.runAsync(() -> {
                        try (FileChannel channel = FileChannel.open(file.toPath(), StandardOpenOption.READ)) {
                            // 逐页读取文件内容，触发页缓存加载
                            long fileSize = channel.size();
                            int pageSize = 4096;
                            
                            for (long pos = 0; pos < fileSize; pos += pageSize) {
                                ByteBuffer buffer = ByteBuffer.allocate(pageSize);
                                channel.read(buffer, pos);
                            }
                            
                            debug("完成文件 {} 的页缓存预热", file.getName());
                        } catch (IOException e) {
                            error("预热文件 {} 失败", file.getName(), e);
                        }
                    }, executor))
                    .collect(Collectors.toList());
                
                // 等待所有预热任务完成
                CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
                    .get(30, TimeUnit.SECONDS);
                
                info("页缓存预热完成");
                
            } catch (Exception e) {
                error("页缓存预热过程中发生异常", e);
            } finally {
                executor.shutdown();
            }
        }
    }
}

/**
 * Kafka中的关键偏移量概念解析
 * 
 * LEO (Log End Offset): 日志结束偏移量
 * - 表示日志中下一条消息将被分配的偏移量
 * - 每当有新消息写入时LEO会递增
 * - 每个副本都有自己的LEO
 * 
 * HW (High Watermark): 高水位偏移量  
 * - 表示消费者可以看到的消息的最大偏移量
 * - 只有当消息被ISR中的所有副本确认后，HW才会前移
 * - 保证了数据的一致性和可见性
 * 
 * LSO (Log Start Offset): 日志起始偏移量
 * - 表示日志中最早消息的偏移量
 * - 随着日志清理和压缩会发生变化
 * 
 * Last Stable Offset: 最后稳定偏移量
 * - 用于事务隔离，表示最后一个已提交事务的偏移量
 */

/**
 * PartitionMetadata - 分区元数据管理
 * 维护分区的LEO、HW等关键偏移量信息
 */
public class Partition {
    
    // 关键偏移量
    private volatile LogOffsetMetadata localLogStartOffset;    // 本地日志起始偏移量
    private volatile LogOffsetMetadata logEndOffset;          // 日志结束偏移量（LEO）
    private volatile LogOffsetMetadata highWatermark;         // 高水位偏移量（HW）
    private volatile LogOffsetMetadata lastStableOffset;      // 最后稳定偏移量
    
    // 副本管理
    private final Map<Integer, Replica> assignedReplicas;     // 分配的副本映射
    private final Set<Integer> inSyncReplicaIds;              // ISR副本ID集合
    private final ReentrantReadWriteLock leaderIsrUpdateLock; // ISR更新锁
    
    /**
     * 更新Follower副本的LEO
     * 由ReplicaFetcherThread调用，用于同步副本状态
     * 
     * @param replicaId 副本ID
     * @param newLogEndOffset 新的日志结束偏移量
     * @param currentTimeMs 当前时间戳
     * @return 是否成功更新
     */
    public boolean updateReplicaLogReadResult(int replicaId, 
                                            LogReadResult readResult,
                                            long currentTimeMs) {
        leaderIsrUpdateLock.readLock().lock();
        try {
            Replica replica = assignedReplicas.get(replicaId);
            if (replica == null) {
                warn("尝试更新不存在的副本 {} 的LEO", replicaId);
                return false;
            }
            
            // 更新副本的LEO
            LogOffsetMetadata newEndOffset = readResult.info.fetchOffsetMetadata;
            replica.updateLogReadResult(readResult, currentTimeMs);
            
            debug("更新副本 {} 的LEO为 {}", replicaId, newEndOffset.messageOffset);
            
            // 检查是否需要更新高水位
            boolean hwUpdated = maybeIncrementHighWatermark(replica, currentTimeMs);
            
            return hwUpdated;
        } finally {
            leaderIsrUpdateLock.readLock().unlock();
        }
    }
    
    /**
     * 尝试增加高水位
     * 当ISR中所有副本的LEO都达到新的水位时，推进HW
     */
    private boolean maybeIncrementHighWatermark(Replica replica, long currentTimeMs) {
        // 计算ISR中所有副本的最小LEO
        long newHighWatermark = inSyncReplicaIds.stream()
            .mapToLong(replicaId -> {
                Replica r = assignedReplicas.get(replicaId);
                return r != null ? r.logEndOffset().messageOffset : Long.MAX_VALUE;
            })
            .min()
            .orElse(highWatermark.messageOffset);
        
        // 高水位只能前移，不能后退
        if (newHighWatermark > highWatermark.messageOffset) {
            LogOffsetMetadata oldHW = highWatermark;
            
            // 从日志中获取精确的偏移量元数据
            LogOffsetMetadata newHWMetadata = log.convertToOffsetMetadataOrThrow(newHighWatermark);
            highWatermark = newHWMetadata;
            
            debug("分区 {} 的高水位从 {} 更新到 {}", 
                 topicPartition, oldHW.messageOffset, newHighWatermark);
            
            // 通知延迟操作（如DelayedFetch）检查完成条件
            tryCompleteDelayedRequests();
            
            return true;
        }
        
        return false;
    }
    
    /**
     * 检查副本是否同步
     * 基于LEO差距和时间间隔判断副本是否在ISR中
     */
    public boolean isReplicaInSync(int replicaId, 
                                  long leaderLogEndOffset,
                                  long currentTimeMs,
                                  long maxLagMs) {
        Replica replica = assignedReplicas.get(replicaId);
        if (replica == null) {
            return false;
        }
        
        // 检查偏移量同步状态
        long replicaLEO = replica.logEndOffset().messageOffset;
        long lagMessages = leaderLogEndOffset - replicaLEO;
        
        // 检查时间同步状态
        long lagTimeMs = currentTimeMs - replica.lastCaughtUpTimeMs;
        
        // 副本被认为同步，如果：
        // 1. LEO与Leader LEO的差距在允许范围内（通常为0）
        // 2. 最后同步时间在允许的时间窗口内
        boolean isInSync = lagMessages <= config.replicaLagTimeMaxMs && 
                          lagTimeMs <= maxLagMs;
        
        if (!isInSync) {
            debug("副本 {} 不同步：消息滞后={}, 时间滞后={}ms", 
                 replicaId, lagMessages, lagTimeMs);
        }
        
        return isInSync;
    }
    
    /**
     * 从ISR中移除不同步的副本
     * 当副本长时间未能跟上Leader时，将其从ISR中移除
     */
    private void shrinkIsr(Set<Integer> outOfSyncReplicas, long currentTimeMs) {
        leaderIsrUpdateLock.writeLock().lock();
        try {
            if (outOfSyncReplicas.isEmpty()) {
                return;
            }
            
            Set<Integer> newInSyncReplicaIds = new HashSet<>(inSyncReplicaIds);
            newInSyncReplicaIds.removeAll(outOfSyncReplicas);
            
            // 确保Leader始终在ISR中
            newInSyncReplicaIds.add(localBrokerId);
            
            // 更新ISR
            updateIsr(newInSyncReplicaIds, currentTimeMs, "收缩ISR");
            
            info("从分区 {} 的ISR中移除副本 {}，新的ISR: {}", 
                 topicPartition, outOfSyncReplicas, newInSyncReplicaIds);
                 
        } finally {
            leaderIsrUpdateLock.writeLock().unlock();
        }
    }
    
    /**
     * 向ISR中添加重新同步的副本
     * 当副本重新跟上Leader时，将其添加回ISR
     */
    private void expandIsr(Set<Integer> newInSyncReplicas, long currentTimeMs) {
        leaderIsrUpdateLock.writeLock().lock();
        try {
            Set<Integer> newInSyncReplicaIds = new HashSet<>(inSyncReplicaIds);
            newInSyncReplicaIds.addAll(newInSyncReplicas);
            
            // 更新ISR
            updateIsr(newInSyncReplicaIds, currentTimeMs, "扩展ISR");
            
            info("向分区 {} 的ISR中添加副本 {}，新的ISR: {}", 
                 topicPartition, newInSyncReplicas, newInSyncReplicaIds);
                 
        } finally {
            leaderIsrUpdateLock.writeLock().unlock();
        }
    }
}

/**
 * LogCleaner - 日志清理器
 * 实现基于时间和大小的日志清理策略，以及基于key的日志压缩
 */
public class LogCleaner {
    
    private final CleanerConfig config;            // 清理器配置
    private final LogDirFailureChannel logDirFailureChannel; // 日志目录失败通道
    private final Map<TopicPartition, LogToClean> logsToClean; // 待清理日志映射
    
    // 清理线程池
    private final List<CleanerThread> cleaners;    // 清理线程列表
    private final Scheduler scheduler;             // 调度器
    
    /**
     * 启动日志清理器
     * 初始化清理线程池并开始清理任务
     */
    public void startup() {
        info("启动日志清理器，清理线程数: {}", config.numThreads);
        
        // 启动清理线程
        for (int i = 0; i < config.numThreads; i++) {
            CleanerThread cleaner = new CleanerThread(i);
            cleaners.add(cleaner);
            cleaner.start();
        }
        
        // 定期检查需要清理的日志
        scheduler.schedule("log-cleanup-scheduler", () -> {
            checkAndScheduleCleanup();
        }, config.backoffMs, config.backoffMs);
        
        info("日志清理器启动完成");
    }
    
    /**
     * 检查并调度清理任务
     * 识别需要清理的日志并分配给清理线程
     */
    private void checkAndScheduleCleanup() {
        try {
            // 获取所有需要清理的日志
            Map<TopicPartition, UnifiedLog> logsToCheck = logManager.allLogs();
            
            for (Map.Entry<TopicPartition, UnifiedLog> entry : logsToCheck.entrySet()) {
                TopicPartition tp = entry.getKey();
                UnifiedLog log = entry.getValue();
                
                // 检查清理策略
                String cleanupPolicy = log.config().cleanupPolicy();
                
                if ("delete".equals(cleanupPolicy)) {
                    // 基于时间和大小的删除策略
                    scheduleDeleteCleanup(tp, log);
                } else if ("compact".equals(cleanupPolicy)) {
                    // 基于key的压缩策略
                    scheduleCompactCleanup(tp, log);
                } else if ("compact,delete".equals(cleanupPolicy)) {
                    // 先压缩再删除
                    scheduleCompactAndDeleteCleanup(tp, log);
                }
            }
        } catch (Exception e) {
            error("检查清理任务时发生异常", e);
        }
    }
    
    /**
     * 执行日志段删除清理
     * 基于retention.ms和retention.bytes配置删除过期日志段
     */
    private void scheduleDeleteCleanup(TopicPartition tp, UnifiedLog log) {
        long currentTimeMs = time.milliseconds();
        
        // 基于时间的清理
        long retentionMs = log.config().retentionMs();
        if (retentionMs > 0) {
            List<LogSegment> deletableSegments = log.deletableSegments(
                () -> currentTimeMs - retentionMs);
                
            for (LogSegment segment : deletableSegments) {
                if (segment != log.activeSegment()) {
                    info("标记删除过期日志段: {} (最后修改时间: {})", 
                        segment.baseOffset(), 
                        new Date(segment.lastModified()));
                    log.deleteSegment(segment);
                }
            }
        }
        
        // 基于大小的清理
        long retentionBytes = log.config().retentionBytes();
        if (retentionBytes > 0) {
            long currentLogSize = log.size();
            if (currentLogSize > retentionBytes) {
                long bytesToDelete = currentLogSize - retentionBytes;
                List<LogSegment> segmentsToDelete = log.candidateSegmentsForDeletion(bytesToDelete);
                
                for (LogSegment segment : segmentsToDelete) {
                    info("基于大小限制删除日志段: {} (大小: {} 字节)", 
                        segment.baseOffset(), segment.size());
                    log.deleteSegment(segment);
                }
            }
        }
    }
    
    /**
     * 执行日志压缩清理
     * 基于消息key保留每个key的最新值
     */
    private void scheduleCompactCleanup(TopicPartition tp, UnifiedLog log) {
        try {
            // 检查是否需要压缩
            double dirtyRatio = log.dirtyRatio();
            if (dirtyRatio < config.minCleanableRatio) {
                debug("分区 {} 的脏数据比例 {} 低于阈值 {}，跳过压缩", 
                     tp, dirtyRatio, config.minCleanableRatio);
                return;
            }
            
            info("开始压缩分区 {} 的日志，脏数据比例: {}", tp, dirtyRatio);
            
            // 添加到清理队列
            logsToClean.put(tp, new LogToClean(log, LogCleaningState.COMPACTION));
            
        } catch (Exception e) {
            error("调度分区 {} 压缩清理时发生异常", tp, e);
        }
    }
}

/**
 * CleanerThread - 日志清理线程
 * 执行具体的日志清理和压缩操作
 */
private class CleanerThread extends Thread {
    
    private final int threadId;
    private volatile boolean shouldStop = false;
    
    public CleanerThread(int threadId) {
        super("kafka-log-cleaner-thread-" + threadId);
        this.threadId = threadId;
    }
    
    @Override
    public void run() {
        info("日志清理线程 {} 启动", threadId);
        
        try {
            while (!shouldStop) {
                try {
                    // 获取下一个需要清理的日志
                    LogToClean logToClean = grabNextLogToClean();
                    
                    if (logToClean != null) {
                        // 执行清理操作
                        cleanLog(logToClean);
                    } else {
                        // 没有待清理的日志，短暂休眠
                        Thread.sleep(config.backoffMs);
                    }
                } catch (InterruptedException e) {
                    info("清理线程 {} 被中断", threadId);
                    break;
                } catch (Exception e) {
                    error("清理线程 {} 发生异常", threadId, e);
                }
            }
        } finally {
            info("日志清理线程 {} 停止", threadId);
        }
    }
    
    /**
     * 执行具体的日志清理操作
     */
    private void cleanLog(LogToClean logToClean) {
        UnifiedLog log = logToClean.log;
        TopicPartition tp = log.topicPartition();
        
        long startTimeMs = time.milliseconds();
        
        try {
            if (logToClean.cleaningState == LogCleaningState.COMPACTION) {
                // 执行日志压缩
                doCompactLog(log);
            } else {
                // 执行日志删除
                doDeleteLog(log);
            }
            
            long elapsedMs = time.milliseconds() - startTimeMs;
            info("完成分区 {} 的日志清理，耗时 {} ms", tp, elapsedMs);
            
        } catch (Exception e) {
            error("清理分区 {} 的日志时发生异常", tp, e);
            
            // 将日志标记为清理失败
            markLogAsCleaningAborted(tp);
        }
    }
    
    /**
     * 执行日志压缩
     * 保留每个key的最新值，删除旧版本
     */
    private void doCompactLog(UnifiedLog log) throws IOException {
        info("开始压缩日志分区: {}", log.topicPartition());
        
        // 1. 构建key -> 偏移量的映射（保留最新值）
        Map<ByteBuffer, Long> offsetMap = buildOffsetMap(log);
        
        // 2. 创建压缩后的新日志段
        List<LogSegment> segmentsToCompact = log.logSegments().stream()
            .filter(segment -> segment != log.activeSegment())
            .collect(Collectors.toList());
        
        for (LogSegment sourceSegment : segmentsToCompact) {
            LogSegment cleanedSegment = compactSegment(sourceSegment, offsetMap);
            
            // 3. 替换原始段
            log.replaceSegments(
                Collections.singletonList(cleanedSegment),
                Collections.singletonList(sourceSegment)
            );
        }
        
        info("完成日志压缩，分区: {}", log.topicPartition());
    }
    
    /**
     * 构建偏移量映射
     * 扫描日志找出每个key的最新偏移量
     */
    private Map<ByteBuffer, Long> buildOffsetMap(UnifiedLog log) throws IOException {
        Map<ByteBuffer, Long> offsetMap = new HashMap<>();
        
        // 从最新的段开始向前扫描
        List<LogSegment> segments = new ArrayList<>(log.logSegments());
        Collections.reverse(segments);
        
        for (LogSegment segment : segments) {
            // 读取段中的所有记录
            FetchDataInfo fetchInfo = segment.read(segment.baseOffset(), 
                Integer.MAX_VALUE, Optional.empty(), false);
            
            for (RecordBatch batch : fetchInfo.records.batches()) {
                for (Record record : batch) {
                    ByteBuffer key = record.key();
                    if (key != null) {
                        // 由于是从新到旧扫描，只保留第一次遇到的（最新的）偏移量
                        offsetMap.putIfAbsent(key.duplicate(), record.offset());
                    }
                }
            }
        }
        
        info("构建偏移量映射完成，总key数: {}", offsetMap.size());
        return offsetMap;
    }
    
    /**
     * 压缩单个日志段
     * 只保留在偏移量映射中标记为最新的记录
     */
    private LogSegment compactSegment(LogSegment sourceSegment, 
                                     Map<ByteBuffer, Long> offsetMap) throws IOException {
        
        // 创建临时的压缩段
        File tempFile = File.createTempFile("kafka-compacted-", ".log", sourceSegment.log().file().getParentFile());
        LogSegment compactedSegment = LogSegment.open(
            tempFile.getParentFile(),
            sourceSegment.baseOffset(),
            sourceSegment.config(),
            time,
            false  // 不预分配
        );
        
        try {
            // 读取源段的所有数据
            FetchDataInfo fetchInfo = sourceSegment.read(sourceSegment.baseOffset(), 
                Integer.MAX_VALUE, Optional.empty(), false);
            
            MemoryRecordsBuilder cleanedRecordsBuilder = MemoryRecords.builder(
                ByteBuffer.allocate(Math.min(1024 * 1024, fetchInfo.records.sizeInBytes())), // 1MB缓冲区
                RecordBatch.CURRENT_MAGIC_VALUE,
                config.compression(),
                TimestampType.CREATE_TIME,
                sourceSegment.baseOffset()
            );
            
            // 过滤并重新构建记录
            for (RecordBatch batch : fetchInfo.records.batches()) {
                for (Record record : batch) {
                    ByteBuffer key = record.key();
                    
                    // 检查这条记录是否应该保留
                    if (key == null || offsetMap.get(key) == record.offset()) {
                        // 保留这条记录（无key或者是最新版本）
                        cleanedRecordsBuilder.append(
                            record.timestamp(),
                            record.key(),
                            record.value(),
                            record.headers()
                        );
                    }
                }
            }
            
            // 将压缩后的记录写入新段
            MemoryRecords cleanedRecords = cleanedRecordsBuilder.build();
            compactedSegment.append(sourceSegment.baseOffset() + cleanedRecords.records().size() - 1, 
                                   cleanedRecords);
            
            return compactedSegment;
            
        } catch (Exception e) {
            // 清理失败时删除临时文件
            compactedSegment.close();
            tempFile.delete();
            throw e;
        }
    }
}