【源码分析】StarRocks EditLog 写入与 Replay 完整流程分析

概述

本文档详细分析 StarRocks FE 中EditLog 的写入（Leader）和 Replay（Follower）的完整流程，以ADD BACKEND操作为例。

完整流程概览

用户操作 (ALTER SYSTEM ADD BACKEND) ↓ Leader FE: SystemInfoService.addBackend() ↓ Leader FE: 更新内存 (idToBackendRef) + 写 EditLog ↓（todo：这里的操作有时间周期吗） EditLog.logAddBackend() → logEdit() → submitLog() → journalQueue ↓ JournalWriter 线程: 从队列取任务 → 序列化 → 写入 BDB JE ↓（todo：这里的操作有时间周期吗） BDB JE: 持久化到磁盘 + 复制到 Follower ↓ Follower FE: Journal Replay 线程读取 BDB JE ↓ EditLog.loadJournal() → 根据 opCode 分发 ↓ SystemInfoService.replayAddBackend() ↓ Follower FE: 更新内存 (idToBackendRef)

第一部分：Leader FE 写入 EditLog

1.1 用户操作触发

场景：用户执行ALTER SYSTEM ADD BACKEND "host:port";或 BE 启动时自动注册

入口：SystemInfoService.addBackends()→addBackend(host, heartbeatPort)

1.2 addBackend() 方法执行

位置：SystemInfoService.java:203-224

privatevoidaddBackend(Stringhost,intheartbeatPort){// 1. 创建新的 Backend 对象BackendnewBackend=newBackend(GlobalStateMgr.getCurrentState().getNextId(),host,heartbeatPort);// 2. 更新 Leader 的内存状态（立即生效）Map<Long,Backend>copiedBackends=Maps.newHashMap(idToBackendRef);copiedBackends.put(newBackend.getId(),newBackend);idToBackendRef=ImmutableMap.copyOf(copiedBackends);// ✅ Leader 立即能看到 BE// 3. 更新 reportVersionMap<Long,AtomicLong>copiedReportVersions=Maps.newHashMap(idToReportVersionRef);copiedReportVersions.put(newBackend.getId(),newAtomicLong(0L));idToReportVersionRef=ImmutableMap.copyOf(copiedReportVersions);// 4. 添加到集群setBackendOwner(newBackend);// 5. ⭐ 关键：记录到 EditLog（用于同步到 Follower）GlobalStateMgr.getCurrentState().getEditLog().logAddBackend(newBackend);LOG.info("finished to add {} ",newBackend);}

关键点：

• 步骤 2：Leader 的idToBackendRef立即更新（Leader 能立即看到 BE）
• 步骤 5：调用logAddBackend()记录到 EditLog（用于同步到 Follower）

1.3 EditLog.logAddBackend() 方法

位置：EditLog.java:1179-1181

publicvoidlogAddBackend(Backendbe){logEdit(OperationType.OP_ADD_BACKEND,be);// OP_ADD_BACKEND = 50}

作用：将操作类型和 Backend 对象传递给logEdit()方法

1.4 EditLog.logEdit() 方法

位置：EditLog.java:976-980

protectedvoidlogEdit(shortop,Writablewritable){longstart=System.nanoTime();Future<Boolean>task=submitLog(op,writable,-1);// 提交到队列waitInfinity(start,task);// 等待写入完成}

作用：

• 调用submitLog()提交日志任务
• 调用waitInfinity()等待写入完成（阻塞直到成功）

1.5 EditLog.submitLog() 方法

位置：EditLog.java:985-1023

注意：只有 Leader 才能写 EditLog

privateFuture<Boolean>submitLog(shortop,Writablewritable,longmaxWaitIntervalMs){// ⚠️ 检查：只有 Leader 才能写 EditLogPreconditions.checkState(GlobalStateMgr.getCurrentState().isLeader(),"Current node is not leader, submit log is not allowed");DataOutputBufferbuffer=newDataOutputBuffer(OUTPUT_BUFFER_INIT_SIZE);// 1. 序列化操作try{JournalEntityentity=newJournalEntity();entity.setOpCode(op);// OP_ADD_BACKEND = 50entity.setData(writable);// Backend 对象entity.write(buffer);// 序列化到 buffer}catch(IOExceptione){LOG.info("failed to serialized: {}",e);}// 2. 创建 JournalTaskJournalTasktask=newJournalTask(buffer,maxWaitIntervalMs);// 3. 放入队列（阻塞直到有空间）intcnt=0;while(true){try{if(cnt!=0){Thread.sleep(1000);}this.journalQueue.put(task);// ⭐ 放入队列，等待 JournalWriter 处理break;}catch(InterruptedExceptione){LOG.warn("failed to put queue, wait and retry {} times..: {}",cnt,e);}cnt++;}returntask;}

关键点：

1. 只有 Leader 才能写 EditLog（Follower 会抛出异常）
2. 序列化：将OperationType.OP_ADD_BACKEND和Backend对象序列化到 buffer
3. 放入队列：journalQueue是一个BlockingQueue<JournalTask>，由JournalWriter 线程消费

1.6 JournalWriter 线程处理

位置：JournalWriter是一个后台线程，从journalQueue取任务并写入 BDB JE

流程：

1. 从队列取任务：JournalTask task = journalQueue.take()
2. 写入 BDB JE：将序列化后的数据写入 BDB JE（Berkeley DB Java Edition）
3. 持久化：BDB JE 将数据持久化到磁盘（meta/bdb/目录）
4. 复制到 Follower：BDB JE 的复制机制自动将数据复制到 Follower FE

BDB JE 存储位置：

• Leader:meta/bdb/目录（本地持久化）
• Follower: 通过 BDB JE 复制机制自动同步

第二部分：Follower FE Replay EditLog

2.1 Journal Replay 线程启动

位置：GlobalStateMgr.java:1795-1884（Replayer 线程）

启动时机：FE 启动时，在GlobalStateMgr.initialize()中启动

线程逻辑：

Daemonreplayer=newDaemon("replayer",2000L){@OverrideprotectedvoidrunOneCycle(){booleanhasLog=false;try{if(cursor==null){// 1. 从上次 replay 的位置开始LOG.info("start to replay from {}",replayedJournalId.get());cursor=journal.read(replayedJournalId.get()+1,JournalCursor.CUROSR_END_KEY);}else{cursor.refresh();// 刷新 cursor，读取新的 journal}// 2. Replay journal（带流控）hasLog=replayJournalInner(cursor,true);metaReplayState.setOk();}catch(Throwablee){LOG.error("replayer thread catch an exception when replay journal {}.",replayedJournalId.get()+1,e);// 处理异常...}}};

关键点：

• 周期性执行：每 2 秒执行一次（2000L）
• 从上次位置继续：replayedJournalId.get() + 1表示从上次 replay 的下一个 journal 开始
• 读取新 journal：cursor.refresh()刷新 cursor，读取 BDB JE 中的新 journal

2.2 replayJournalInner() 方法

位置：GlobalStateMgr.java:1937-1999

protectedbooleanreplayJournalInner(JournalCursorcursor,booleanflowControl)throwsJournalException,InterruptedException,JournalInconsistentException{longstartReplayId=replayedJournalId.get();longstartTime=System.currentTimeMillis();longlineCnt=0;while(true){JournalEntityentity=null;try{// 1. 从 cursor 读取下一个 journal entityentity=cursor.next();// EOF 或没有更多 journalif(entity==null){break;}// 2. ⭐ 关键：调用 EditLog.loadJournal() 处理 journalEditLog.loadJournal(this,entity);}catch(Throwablee){// 处理异常...throwe;}// 3. 更新 replayedJournalIdreplayedJournalId.incrementAndGet();LOG.debug("journal {} replayed.",replayedJournalId);// 4. 流控：避免一次 replay 太多 journalif(flowControl){longcost=System.currentTimeMillis()-startTime;if(cost>REPLAYER_MAX_MS_PER_LOOP){LOG.warn("replay journal cost too much time: {} replayedJournalId: {}",cost,replayedJournalId);break;// 超时，下次继续}lineCnt+=1;if(lineCnt>REPLAYER_MAX_LOGS_PER_LOOP){LOG.warn("replay too many journals: lineCnt {}, replayedJournalId: {}",lineCnt,replayedJournalId);break;// 数量限制，下次继续}}}if(replayedJournalId.get()-startReplayId>0){LOG.info("replayed journal from {} - {}",startReplayId,replayedJournalId);returntrue;}returnfalse;}

关键点：

1. 从 cursor 读取：cursor.next()从 BDB JE 读取下一个 journal entity
2. 调用 loadJournal：EditLog.loadJournal(this, entity)处理 journal
3. 流控机制：避免一次 replay 太多 journal，防止阻塞其他操作
4. 更新 replayedJournalId：记录已 replay 的 journal ID

2.3 EditLog.loadJournal() 方法

位置：EditLog.java:114-1689

publicstaticvoidloadJournal(GlobalStateMgrglobalStateMgr,JournalEntityjournal)throwsJournalInconsistentException{shortopCode=journal.getOpCode();// 获取操作类型if(opCode!=OperationType.OP_SAVE_NEXTID&&opCode!=OperationType.OP_TIMESTAMP){LOG.debug("replay journal op code: {}",opCode);}try{switch(opCode){// ... 其他操作类型 ...caseOperationType.OP_ADD_BACKEND:{// opCode = 50// 1. 反序列化 Backend 对象Backendbe=(Backend)journal.getData();// 2. ⭐ 关键：调用 SystemInfoService.replayAddBackend()GlobalStateMgr.getCurrentSystemInfo().replayAddBackend(be);break;}caseOperationType.OP_DROP_BACKEND:{Backendbe=(Backend)journal.getData();GlobalStateMgr.getCurrentSystemInfo().replayDropBackend(be);break;}// ... 其他操作类型 ...}}catch(Throwablee){// 处理异常...}}

关键点：

1. 根据 opCode 分发：switch (opCode)根据操作类型分发到对应的 replay 方法
2. 反序列化：journal.getData()反序列化出Backend对象
3. 调用 replay 方法：replayAddBackend(be)更新 Follower 的内存状态

2.4 SystemInfoService.replayAddBackend() 方法

位置：SystemInfoService.java:909-934

publicvoidreplayAddBackend(BackendnewBackend){// 1. 兼容性处理（旧版本）if(GlobalStateMgr.getCurrentStateJournalVersion()<FeMetaVersion.VERSION_30){newBackend.setBackendState(BackendState.using);}// 2. ⭐ 关键：更新 Follower 的 idToBackendRef（Copy-on-Write）Map<Long,Backend>copiedBackends=Maps.newHashMap(idToBackendRef);copiedBackends.put(newBackend.getId(),newBackend);idToBackendRef=ImmutableMap.copyOf(copiedBackends);// ✅ Follower 现在能看到 BE// 3. 更新 reportVersionMap<Long,AtomicLong>copiedReportVerions=Maps.newHashMap(idToReportVersionRef);copiedReportVerions.put(newBackend.getId(),newAtomicLong(0L));idToReportVersionRef=ImmutableMap.copyOf(copiedReportVerions);// 4. 添加到集群if(newBackend.getBackendState()==BackendState.using){finalClustercluster=GlobalStateMgr.getCurrentState().getCluster();if(null!=cluster){// replay logcluster.addBackend(newBackend.getId());}else{// 这种情况发生在加载 image 时，cluster 还没创建// BE 会在 loadCluster 时更新}}}

关键点：

1. Copy-on-Write 更新：使用 Copy-on-Write 模式更新idToBackendRef
2. Follower 现在能看到 BE：idToBackendRef更新后，Follower 的getBackends()方法能返回这个 BE
3. 添加到集群：将 BE 添加到DEFAULT_CLUSTER

第三部分：为什么 Follower 可能看不到 BE？

3.1 时序问题

场景：初始化集群时，先启动 FE，然后启动 BE

时间线： T1: 启动 3 个 FE - FE1 成为 Leader - FE2, FE3 成为 Follower（可能还在启动/同步中） T2: 启动 3 个 BE - BE1 连接到 Leader FE1，注册成功 - Leader FE1: * addBackend() → idToBackendRef 立即更新 ✅ * logAddBackend() → 写入 EditLog → BDB JE - BE2, BE3 同样注册到 Leader T3: Follower FE2, FE3 的状态 - 如果 Follower 的 journal replay 还没追上 - replayedJournalId < Leader 的最新 journal ID - Follower 还没 replay 到 BE 注册的 EditLog - 因此 idToBackendRef 还是空的（或只有部分 BE） - 导致 getClusterAvailableCapacityB() 返回 0 - 触发 "Cluster has no available capacity" 错误 ❌

3.2 代码验证

Follower 的容量检查：

// SystemInfoService.getClusterAvailableCapacityB()publiclonggetClusterAvailableCapacityB(){List<Backend>clusterBackends=getBackends();// 从 idToBackendRef 获取longcapacity=0L;for(Backendbackend:clusterBackends){if(backend.isDecommissioned()){capacity-=backend.getDataUsedCapacityB();}else{capacity+=backend.getAvailableCapacityB();}}returncapacity;// 如果 idToBackendRef 为空，返回 0}

如果 Follower 的idToBackendRef还是空的：

• getBackends()返回空列表
• capacity = 0
• checkClusterCapacity()抛出异常：Cluster has no available capacity

第四部分：完整数据流图

┌─────────────────────────────────────────────────────────────────┐ │ Leader FE │ ├─────────────────────────────────────────────────────────────────┤ │ │ │ 1. addBackend(host, port) │ │ ├─> 创建 Backend 对象 │ │ ├─> 更新 idToBackendRef (立即生效) ✅ │ │ └─> logAddBackend(be) │ │ └─> logEdit(OP_ADD_BACKEND, be) │ │ └─> submitLog() │ │ ├─> 序列化 JournalEntity │ │ └─> journalQueue.put(task) │ │ │ │ 2. JournalWriter 线程 │ │ ├─> 从 journalQueue 取任务 │ │ ├─> 写入 BDB JE │ │ └─> 持久化到 meta/bdb/ │ │ │ └─────────────────────────────────────────────────────────────────┘ │ │ BDB JE 复制机制 ▼ ┌─────────────────────────────────────────────────────────────────┐ │ Follower FE │ ├─────────────────────────────────────────────────────────────────┤ │ │ │ 3. Replayer 线程（每 2 秒执行一次） │ │ ├─> cursor = journal.read(replayedJournalId + 1, END) │ │ └─> replayJournalInner(cursor) │ │ └─> while (entity = cursor.next()) │ │ └─> EditLog.loadJournal(this, entity) │ │ └─> switch (opCode) │ │ case OP_ADD_BACKEND: │ │ └─> replayAddBackend(be) │ │ └─> 更新 idToBackendRef ✅ │ │ │ │ 4. 容量检查 │ │ ├─> getClusterAvailableCapacityB() │ │ │ └─> getBackends() → 从 idToBackendRef 获取 │ │ └─> 如果 idToBackendRef 为空 → capacity = 0 → 报错 ❌ │ │ │ └─────────────────────────────────────────────────────────────────┘

第五部分：关键代码位置总结

Leader 写入流程

1. 用户操作入口：SystemInfoService.addBackend()(203-224)
2. 写 EditLog：EditLog.logAddBackend()(1179-1181)
3. 提交到队列：EditLog.logEdit()(976-980) →submitLog()(985-1023)
4. JournalWriter 写入：后台线程从journalQueue取任务，写入 BDB JE

Follower Replay 流程

1. Replay 线程：GlobalStateMgr.Replayer(1795-1884)
2. 读取 Journal：replayJournalInner()(1937-1999)
3. 分发处理：EditLog.loadJournal()(114-1689)
4. 更新内存：SystemInfoService.replayAddBackend()(909-934)

第六部分：诊断和验证

6.1 检查 Leader 的 EditLog 写入

# 在 Leader FE 上查看日志tail-100fe/log/fe.log|grep-i"add.*backend\|logAddBackend"

期望看到：

finished to add Backend [id=xxx, host=xxx, heartbeatPort=xxx]

6.2 检查 Follower 的 Replay 状态

-- 在 Leader 上查看SHOWFRONTENDS;-- 关注：-- - Follower 的 ReplayedJournalId-- - 是否接近 Leader 的 ReplayedJournalId

6.3 检查 Follower 的 BE 视图

-- 在 Follower 上查看（如果能连接）SHOWBACKENDS;-- 如果看不到 BE，说明还没 replay 到 BE 注册的 EditLog

6.4 检查 Replay 日志

# 在 Follower FE 上查看日志tail-100fe/log/fe.log|grep-i"replay\|replayed journal"

期望看到：

replayed journal from xxx - yyy

总结

1. Leader 写入：addBackend()→ 更新内存 →logAddBackend()→ 序列化 → 队列 → BDB JE
2. Follower Replay：Replayer 线程 → 读取 BDB JE →loadJournal()→replayAddBackend()→ 更新内存
3. 时序问题：如果 Follower 的 journal replay 还没追上，就看不到新注册的 BE
4. 解决方案：等待 Follower 同步完成，或重启 Follower 使其从 Leader 重新同步