★ FEATURED ARTICLE 文章目录概述完整流程概览第一部分Leader FE 写入 EditLog1.1 用户操作触发1.2 addBackend() 方法执行1.3 EditLog.logAddBackend() 方法1.4 EditLog.logEdit() 方法1.5 EditLog.submitLog() 方法1.6 JournalWriter 线程处理第二部分Follower FE Replay EditLog2.1 Journal Replay 线程启动2.2 replayJournalInner() 方法2.3 EditLog.loadJournal() 方法2.4 SystemInfoService.replayAddBackend() 方法第三部分为什么 Follower 可能看不到 BE3.1 时序问题3.2 代码验证第四部分完整数据流图第五部分关键代码位置总结Leader 写入流程Follower Replay 流程第六部分诊断和验证6.1 检查 Leader 的 EditLog 写入6.2 检查 Follower 的 Replay 状态6.3 检查 Follower 的 BE 视图6.4 检查 Replay 日志总结概述本文档详细分析 StarRocks FE 中EditLog 的写入Leader和 ReplayFollower的完整流程以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 写入 EditLog1.1 用户操作触发场景用户执行ALTER SYSTEM ADD BACKEND host:port;或 BE 启动时自动注册入口SystemInfoService.addBackends()→addBackend(host, heartbeatPort)1.2 addBackend() 方法执行位置SystemInfoService.java:203-224privatevoidaddBackend(Stringhost,intheartbeatPort){// 1. 创建新的 Backend 对象BackendnewBackendnewBackend(GlobalStateMgr.getCurrentState().getNextId(),host,heartbeatPort);// 2. 更新 Leader 的内存状态立即生效MapLong,BackendcopiedBackendsMaps.newHashMap(idToBackendRef);copiedBackends.put(newBackend.getId(),newBackend);idToBackendRefImmutableMap.copyOf(copiedBackends);// ✅ Leader 立即能看到 BE// 3. 更新 reportVersionMapLong,AtomicLongcopiedReportVersionsMaps.newHashMap(idToReportVersionRef);copiedReportVersions.put(newBackend.getId(),newAtomicLong(0L));idToReportVersionRefImmutableMap.copyOf(copiedReportVersions);// 4. 添加到集群setBackendOwner(newBackend);// 5. ⭐ 关键记录到 EditLog用于同步到 FollowerGlobalStateMgr.getCurrentState().getEditLog().logAddBackend(newBackend);LOG.info(finished to add {} ,newBackend);}关键点步骤 2Leader 的idToBackendRef立即更新Leader 能立即看到 BE步骤 5调用logAddBackend()记录到 EditLog用于同步到 Follower1.3 EditLog.logAddBackend() 方法位置EditLog.java:1179-1181publicvoidlogAddBackend(Backendbe){logEdit(OperationType.OP_ADD_BACKEND,be);// OP_ADD_BACKEND 50}作用将操作类型和 Backend 对象传递给logEdit()方法1.4 EditLog.logEdit() 方法位置EditLog.java:976-980protectedvoidlogEdit(shortop,Writablewritable){longstartSystem.nanoTime();FutureBooleantasksubmitLog(op,writable,-1);// 提交到队列waitInfinity(start,task);// 等待写入完成}作用调用submitLog()提交日志任务调用waitInfinity()等待写入完成阻塞直到成功1.5 EditLog.submitLog() 方法位置EditLog.java:985-1023注意只有 Leader 才能写 EditLogprivateFutureBooleansubmitLog(shortop,Writablewritable,longmaxWaitIntervalMs){// ⚠️ 检查只有 Leader 才能写 EditLogPreconditions.checkState(GlobalStateMgr.getCurrentState().isLeader(),Current node is not leader, submit log is not allowed);DataOutputBufferbuffernewDataOutputBuffer(OUTPUT_BUFFER_INIT_SIZE);// 1. 序列化操作try{JournalEntityentitynewJournalEntity();entity.setOpCode(op);// OP_ADD_BACKEND 50entity.setData(writable);// Backend 对象entity.write(buffer);// 序列化到 buffer}catch(IOExceptione){LOG.info(failed to serialized: {},e);}// 2. 创建 JournalTaskJournalTasktasknewJournalTask(buffer,maxWaitIntervalMs);// 3. 放入队列阻塞直到有空间intcnt0;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;}关键点只有 Leader 才能写 EditLogFollower 会抛出异常序列化将OperationType.OP_ADD_BACKEND和Backend对象序列化到 buffer放入队列journalQueue是一个BlockingQueueJournalTask由JournalWriter 线程消费1.6 JournalWriter 线程处理位置JournalWriter是一个后台线程从journalQueue取任务并写入 BDB JE流程从队列取任务JournalTask task journalQueue.take()写入 BDB JE将序列化后的数据写入 BDB JEBerkeley DB Java Edition持久化BDB JE 将数据持久化到磁盘meta/bdb/目录复制到 FollowerBDB JE 的复制机制自动将数据复制到 Follower FEBDB JE 存储位置Leader:meta/bdb/目录本地持久化Follower: 通过 BDB JE 复制机制自动同步第二部分Follower FE Replay EditLog2.1 Journal Replay 线程启动位置GlobalStateMgr.java:1795-1884Replayer 线程启动时机FE 启动时在GlobalStateMgr.initialize()中启动线程逻辑DaemonreplayernewDaemon(replayer,2000L){OverrideprotectedvoidrunOneCycle(){booleanhasLogfalse;try{if(cursornull){// 1. 从上次 replay 的位置开始LOG.info(start to replay from {},replayedJournalId.get());cursorjournal.read(replayedJournalId.get()1,JournalCursor.CUROSR_END_KEY);}else{cursor.refresh();// 刷新 cursor读取新的 journal}// 2. Replay journal带流控hasLogreplayJournalInner(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 开始读取新 journalcursor.refresh()刷新 cursor读取 BDB JE 中的新 journal2.2 replayJournalInner() 方法位置GlobalStateMgr.java:1937-1999protectedbooleanreplayJournalInner(JournalCursorcursor,booleanflowControl)throwsJournalException,InterruptedException,JournalInconsistentException{longstartReplayIdreplayedJournalId.get();longstartTimeSystem.currentTimeMillis();longlineCnt0;while(true){JournalEntityentitynull;try{// 1. 从 cursor 读取下一个 journal entityentitycursor.next();// EOF 或没有更多 journalif(entitynull){break;}// 2. ⭐ 关键调用 EditLog.loadJournal() 处理 journalEditLog.loadJournal(this,entity);}catch(Throwablee){// 处理异常...throwe;}// 3. 更新 replayedJournalIdreplayedJournalId.incrementAndGet();LOG.debug(journal {} replayed.,replayedJournalId);// 4. 流控避免一次 replay 太多 journalif(flowControl){longcostSystem.currentTimeMillis()-startTime;if(costREPLAYER_MAX_MS_PER_LOOP){LOG.warn(replay journal cost too much time: {} replayedJournalId: {},cost,replayedJournalId);break;// 超时下次继续}lineCnt1;if(lineCntREPLAYER_MAX_LOGS_PER_LOOP){LOG.warn(replay too many journals: lineCnt {}, replayedJournalId: {},lineCnt,replayedJournalId);break;// 数量限制下次继续}}}if(replayedJournalId.get()-startReplayId0){LOG.info(replayed journal from {} - {},startReplayId,replayedJournalId);returntrue;}returnfalse;}关键点从 cursor 读取cursor.next()从 BDB JE 读取下一个 journal entity调用 loadJournalEditLog.loadJournal(this, entity)处理 journal流控机制避免一次 replay 太多 journal防止阻塞其他操作更新 replayedJournalId记录已 replay 的 journal ID2.3 EditLog.loadJournal() 方法位置EditLog.java:114-1689publicstaticvoidloadJournal(GlobalStateMgrglobalStateMgr,JournalEntityjournal)throwsJournalInconsistentException{shortopCodejournal.getOpCode();// 获取操作类型if(opCode!OperationType.OP_SAVE_NEXTIDopCode!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){// 处理异常...}}关键点根据 opCode 分发switch (opCode)根据操作类型分发到对应的 replay 方法反序列化journal.getData()反序列化出Backend对象调用 replay 方法replayAddBackend(be)更新 Follower 的内存状态2.4 SystemInfoService.replayAddBackend() 方法位置SystemInfoService.java:909-934publicvoidreplayAddBackend(BackendnewBackend){// 1. 兼容性处理旧版本if(GlobalStateMgr.getCurrentStateJournalVersion()FeMetaVersion.VERSION_30){newBackend.setBackendState(BackendState.using);}// 2. ⭐ 关键更新 Follower 的 idToBackendRefCopy-on-WriteMapLong,BackendcopiedBackendsMaps.newHashMap(idToBackendRef);copiedBackends.put(newBackend.getId(),newBackend);idToBackendRefImmutableMap.copyOf(copiedBackends);// ✅ Follower 现在能看到 BE// 3. 更新 reportVersionMapLong,AtomicLongcopiedReportVerionsMaps.newHashMap(idToReportVersionRef);copiedReportVerions.put(newBackend.getId(),newAtomicLong(0L));idToReportVersionRefImmutableMap.copyOf(copiedReportVerions);// 4. 添加到集群if(newBackend.getBackendState()BackendState.using){finalClusterclusterGlobalStateMgr.getCurrentState().getCluster();if(null!cluster){// replay logcluster.addBackend(newBackend.getId());}else{// 这种情况发生在加载 image 时cluster 还没创建// BE 会在 loadCluster 时更新}}}关键点Copy-on-Write 更新使用 Copy-on-Write 模式更新idToBackendRefFollower 现在能看到 BEidToBackendRef更新后Follower 的getBackends()方法能返回这个 BE添加到集群将 BE 添加到DEFAULT_CLUSTER第三部分为什么 Follower 可能看不到 BE3.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(){ListBackendclusterBackendsgetBackends();// 从 idToBackendRef 获取longcapacity0L;for(Backendbackend:clusterBackends){if(backend.isDecommissioned()){capacity-backend.getDataUsedCapacityB();}else{capacitybackend.getAvailableCapacityB();}}returncapacity;// 如果 idToBackendRef 为空返回 0}如果 Follower 的idToBackendRef还是空的getBackends()返回空列表capacity 0checkClusterCapacity()抛出异常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 写入流程用户操作入口SystemInfoService.addBackend()(203-224)写 EditLogEditLog.logAddBackend()(1179-1181)提交到队列EditLog.logEdit()(976-980) →submitLog()(985-1023)JournalWriter 写入后台线程从journalQueue取任务写入 BDB JEFollower Replay 流程Replay 线程GlobalStateMgr.Replayer(1795-1884)读取 JournalreplayJournalInner()(1937-1999)分发处理EditLog.loadJournal()(114-1689)更新内存SystemInfoService.replayAddBackend()(909-934)第六部分诊断和验证6.1 检查 Leader 的 EditLog 写入# 在 Leader FE 上查看日志tail-100fe/log/fe.log|grep-iadd.*backend\|logAddBackend期望看到finished to add Backend [idxxx, hostxxx, heartbeatPortxxx]6.2 检查 Follower 的 Replay 状态-- 在 Leader 上查看SHOWFRONTENDS;-- 关注-- - Follower 的 ReplayedJournalId-- - 是否接近 Leader 的 ReplayedJournalId6.3 检查 Follower 的 BE 视图-- 在 Follower 上查看如果能连接SHOWBACKENDS;-- 如果看不到 BE说明还没 replay 到 BE 注册的 EditLog6.4 检查 Replay 日志# 在 Follower FE 上查看日志tail-100fe/log/fe.log|grep-ireplay\|replayed journal期望看到replayed journal from xxx - yyy总结Leader 写入addBackend()→ 更新内存 →logAddBackend()→ 序列化 → 队列 → BDB JEFollower ReplayReplayer 线程 → 读取 BDB JE →loadJournal()→replayAddBackend()→ 更新内存时序问题如果 Follower 的 journal replay 还没追上就看不到新注册的 BE解决方案等待 Follower 同步完成或重启 Follower 使其从 Leader 重新同步
阅读完成 · 觉得有帮助？
