flowchart TB
    subgraph Controller["Controller (KRaft 模式)"]
        CTRL[QuorumController]
        RAFT[Raft Log]
        CTRL -->|写入元数据变更| RAFT
    end
    
    subgraph MetadataModule["Metadata 模块"]
        LOADER[MetadataLoader]
        DELTA[MetadataDelta]
        IMAGE[MetadataImage]
        CACHE[KRaftMetadataCache]
        
        RAFT -->|同步日志| LOADER
        LOADER -->|构建增量| DELTA
        DELTA -->|应用到| IMAGE
        IMAGE -->|更新| CACHE
    end
    
    subgraph Publishers["发布者"]
        CACHEPUB[MetadataCachePublisher]
        BROKERPUB[BrokerMetadataPublisher]
        FEATUREPUB[FeaturesPublisher]
        
        IMAGE -->|通知| CACHEPUB
        IMAGE -->|通知| BROKERPUB
        IMAGE -->|通知| FEATUREPUB
    end
    
    subgraph Consumers["元数据消费者"]
        APIS[KafkaApis]
        REPLICA[ReplicaManager]
        GC[GroupCoordinator]
        TC[TransactionCoordinator]
        
        CACHE -->|查询元数据| APIS
        CACHE -->|查询元数据| REPLICA
        CACHE -->|查询元数据| GC
        CACHE -->|查询元数据| TC
    end
    
    CLIENT[客户端] -->|MetadataRequest| APIS
    APIS -->|MetadataResponse| CLIENT

public record MetadataImage(
    MetadataProvenance provenance,      // 元数据来源（offset, epoch, lastContainedLogTimeMs）
    FeaturesImage features,             // 特性版本（metadata.version, KRaft version）
    ClusterImage cluster,               // 集群信息（Brokers, Controllers）
    TopicsImage topics,                 // Topic/Partition 元数据
    ConfigurationsImage configs,        // 配置信息（Topic/Broker/Client 配置）
    ClientQuotasImage clientQuotas,     // 客户端配额
    ProducerIdsImage producerIds,       // Producer ID 分配
    AclsImage acls,                     // ACL 权限
    ScramImage scram,                   // SCRAM 认证
    DelegationTokenImage delegationTokens  // 委派令牌
) {
    public static final MetadataImage EMPTY = new MetadataImage(...);
    
    public OffsetAndEpoch highestOffsetAndEpoch() {
        return new OffsetAndEpoch(
            provenance.lastContainedOffset(),
            provenance.lastContainedEpoch()
        );
    }
}

public record ClusterImage(
    Map<Integer, BrokerRegistration> brokers,        // Broker 注册信息
    Map<Integer, ControllerRegistration> controllers  // Controller 注册信息
) {
    public BrokerRegistration broker(int nodeId) {
        return brokers.get(nodeId);
    }
    
    public long brokerEpoch(int brokerId) {
        BrokerRegistration brokerRegistration = broker(brokerId);
        return brokerRegistration == null ? -1L : brokerRegistration.epoch();
    }
}

public class TopicsImage {
    private final Map<Uuid, TopicImage> topicsById;
    private final Map<String, TopicImage> topicsByName;
    
    public TopicImage getTopic(Uuid id) { /* ... */ }
    public TopicImage getTopic(String name) { /* ... */ }
}

public class TopicImage {
    private final String name;
    private final Uuid id;
    private final Map<Integer, PartitionRegistration> partitions;
    
    public Map<Integer, PartitionRegistration> partitions() {
        return partitions;
    }
}

public record FeaturesImage(
    Map<String, Short> supportedFeatures,   // 支持的特性及版本
    MetadataVersion metadataVersion,        // 当前 metadata.version
    KRaftVersion kraftVersion               // 当前 KRaft version
) {}

public class ConfigurationsImage {
    private final Map<ConfigResource, ConfigurationImage> data;
    
    public Map<String, String> configMapForResource(ConfigResource resource) {
        ConfigurationImage image = data.get(resource);
        return image == null ? Collections.emptyMap() : image.data();
    }
}

public interface MetadataCache extends ConfigRepository {
    
    // Topic 元数据查询
    List<MetadataResponseData.MetadataResponseTopic> getTopicMetadata(
        Set<String> topics,
        ListenerName listenerName,
        boolean errorUnavailableEndpoints,
        boolean errorUnavailableListeners
    );
    
    Set<String> getAllTopics();
    
    Uuid getTopicId(String topicName);
    
    Optional<String> getTopicName(Uuid topicId);
    
    Optional<Integer> numPartitions(String topic);
    
    // Partition 元数据查询
    Optional<LeaderAndIsr> getLeaderAndIsr(String topic, int partitionId);
    
    Optional<Node> getPartitionLeaderEndpoint(String topic, int partitionId, ListenerName listenerName);
    
    Map<Integer, Node> getPartitionReplicaEndpoints(TopicPartition tp, ListenerName listenerName);
    
    // Broker 元数据查询
    boolean hasAliveBroker(int brokerId);
    
    Optional<Long> getAliveBrokerEpoch(int brokerId);
    
    boolean isBrokerFenced(int brokerId);
    
    boolean isBrokerShuttingDown(int brokerId);
    
    Optional<Node> getAliveBrokerNode(int brokerId, ListenerName listenerName);
    
    List<Node> getAliveBrokerNodes(ListenerName listenerName);
    
    List<Node> getBrokerNodes(ListenerName listenerName);
    
    // 映射查询
    Map<Uuid, String> topicIdsToNames();
    
    Map<String, Uuid> topicNamesToIds();
    
    // 特性查询
    MetadataVersion metadataVersion();
    
    FinalizedFeatures features();
    
    // 配置查询
    DescribeClientQuotasResponseData describeClientQuotas(DescribeClientQuotasRequestData request);
    
    DescribeUserScramCredentialsResponseData describeScramCredentials(DescribeUserScramCredentialsRequestData request);
}

public class LeaderAndIsr {
    private final int leader;             // Leader Broker ID
    private final int leaderEpoch;        // Leader Epoch
    private final List<Integer> isr;      // ISR 列表
    private final int partitionEpoch;     // Partition Epoch
    
    // 用于 Replica Fetcher 判断是否需要更新
    public boolean hasSameIsr(LeaderAndIsr other) {
        return new HashSet<>(isr).equals(new HashSet<>(other.isr));
    }
}

class KRaftMetadataCache(
    val brokerId: Int,
    val kraftVersionSupplier: Supplier[KRaftVersion]
) extends MetadataCache with Logging {
    
    // 核心：当前元数据快照（volatile 保证可见性）
    @volatile private var _currentImage: MetadataImage = MetadataImage.EMPTY
    
    def currentImage(): MetadataImage = _currentImage
    
    // 更新元数据快照（由 MetadataCachePublisher 调用）
    def setImage(newImage: MetadataImage): Unit = {
        _currentImage = newImage
    }
    
    // ... 查询方法实现 ...
}

override def getTopicMetadata(
    topics: Set[String],
    listenerName: ListenerName,
    errorUnavailableEndpoints: Boolean,
    errorUnavailableListeners: Boolean
): List[MetadataResponseTopic] = {
    val image = _currentImage  // 获取快照引用
    
    topics.map { topicName =>
        val topicImage = Option(image.topics().getTopic(topicName))
        
        topicImage match {
            case None =>
                // Topic 不存在
                new MetadataResponseTopic()
                    .setName(topicName)
                    .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code)
            
            case Some(topic) =>
                // 构建 Partition 元数据列表
                val partitions = topic.partitions().entrySet().asScala.map { entry =>
                    val partitionId = entry.getKey
                    val partition = entry.getValue
                    
                    // 查找 Leader 节点
                    val maybeLeader = getAliveEndpoint(image, partition.leader, listenerName)
                    
                    maybeLeader match {
                        case None =>
                            // Leader 不可用
                            new MetadataResponsePartition()
                                .setErrorCode(Errors.LEADER_NOT_AVAILABLE.code)
                                .setPartitionIndex(partitionId)
                                .setLeaderId(MetadataResponse.NO_LEADER_ID)
                                .setLeaderEpoch(partition.leaderEpoch)
                                .setReplicaNodes(Replicas.toList(partition.replicas))
                                .setIsrNodes(Replicas.toList(partition.isr))
                        
                        case Some(leader) =>
                            // Leader 可用
                            new MetadataResponsePartition()
                                .setErrorCode(Errors.NONE.code)
                                .setPartitionIndex(partitionId)
                                .setLeaderId(leader.id())
                                .setLeaderEpoch(partition.leaderEpoch)
                                .setReplicaNodes(Replicas.toList(partition.replicas))
                                .setIsrNodes(Replicas.toList(partition.isr))
                    }
                }.toList
                
                new MetadataResponseTopic()
                    .setName(topicName)
                    .setTopicId(topic.id())
                    .setErrorCode(Errors.NONE.code)
                    .setPartitions(partitions.asJava)
        }
    }.toList.asJava
}

override def getLeaderAndIsr(topic: String, partitionId: Int): Optional[LeaderAndIsr] = {
    val image = _currentImage
    
    Option(image.topics().getTopic(topic)).flatMap { topicImage =>
        Option(topicImage.partitions().get(partitionId)).map { partition =>
            new LeaderAndIsr(
                partition.leader,
                partition.leaderEpoch,
                Replicas.toList(partition.isr).asScala.map(_.toInt).toList.asJava,
                partition.partitionEpoch
            )
        }
    }.toJava
}

override def getAliveBrokerNode(
    brokerId: Int,
    listenerName: ListenerName
): Optional[Node] = {
    val image = _currentImage
    
    Option(image.cluster().broker(brokerId)).flatMap { broker =>
        if (broker.fenced()) {
            None  // Fenced 的 Broker 视为不可用
        } else {
            Option(broker.listeners().get(listenerName.value())).map { listener =>
                new Node(brokerId, listener.host(), listener.port(), broker.rack())
            }
        }
    }.toJava
}

override def getAliveBrokerNodes(listenerName: ListenerName): List[Node] = {
    val image = _currentImage
    
    image.cluster().brokers().values().asScala
        .filterNot(_.fenced())  // 过滤 Fenced 的 Broker
        .flatMap { broker =>
            Option(broker.listeners().get(listenerName.value())).map { listener =>
                new Node(broker.id(), listener.host(), listener.port(), broker.rack())
            }
        }
        .toList.asJava
}

override def getTopicId(topicName: String): Uuid = {
    val image = _currentImage
    val topic = image.topics().getTopic(topicName)
    if (topic == null) Uuid.ZERO_UUID else topic.id()
}

override def getTopicName(topicId: Uuid): Optional[String] = {
    val image = _currentImage
    Option(image.topics().getTopic(topicId)).map(_.name()).toJava
}

override def topicNamesToIds(): Map[String, Uuid] = {
    val image = _currentImage
    image.topics().topicsByName().asScala
        .map { case (name, topic) => (name, topic.id()) }
        .toMap.asJava
}

public class MetadataDelta {
    private final MetadataImage image;  // 基线 Image
    private MetadataProvenance provenance;
    
    // 各类增量
    private FeaturesDelta featuresDelta;
    private ClusterDelta clusterDelta;
    private TopicsDelta topicsDelta;
    private ConfigurationsDelta configsDelta;
    private ClientQuotasDelta clientQuotasDelta;
    private ProducerIdsDelta producerIdsDelta;
    private AclsDelta aclsDelta;
    private ScramDelta scramDelta;
    private DelegationTokenDelta delegationTokensDelta;
    
    public MetadataDelta(MetadataImage image) {
        this.image = image;
        this.provenance = image.provenance();
    }
    
    // 应用单条元数据记录
    public void replay(ApiMessageAndVersion record) {
        ApiMessage message = record.message();
        switch (message.apiKey()) {
            case REGISTER_BROKER_RECORD:
                clusterDelta().replay((RegisterBrokerRecord) message);
                break;
            case TOPIC_RECORD:
                topicsDelta().replay((TopicRecord) message);
                break;
            case PARTITION_RECORD:
                topicsDelta().replay((PartitionRecord) message);
                break;
            case CONFIG_RECORD:
                configsDelta().replay((ConfigRecord) message);
                break;
            // ... 其他记录类型 ...
        }
    }
    
    // 生成新的 MetadataImage
    public MetadataImage apply() {
        return new MetadataImage(
            provenance,
            featuresDelta == null ? image.features() : featuresDelta.apply(),
            clusterDelta == null ? image.cluster() : clusterDelta.apply(),
            topicsDelta == null ? image.topics() : topicsDelta.apply(),
            configsDelta == null ? image.configs() : configsDelta.apply(),
            clientQuotasDelta == null ? image.clientQuotas() : clientQuotasDelta.apply(),
            producerIdsDelta == null ? image.producerIds() : producerIdsDelta.apply(),
            aclsDelta == null ? image.acls() : aclsDelta.apply(),
            scramDelta == null ? image.scram() : scramDelta.apply(),
            delegationTokensDelta == null ? image.delegationTokens() : delegationTokensDelta.apply()
        );
    }
}

public class TopicsDelta {
    private final TopicsImage image;
    private final Map<Uuid, TopicDelta> changedTopics = new HashMap<>();
    private final Map<Uuid, TopicImage> deletedTopics = new HashMap<>();
    
    public void replay(TopicRecord record) {
        TopicDelta delta = getOrCreateTopicDelta(record.topicId());
        delta.replay(record);
    }
    
    public void replay(PartitionRecord record) {
        TopicDelta delta = getOrCreateTopicDelta(record.topicId());
        delta.replay(record);
    }
    
    public void replay(RemoveTopicRecord record) {
        Uuid topicId = record.topicId();
        TopicImage topic = image.getTopic(topicId);
        if (topic != null) {
            changedTopics.remove(topicId);
            deletedTopics.put(topicId, topic);
        }
    }
    
    public TopicsImage apply() {
        Map<Uuid, TopicImage> newTopicsById = new HashMap<>(image.topicsById());
        Map<String, TopicImage> newTopicsByName = new HashMap<>(image.topicsByName());
        
        // 应用删除
        for (TopicImage deleted : deletedTopics.values()) {
            newTopicsById.remove(deleted.id());
            newTopicsByName.remove(deleted.name());
        }
        
        // 应用变更
        for (TopicDelta delta : changedTopics.values()) {
            TopicImage newTopic = delta.apply();
            newTopicsById.put(newTopic.id(), newTopic);
            newTopicsByName.put(newTopic.name(), newTopic);
        }
        
        return new TopicsImage(newTopicsById, newTopicsByName);
    }
}

public class TopicDelta {
    private final TopicImage image;  // 基线 Topic
    private String name;
    private Uuid id;
    private final Map<Integer, PartitionRegistration> changedPartitions = new HashMap<>();
    private final Set<Integer> deletedPartitions = new HashSet<>();
    
    public void replay(PartitionRecord record) {
        changedPartitions.put(record.partitionId(), new PartitionRegistration(
            Replicas.toArray(record.replicas()),
            Replicas.toArray(record.isr()),
            Replicas.toArray(record.removingReplicas()),
            Replicas.toArray(record.addingReplicas()),
            record.leader(),
            record.leaderEpoch(),
            record.partitionEpoch()
        ));
    }
    
    public TopicImage apply() {
        Map<Integer, PartitionRegistration> newPartitions =
            new HashMap<>(image.partitions());
        
        // 删除 Partitions
        for (Integer partitionId : deletedPartitions) {
            newPartitions.remove(partitionId);
        }
        
        // 更新/新增 Partitions
        newPartitions.putAll(changedPartitions);
        
        return new TopicImage(name, id, newPartitions);
    }
}

// 在 MetadataLoader 中
public void handleSnapshot(SnapshotReader<ApiMessageAndVersion> reader) {
    MetadataDelta delta = new MetadataDelta(image);
    
    while (reader.hasNext()) {
        Batch<ApiMessageAndVersion> batch = reader.next();
        for (ApiMessageAndVersion record : batch) {
            delta.replay(record);  // 应用每条记录
        }
    }
    
    // 更新 provenance
    delta.setProvenance(new MetadataProvenance(
        reader.lastContainedLogOffset(),
        reader.lastContainedLogEpoch(),
        reader.lastContainedLogTimeMs()
    ));
    
    // 生成新 Image 并发布
    MetadataImage newImage = delta.apply();
    publish(delta, newImage);
}

public void handleCommit(long offset, long epoch, long lastContainedLogTimeMs,
                         List<ApiMessageAndVersion> records) {
    MetadataDelta delta = new MetadataDelta(image);
    
    for (ApiMessageAndVersion record : records) {
        delta.replay(record);
    }
    
    delta.setProvenance(new MetadataProvenance(offset, epoch, lastContainedLogTimeMs));
    
    MetadataImage newImage = delta.apply();
    publish(delta, newImage);
}

public interface MetadataPublisher extends AutoCloseable {
    
    String name();
    
    /**

     * 初始化（在第一次发布前调用）
     */
    default void onMetadataUpdate(
        MetadataDelta delta,
        MetadataImage newImage,
        LoaderManifest manifest
    ) {}
    
    /**
     * 发布元数据变更
     */
    void onMetadataUpdate(
        MetadataDelta delta,
        MetadataImage newImage,
        LoaderManifest manifest
    );
    
    @Override
    default void close() throws Exception {}

}

class BrokerMetadataPublisher extends MetadataPublisher {
    private val metadataCache: KRaftMetadataCache
    private val logManager: LogManager
    private val replicaManager: ReplicaManager
    // ... 其他管理器 ...
    
    override def onMetadataUpdate(
        delta: MetadataDelta,
        newImage: MetadataImage,
        manifest: LoaderManifest
    ): Unit = {
        val highestOffsetAndEpoch = newImage.highestOffsetAndEpoch()
        
        // 1. 更新 MetadataCache
        metadataCache.setImage(newImage)
        
        if (_firstPublish) {
            info(s"Publishing initial metadata at offset $highestOffsetAndEpoch")
            initializeManagers(newImage)
        } else {
            debug(s"Publishing metadata at offset $highestOffsetAndEpoch")
        }
        
        // 2. 处理 Topic 变更
        Option(delta.topicsDelta()).foreach { topicsDelta =>
            // 删除 Topic
            topicsDelta.deletedTopicIds().forEach { topicId =>
                val topicName = newImage.topics().getTopic(topicId).name()
                replicaManager.stopReplicas(topicName)
                logManager.asyncDelete(topicName)
            }
            
            // 新增/变更 Partitions
            topicsDelta.changedTopics().forEach { (topicId, topicDelta) =>
                topicDelta.changedPartitions().forEach { (partitionId, partition) =>
                    val topicPartition = new TopicPartition(topicDelta.name(), partitionId)
                    
                    // 如果本 Broker 是 Replica，创建/更新 Partition
                    if (partition.replicas.contains(brokerId)) {
                        replicaManager.makeFollowers(
                            topicPartition,
                            partition.leader,
                            partition.leaderEpoch,
                            partition.isr
                        )
                    } else {
                        // 不再是 Replica，删除本地数据
                        replicaManager.stopReplicas(topicPartition)
                        logManager.asyncDelete(topicPartition)
                    }
                }
            }
        }
        
        // 3. 处理配置变更
        Option(delta.configsDelta()).foreach { configsDelta =>
            configsDelta.changes().forEach { (resource, configDelta) =>
                resource.`type`() match {
                    case ConfigResource.Type.TOPIC =>
                        logManager.updateTopicConfig(resource.name(), configDelta.data())
                    case ConfigResource.Type.BROKER =>
                        if (resource.name() == brokerId.toString) {
                            dynamicConfigManager.updateBrokerConfig(brokerId, configDelta.data())
                        }
                    case _ =>
                }
            }
        }
        
        _firstPublish = false
    }
}

class KRaftMetadataCachePublisher(
    val metadataCache: KRaftMetadataCache
) extends MetadataPublisher {
    
    override def name(): String = "KRaftMetadataCachePublisher"
    
    override def onMetadataUpdate(
        delta: MetadataDelta,
        newImage: MetadataImage,
        manifest: LoaderManifest
    ): Unit = {
        metadataCache.setImage(newImage)
    }
}

class FeaturesPublisher(
    logContext: LogContext,
    faultHandler: FaultHandler
) extends MetadataPublisher {
    
    @volatile private var _features: FinalizedFeatures = FinalizedFeatures.EMPTY
    
    def features(): FinalizedFeatures = _features
    
    override def onMetadataUpdate(
        delta: MetadataDelta,
        newImage: MetadataImage,
        manifest: LoaderManifest
    ): Unit = {
        Option(delta.featuresDelta()).foreach { featuresDelta =>
            val newFeatures = new FinalizedFeatures(
                newImage.features().metadataVersion(),
                newImage.features().supportedFeatures(),
                manifest.kraftVersion()
            )
            
            if (!_features.equals(newFeatures)) {
                info(s"Updating features from ${_features} to ${newFeatures}")
                _features = newFeatures
            }
        }
    }
}

sequenceDiagram
    autonumber
    participant Admin as AdminClient
    participant Ctrl as QuorumController
    participant Raft as RaftLog
    participant Loader as MetadataLoader
    participant Delta as MetadataDelta
    participant Image as MetadataImage
    participant Pubs as Publishers
    participant Cache as KRaftMetadataCache
    participant Mgrs as Managers (LogMgr/ReplicaMgr)
    
    Admin->>Ctrl: CreateTopicsRequest
    Ctrl->>Ctrl: 验证请求、生成记录
    Ctrl->>Raft: append(TopicRecord, PartitionRecord)
    
    Note over Raft: Raft Quorum 复制日志
    
    Raft->>Raft: commit(offset=100)
    Raft->>Loader: handleCommit(offset=100, records)
    
    Loader->>Delta: new MetadataDelta(currentImage)
    loop 每条记录
        Loader->>Delta: replay(record)
        Delta->>Delta: 应用到相应 Delta
    end
    
    Loader->>Delta: setProvenance(offset=100, epoch=5)
    Loader->>Image: delta.apply()
    Image->>Image: 生成新 MetadataImage
    
    Loader->>Pubs: onMetadataUpdate(delta, newImage)
    
    par 并行通知所有 Publishers
        Pubs->>Cache: KRaftMetadataCachePublisher.onMetadataUpdate
        Cache->>Cache: setImage(newImage)
        
        Pubs->>Mgrs: BrokerMetadataPublisher.onMetadataUpdate
        Mgrs->>Mgrs: 创建 Partitions、更新配置
        
        Pubs->>Pubs: FeaturesPublisher.onMetadataUpdate
    end
    
    Ctrl-->>Admin: CreateTopicsResponse

// 在 QuorumController 中
public CompletableFuture<CreateTopicsResponseData> createTopics(
    CreateTopicsRequestData request
) {
    CompletableFuture<CreateTopicsResponseData> future = new CompletableFuture<>();
    
    appendControlEvent("createTopics", () -> {
        CreateTopicsResponseData response = new CreateTopicsResponseData();
        
        for (CreatableTopic topic : request.topics()) {
            try {
                // 1. 验证 Topic 名称、配置
                validateTopicName(topic.name());
                validateTopicConfigs(topic.configs());
                
                // 2. 分配 Partitions 到 Brokers
                List<List<Integer>> replicaAssignment =
                    assignReplicasToBrokers(topic.numPartitions(), topic.replicationFactor());
                
                // 3. 生成元数据记录
                Uuid topicId = Uuid.randomUuid();
                List<ApiMessageAndVersion> records = new ArrayList<>();
                
                records.add(new ApiMessageAndVersion(
                    new TopicRecord()
                        .setName(topic.name())
                        .setTopicId(topicId),
                    (short) 0
                ));
                
                for (int partitionId = 0; partitionId < topic.numPartitions(); partitionId++) {
                    List<Integer> replicas = replicaAssignment.get(partitionId);
                    records.add(new ApiMessageAndVersion(
                        new PartitionRecord()
                            .setTopicId(topicId)
                            .setPartitionId(partitionId)
                            .setReplicas(replicas)
                            .setIsr(replicas)
                            .setLeader(replicas.get(0))
                            .setLeaderEpoch(0)
                            .setPartitionEpoch(0),
                        (short) 0
                    ));
                }
                
                // 4. 写入 Raft Log
                appendRecords(records).thenApply(offset -> {
                    response.topics().add(new CreatableTopicResult()
                        .setName(topic.name())
                        .setTopicId(topicId)
                        .setErrorCode(Errors.NONE.code()));
                    return null;
                });
                
            } catch (Exception e) {
                response.topics().add(new CreatableTopicResult()
                    .setName(topic.name())
                    .setErrorCode(Errors.forException(e).code())
                    .setErrorMessage(e.getMessage()));
            }
        }
        
        future.complete(response);
    });
    
    return future;
}

// 在 KafkaRaftClient 中
public long append(List<ApiMessageAndVersion> records, long maxBatchSizeInBytes) {
    // 1. 序列化记录
    MemoryRecords memoryRecords = buildMemoryRecords(records);
    
    // 2. 追加到本地 Log
    long offset = log.appendAsLeader(memoryRecords);
    
    // 3. 触发 Followers 拉取
    followers.forEach(follower -> follower.wakeup());
    
    // 4. 等待多数派确认
    return offset;
}

// Followers 拉取并确认
public void handleFetchRequest(FetchRequest request) {
    // 从 Log 读取记录
    Records records = log.read(request.fetchOffset(), request.maxBytes());
    
    // 返回给 Follower
    FetchResponse response = new FetchResponse()
        .setRecords(records)
        .setHighWatermark(log.highWatermark());
    
    send(response);
}

// Leader 更新 High Watermark
public void maybeAdvanceHighWatermark() {
    // 计算多数派的最小 offset
    long newHighWatermark = calculateHighWatermark(followers);
    
    if (newHighWatermark > log.highWatermark()) {
        log.updateHighWatermark(newHighWatermark);
        // 触发 commit 回调
        handleCommit(newHighWatermark);
    }
}

// 在 MetadataLoader 中
public void handleCommit(long offset, long epoch, long lastContainedLogTimeMs,
                         List<ApiMessageAndVersion> records) {
    MetadataDelta delta = new MetadataDelta(image);
    
    for (ApiMessageAndVersion record : records) {
        try {
            delta.replay(record);
        } catch (Exception e) {
            faultHandler.handleFault("Error replaying record at offset " + offset, e);
            throw e;
        }
    }
    
    delta.setProvenance(new MetadataProvenance(offset, epoch, lastContainedLogTimeMs));
    
    MetadataImage newImage = delta.apply();
    publish(delta, newImage);
}

// 在 MetadataDelta 中
public void replay(ApiMessageAndVersion record) {
    ApiMessage message = record.message();
    
    switch (message.apiKey()) {
        case TOPIC_RECORD:
            topicsDelta().replay((TopicRecord) message);
            break;
            
        case PARTITION_RECORD:
            topicsDelta().replay((PartitionRecord) message);
            break;
            
        case REGISTER_BROKER_RECORD:
            clusterDelta().replay((RegisterBrokerRecord) message);
            break;
            
        case CONFIG_RECORD:
            configsDelta().replay((ConfigRecord) message);
            break;
            
        // ... 其他记录类型 ...
    }
}

// 在 TopicsDelta 中
public void replay(TopicRecord record) {
    TopicDelta delta = getOrCreateTopicDelta(record.topicId());
    delta.setName(record.name());
    delta.setId(record.topicId());
}

public void replay(PartitionRecord record) {
    TopicDelta delta = getOrCreateTopicDelta(record.topicId());
    delta.replay(record);
}

// 在 TopicDelta 中
public void replay(PartitionRecord record) {
    changedPartitions.put(record.partitionId(), new PartitionRegistration(
        Replicas.toArray(record.replicas()),
        Replicas.toArray(record.isr()),
        Replicas.toArray(record.removingReplicas()),
        Replicas.toArray(record.addingReplicas()),
        record.leader(),
        record.leaderEpoch(),
        record.partitionEpoch()
    ));
}