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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

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

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

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236

/**
 * 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;
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

/**
 * 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封装
 */

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

/**
 * 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();
            }
        }
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190

/**
 * 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();
        }
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327

/**
 * 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;
        }
    }
}