flowchart TB
    subgraph "Kafka 集群"
        Controller[Controller<br/>元数据管理]
        Broker1[Broker 1<br/>Core 模块]
        Broker2[Broker 2<br/>Core 模块]
        Broker3[Broker 3<br/>Core 模块]
        
        Controller -->|元数据更新| Broker1
        Controller -->|元数据更新| Broker2
        Controller -->|元数据更新| Broker3
        
        Broker1 <-->|副本同步| Broker2
        Broker2 <-->|副本同步| Broker3
        Broker3 <-->|副本同步| Broker1
    end
    
    subgraph "客户端"
        Producer[Producer]
        Consumer[Consumer]
    end
    
    Producer -->|ProduceRequest| Broker1
    Consumer -->|FetchRequest| Broker2
    
    subgraph "存储"
        Disk1[(磁盘 1)]
        Disk2[(磁盘 2)]
        Disk3[(磁盘 3)]
    end
    
    Broker1 --> Disk1
    Broker2 --> Disk2
    Broker3 --> Disk3

flowchart TB
    subgraph "网络层 Network Layer"
        Acceptor[Acceptor<br/>接受连接]
        Processor1[Processor 1<br/>I/O处理]
        Processor2[Processor 2<br/>I/O处理]
        ProcessorN[Processor N<br/>I/O处理]
        
        Acceptor --> Processor1
        Acceptor --> Processor2
        Acceptor --> ProcessorN
    end
    
    subgraph "请求队列 Request Queue"
        RequestChannel[RequestChannel<br/>请求队列]
    end
    
    Processor1 --> RequestChannel
    Processor2 --> RequestChannel
    ProcessorN --> RequestChannel
    
    subgraph "请求处理层 Request Handler"
        Handler1[Handler 1]
        Handler2[Handler 2]
        HandlerN[Handler N]
        KafkaApis[KafkaApis<br/>请求分发]
    end
    
    RequestChannel --> Handler1
    RequestChannel --> Handler2
    RequestChannel --> HandlerN
    Handler1 --> KafkaApis
    Handler2 --> KafkaApis
    HandlerN --> KafkaApis
    
    subgraph "核心服务层 Core Services"
        ReplicaManager[ReplicaManager<br/>副本管理]
        LogManager[LogManager<br/>日志管理]
        MetadataCache[MetadataCache<br/>元数据缓存]
        GroupCoordinator[GroupCoordinator<br/>消费者组协调]
        TxnCoordinator[TransactionCoordinator<br/>事务协调]
    end
    
    KafkaApis --> ReplicaManager
    KafkaApis --> LogManager
    KafkaApis --> MetadataCache
    KafkaApis --> GroupCoordinator
    KafkaApis --> TxnCoordinator
    
    subgraph "存储层 Storage"
        Log[Log<br/>分区日志]
        LogSegment[LogSegment<br/>日志段]
        Index[Index<br/>索引文件]
    end
    
    LogManager --> Log
    ReplicaManager --> Log
    Log --> LogSegment
    Log --> Index
    
    subgraph "元数据层 Metadata"
        KRaftManager[KRaftManager<br/>Raft管理]
        MetadataPublisher[MetadataPublisher<br/>元数据发布]
    end
    
    KRaftManager --> MetadataPublisher
    MetadataPublisher --> MetadataCache
    MetadataPublisher --> ReplicaManager

1 个 Acceptor 线程（每个监听端口）
  ↓
N 个 Processor 线程（默认 3 个，可配置 num.network.threads）
  ↓
RequestChannel（请求队列）
  ↓
M 个 RequestHandler 线程（默认 8 个，可配置 num.io.threads）

// 所有分区的映射：TopicPartition -> HostedPartition
allPartitions: ConcurrentHashMap[TopicPartition, HostedPartition]

// HostedPartition 有三种状态：
sealed trait HostedPartition
case object None extends HostedPartition  // 无状态
case class Online(partition: Partition) extends HostedPartition  // 在线
case class Offline(partition: Option[Partition]) extends HostedPartition  // 离线

/var/kafka-logs/
  ├── topic1-0/              # Topic: topic1, Partition: 0
  │   ├── 00000000000000000000.log      # 日志段文件
  │   ├── 00000000000000000000.index    # 偏移量索引
  │   ├── 00000000000000000000.timeindex # 时间戳索引
  │   ├── 00000000000000012345.log
  │   ├── 00000000000000012345.index
  │   └── 00000000000000012345.timeindex
  ├── topic1-1/
  └── topic2-0/

sequenceDiagram
    autonumber
    participant Main as Kafka.main()
    participant Server as KafkaRaftServer
    participant Shared as SharedServer
    participant Controller as ControllerServer
    participant Broker as BrokerServer
    participant KRaft as KRaftManager
    participant RM as ReplicaManager
    participant LM as LogManager
    participant SS as SocketServer

    Main->>Server: new KafkaRaftServer(config)
    Server->>Server: initializeLogDirs()
    Server->>Shared: new SharedServer()
    
    alt 如果配置了 Controller 角色
        Server->>Controller: new ControllerServer()
    end
    
    alt 如果配置了 Broker 角色
        Server->>Broker: new BrokerServer()
    end
    
    Main->>Server: startup()
    
    alt 启动 Controller（如果存在）
        Server->>Controller: startup()
        Controller->>KRaft: start()
        KRaft->>KRaft: 加载元数据日志
        KRaft->>KRaft: 参与 Leader 选举
    end
    
    Server->>Broker: startup()
    Broker->>Shared: startForBroker()
    Shared->>KRaft: start() 或 connect()
    
    Broker->>Broker: 初始化 KafkaScheduler
    Broker->>Broker: 初始化 QuotaManagers
    Broker->>Broker: 初始化 BrokerTopicStats
    
    Broker->>LM: new LogManager()
    LM->>LM: loadLogs() 加载所有日志
    LM->>LM: startLogCleaner() 启动清理线程
    
    Broker->>RM: new ReplicaManager()
    RM->>RM: 初始化延迟操作 Purgatory
    
    Broker->>SS: new SocketServer()
    SS->>SS: 启动 Acceptor 线程
    SS->>SS: 启动 Processor 线程
    
    Broker->>Broker: 启动 KafkaRequestHandler 线程池
    
    Broker->>Broker: 注册到 Controller
    Broker->>Broker: 等待 Controller 分配分区
    
    Broker->>SS: enableRequestProcessing()
    SS->>SS: 开始接受客户端连接
    
    Broker->>Main: 启动完成

// KafkaRaftServer.initializeLogDirs()
private def initializeLogDirs(): (MetaPropertiesEnsemble, BootstrapMetadata) = {
  // 1. 加载 meta.properties 文件（包含 cluster.id, node.id, version）
  val metaPropsEnsemble = new MetaPropertiesEnsemble.Loader()
    .addLogDirs(config.logDirs)
    .load()
  
  // 2. 验证 cluster.id 一致性
  if (!metaPropsEnsemble.clusterId().isPresent) {
    throw new RuntimeException("未找到 cluster.id")
  }
  
  // 3. 加载引导元数据（如果存在）
  val bootstrapMetadata = BootstrapMetadata.fromVersion(...)
  
  (metaPropsEnsemble, bootstrapMetadata)
}

// SharedServer.startForBroker()
def startForBroker(): Unit = {
  // 启动 Raft 管理器（连接到 Controller Quorum）
  raftManager.startup()
  
  // 注册元数据监听器
  raftManager.register(new BrokerMetadataPublisher())
}

// LogManager.loadLogs()
private def loadLogs(): Unit = {
  // 1. 扫描所有日志目录
  for (dir <- logDirs) {
    val dirTopics = dir.listFiles.filter(_.isDirectory)
    
    // 2. 为每个分区目录加载日志
    for (topicDir <- dirTopics) {
      val (topic, partition) = parseTopicPartition(topicDir.getName)
      val log = Log.loadLog(topicDir, ...)
      logs.put(TopicPartition(topic, partition), log)
    }
  }
  
  // 3. 启动日志清理线程
  if (cleanerConfig.enableCleaner) {
    cleaner = new LogCleaner(...)
    cleaner.startup()
  }
}

// ReplicaManager.startup()
def startup(): Unit = {
  // 1. 启动 ISR 过期检查线程（每 replicaLagTimeMaxMs / 2 执行一次）
  scheduler.schedule("isr-expiration", () => maybeShrinkIsr(), ...)
  
  // 2. 启动日志目录故障处理线程
  logDirFailureHandler.start()
  
  // 3. 启动延迟操作 Purgatory
  delayedProducePurgatory.start()
  delayedFetchPurgatory.start()
}

// SocketServer.startup()
def startup(): Unit = {
  // 1. 为每个监听器创建 Acceptor 和 Processors
  config.listeners.foreach { listener =>
    val acceptor = new Acceptor(listener, ...)
    val processors = (0 until numProcessorThreads).map { i =>
      new Processor(i, requestChannel, ...)
    }
    
    acceptor.start()
    processors.foreach(_.start())
  }
}

// BrokerLifecycleManager.start()
def start(): Unit = {
  // 1. 发送 BrokerRegistrationRequest
  sendBrokerRegistration()
  
  // 2. 启动心跳线程
  scheduler.schedule("broker-heartbeat", () => sendHeartbeat(), ...)
}

sequenceDiagram
    autonumber
    participant P as Producer
    participant Proc as Processor
    participant RC as RequestChannel
    participant H as Handler
    participant KA as KafkaApis
    participant RM as ReplicaManager
    participant Log as Log
    participant F as Follower

    P->>Proc: TCP 数据包
    Proc->>Proc: 解析 ProduceRequest
    Proc->>RC: 放入请求队列
    
    RC->>H: 取出请求
    H->>KA: handle(ProduceRequest)
    
    KA->>KA: authorize() 权限检查
    KA->>KA: checkQuota() 限流检查
    
    KA->>RM: appendRecords()
    RM->>RM: 检查是否为 Leader
    RM->>Log: append()
    Log->>Log: 写入 LogSegment
    Log->>Log: 更新 LEO
    
    alt acks=all
        RM->>RM: 等待 ISR 副本确认
        F->>RM: FetchRequest
        RM->>RM: 更新 Follower LEO
        RM->>RM: 更新 HW
        RM->>RM: 检查 ISR 是否都已确认
    end
    
    RM-->>KA: AppendResult
    KA->>KA: 构造 ProduceResponse
    KA->>RC: 放入响应队列
    RC->>Proc: 取出响应
    Proc->>P: TCP 响应

def handleProduceRequest(request: RequestChannel.Request, requestLocal: RequestLocal): Unit = {
  val produceRequest = request.body[ProduceRequest]
  
  // 1. 权限检查
  val unauthorizedTopicResponses = authorizeProduceTopics(produceRequest)
  
  // 2. 限流检查
  val bandwidthThrottleTimeMs = quotas.produce.maybeRecordAndGetThrottleTimeMs(...)
  
  // 3. 调用 ReplicaManager 写入
  replicaManager.appendRecords(
    timeout = produceRequest.timeout,
    requiredAcks = produceRequest.acks,
    internalTopicsAllowed = false,
    transactional = produceRequest.isTransactional,
    entriesPerPartition = produceRequest.partitionRecords,
    responseCallback = sendResponseCallback(request, ...)
  )
}

def appendRecords(
  timeout: Long,
  requiredAcks: Short,
  internalTopicsAllowed: Boolean,
  transactional: Boolean,
  entriesPerPartition: Map[TopicPartition, MemoryRecords],
  responseCallback: Map[TopicPartition, PartitionResponse] => Unit
): Unit = {
  
  // 1. 按分区追加消息
  val localAppendResults = entriesPerPartition.map { case (tp, records) =>
    val partition = getPartitionOrException(tp)
    
    // 2. 检查是否为 Leader
    if (!partition.isLeader) {
      throw new NotLeaderForPartitionException(...)
    }
    
    // 3. 追加到 Log
    val info = partition.appendRecordsToLeader(records, ...)
    (tp, LogAppendResult(info))
  }
  
  // 4. 根据 acks 决定是否延迟响应
  if (requiredAcks == -1) {  // acks=all
    val delayedProduce = new DelayedProduce(timeout, localAppendResults, ...)
    delayedProducePurgatory.tryCompleteElseWatch(delayedProduce, ...)
  } else {
    // acks=0 或 acks=1，立即响应
    responseCallback(localAppendResults)
  }
}

sequenceDiagram
    autonumber
    participant C as Consumer
    participant Proc as Processor
    participant RC as RequestChannel
    participant H as Handler
    participant KA as KafkaApis
    participant RM as ReplicaManager
    participant Log as Log

    C->>Proc: FetchRequest
    Proc->>RC: 放入请求队列
    RC->>H: 取出请求
    H->>KA: handle(FetchRequest)
    
    KA->>KA: authorize() 权限检查
    KA->>RM: fetchMessages()
    
    RM->>RM: 检查是否有足够数据
    
    alt 数据足够
        RM->>Log: read()
        Log->>Log: 从 LogSegment 读取
        Log-->>RM: 消息数据
        RM-->>KA: FetchResponse
    else 数据不足且未超时
        RM->>RM: 创建 DelayedFetch
        RM->>RM: 等待数据到达或超时
        Note over RM: 新数据到达时触发
        RM->>Log: read()
        Log-->>RM: 消息数据
        RM-->>KA: FetchResponse
    end
    
    KA->>RC: 放入响应队列
    RC->>Proc: 取出响应
    Proc->>C: FetchResponse

def handleFetchRequest(request: RequestChannel.Request): Unit = {
  val fetchRequest = request.body[FetchRequest]
  
  // 1. 权限检查
  val authorizedTopicPartitions = authorizeFetchTopics(fetchRequest)
  
  // 2. 调用 ReplicaManager 读取
  replicaManager.fetchMessages(
    timeout = fetchRequest.maxWait,
    replicaId = fetchRequest.replicaId,
    fetchMinBytes = fetchRequest.minBytes,
    fetchMaxBytes = fetchRequest.maxBytes,
    hardMaxBytesLimit = fetchRequest.metadata.hardMaxBytesLimit,
    fetchInfos = fetchRequest.fetchData,
    quota = quota,
    responseCallback = sendResponseCallback(request, ...)
  )
}

def fetchMessages(
  timeout: Long,
  replicaId: Int,
  fetchMinBytes: Int,
  fetchMaxBytes: Int,
  fetchInfos: Seq[(TopicPartition, PartitionData)],
  quota: ReplicaQuota,
  responseCallback: Seq[(TopicPartition, FetchPartitionData)] => Unit
): Unit = {
  
  // 1. 从各分区读取消息
  val logReadResults = fetchInfos.map { case (tp, partitionData) =>
    val partition = getPartitionOrException(tp)
    val fetchInfo = partition.readRecords(
      fetchOffset = partitionData.fetchOffset,
      maxBytes = partitionData.maxBytes,
      ...
    )
    (tp, fetchInfo)
  }
  
  // 2. 检查是否满足 fetch.min.bytes
  val bytesReadable = logReadResults.map(_._2.sizeInBytes).sum
  
  if (bytesReadable >= fetchMinBytes || timeout == 0) {
    // 数据足够或不等待，立即响应
    responseCallback(logReadResults)
  } else {
    // 数据不足，创建延迟操作
    val delayedFetch = new DelayedFetch(timeout, fetchMinBytes, ...)
    delayedFetchPurgatory.tryCompleteElseWatch(delayedFetch, ...)
  }
}

flowchart LR
    subgraph "Leader Broker"
        L[Leader Replica]
        LLog[Leader Log]
        L --> LLog
    end
    
    subgraph "Follower Broker 1"
        F1[Follower Replica]
        F1Fetcher[ReplicaFetcher Thread]
        F1Log[Follower Log]
        F1Fetcher --> F1
        F1 --> F1Log
    end
    
    subgraph "Follower Broker 2"
        F2[Follower Replica]
        F2Fetcher[ReplicaFetcher Thread]
        F2Log[Follower Log]
        F2Fetcher --> F2
        F2 --> F2Log
    end
    
    F1Fetcher -->|FetchRequest| L
    F2Fetcher -->|FetchRequest| L
    L -->|FetchResponse| F1Fetcher
    L -->|FetchResponse| F2Fetcher

// ReplicaManager.maybeShrinkIsr()
private def maybeShrinkIsr(): Unit = {
  allPartitions.foreach { (tp, partition) =>
    partition.maybeShrinkIsr()
  }
}

// Partition.maybeShrinkIsr()
def maybeShrinkIsr(): Unit = {
  val now = time.milliseconds()
  val laggingReplicas = inSyncReplicaIds.filter { replicaId =>
    val replica = getReplica(replicaId)
    replica.lastCaughtUpTimeMs < now - replicaLagTimeMaxMs
  }
  
  if (laggingReplicas.nonEmpty) {
    val newIsr = inSyncReplicaIds -- laggingReplicas
    alterPartitionManager.submit(topicPartition, newLeaderAndIsr(newIsr))
  }
}

// Partition.maybeExpandIsr()
def maybeExpandIsr(replicaId: Int): Unit = {
  if (!inSyncReplicaIds.contains(replicaId)) {
    val replica = getReplica(replicaId)
    if (replica.logEndOffset >= highWatermark) {
      val newIsr = inSyncReplicaIds + replicaId
      alterPartitionManager.submit(topicPartition, newLeaderAndIsr(newIsr))
    }
  }
}

// FileRecords.writeTo()
def writeTo(channel: GatheringByteChannel, position: Long, length: Int): Int = {
  // 使用 FileChannel.transferTo() 实现零拷贝
  channel.write(buffer.duplicate())
}

// ProduceRequest.java
public class ProduceRequestData {
    private String transactionalId;          // 事务 ID（可选）
    private short acks;                      // 确认级别：-1/0/1
    private int timeoutMs;                   // 请求超时
    private List<TopicProduceData> topicData; // Topic 数据列表
}

public class TopicProduceData {
    private String name;                     // Topic 名称
    private List<PartitionProduceData> partitionData; // 分区数据列表
}

public class PartitionProduceData {
    private int index;                       // 分区 ID
    private MemoryRecords records;           // 消息记录
}

public class ProduceResponseData {
    private List<TopicProduceResponse> responses;
    private int throttleTimeMs;
}

public class TopicProduceResponse {
    private String name;
    private List<PartitionProduceResponse> partitionResponses;
}

public class PartitionProduceResponse {
    private int index;                    // 分区 ID
    private short errorCode;              // 错误码
    private long baseOffset;              // 起始偏移量
    private long logAppendTimeMs;         // 追加时间
    private long logStartOffset;          // 日志起始偏移量
    private List<RecordError> recordErrors; // 记录级错误（批次中部分失败）
    private String errorMessage;          // 错误消息
}

// KafkaApis.scala
def handleProduceRequest(request: RequestChannel.Request, requestLocal: RequestLocal): Unit = {
  val produceRequest = request.body[ProduceRequest]
  val numBytesAppended = request.header.toStruct.sizeOf + request.sizeOfBodyInBytes
  
  // 权限验证
  val unauthorizedTopicResponses = mutable.Map[TopicPartition, PartitionResponse]()
  val nonExistingTopicResponses = mutable.Map[TopicPartition, PartitionResponse]()
  val invalidRequestResponses = mutable.Map[TopicPartition, PartitionResponse]()
  
  // 验证请求
  produceRequest.data.topicData.forEach { topic =>
    topic.partitionData.forEach { partition =>
      val tp = new TopicPartition(topic.name, partition.index)
      
      // 检查授权
      if (!authorize(request.context, WRITE, TOPIC, topic.name)) {
        unauthorizedTopicResponses += tp -> new PartitionResponse(
          Errors.TOPIC_AUTHORIZATION_FAILED)
      }
      // 检查 Topic 是否存在
      else if (!metadataCache.contains(topic.name)) {
        nonExistingTopicResponses += tp -> new PartitionResponse(
          Errors.UNKNOWN_TOPIC_OR_PARTITION)
      }
    }
  }
  
  // 转换为追加操作
  val authorizedRequestInfo = mutable.Map[TopicPartition, MemoryRecords]()
  produceRequest.data.topicData.forEach { topic =>
    topic.partitionData.forEach { partition =>
      val tp = new TopicPartition(topic.name, partition.index)
      if (!unauthorizedTopicResponses.contains(tp) &&
          !nonExistingTopicResponses.contains(tp)) {
        authorizedRequestInfo += tp -> partition.records
      }
    }
  }
  
  // 追加到日志
  replicaManager.appendRecords(
    timeout = produceRequest.timeout.toLong,
    requiredAcks = produceRequest.acks,
    internalTopicsAllowed = request.header.clientId == AdminClientId,
    origin = AppendOrigin.Client,
    entriesPerPartition = authorizedRequestInfo,
    responseCallback = sendResponseCallback,
    requestLocal = requestLocal
  )
}

handleProduceRequest
  ↓
ReplicaManager.appendRecords
  ↓
Partition.appendRecordsToLeader
  ↓
UnifiedLog.appendAsLeader
  ↓
LogSegment.append
  ↓
FileRecords.append

sequenceDiagram
    autonumber
    participant P as Producer
    participant N as NetworkServer
    participant A as KafkaApis
    participant R as ReplicaManager
    participant L as UnifiedLog
    participant F as FileRecords
    
    P->>N: ProduceRequest
    N->>A: handleProduceRequest()
    
    A->>A: 权限验证
    A->>A: Topic 存在性检查
    
    A->>R: appendRecords()
    
    R->>R: 获取 Partition
    R->>L: appendAsLeader()
    
    L->>L: 验证 sequence number
    L->>F: append()
    
    F->>F: 写入磁盘
    F-->>L: baseOffset
    
    L-->>R: LogAppendInfo
    
    alt acks=-1 (所有 ISR)
        R->>R: 等待 ISR 确认
        R->>R: updateHighWatermark()
    else acks=1 (Leader only)
        R->>R: Leader 写入即返回
    end
    
    R-->>A: ProduceResult
    A-->>N: ProduceResponse
    N-->>P: Response

# 吞吐量优先
acks=1
linger.ms=10
batch.size=32768
compression.type=lz4

# 可靠性优先
acks=-1
linger.ms=0
max.in.flight.requests.per.connection=1
enable.idempotence=true

public class FetchRequestData {
    private int replicaId;                   // Replica ID（-1=Consumer）
    private int maxWaitMs;                   // 最大等待时间
    private int minBytes;                    // 最小字节数
    private int maxBytes;                    // 最大字节数
    private byte isolationLevel;             // 隔离级别
    private int sessionId;                   // Session ID
    private int sessionEpoch;                // Session Epoch
    private List<FetchTopic> topics;         // Topic 列表
    private List<ForgottenTopic> forgottenTopicsData; // 不再 Fetch 的 Topic
    private String rackId;                   // 机架 ID
}

public class FetchTopic {
    private Uuid topicId;                    // Topic UUID
    private List<FetchPartition> partitions; // 分区列表
}

public class FetchPartition {
    private int partition;                   // 分区 ID
    private long fetchOffset;                // 拉取起始偏移量
    private int partitionMaxBytes;           // 分区最大字节数
    private int currentLeaderEpoch;          // 当前 Leader Epoch
    private long logStartOffset;             // 日志起始偏移量
}

public class FetchResponseData {
    private int throttleTimeMs;
    private short errorCode;
    private int sessionId;
    private List<FetchableTopicResponse> responses;
}

public class FetchableTopicResponse {
    private Uuid topicId;
    private List<PartitionData> partitions;
}

public class PartitionData {
    private int partitionIndex;
    private short errorCode;
    private long highWatermark;              // High Watermark
    private long lastStableOffset;           // Last Stable Offset（事务）
    private long logStartOffset;             // 日志起始偏移量
    private List<AbortedTransaction> abortedTransactions; // 已中止事务
    private int preferredReadReplica;        // 推荐读取副本
    private MemoryRecords records;           // 消息记录
}

def handleFetchRequest(request: RequestChannel.Request): Unit = {
  val fetchRequest = request.body[FetchRequest]
  val versionId = request.header.apiVersion
  
  // 权限验证
  val fetchContext = fetchManager.newContext(
    fetchRequest.metadata,
    fetchRequest.fetchData,
    fetchRequest.toForget,
    fetchRequest.isFromFollower
  )
  
  // 构建 Fetch 参数
  val fetchParams = FetchParams(
    requestVersion = versionId,
    replicaId = fetchRequest.replicaId,
    maxWaitMs = fetchRequest.maxWait,
    minBytes = fetchRequest.minBytes,
    maxBytes = fetchRequest.maxBytes,
    isolation = fetchRequest.isolationLevel,
    clientMetadata = fetchRequest.metadata
  )
  
  // 从副本管理器拉取
  replicaManager.fetchMessages(
    fetchParams,
    fetchContext.getFetchData,
    quota,
    responseCallback
  )
}

sequenceDiagram
    autonumber
    participant C as Consumer
    participant A as KafkaApis
    participant R as ReplicaManager
    participant P as Partition
    participant L as UnifiedLog
    
    C->>A: FetchRequest(offset=100)
    A->>A: 权限验证
    A->>R: fetchMessages()
    
    R->>P: readRecords()
    P->>L: read(startOffset=100)
    
    alt 有数据
        L->>L: 读取 FileRecords
        L-->>P: Records
        P-->>R: FetchDataInfo
        R-->>A: FetchResult
        A-->>C: FetchResponse(records)
    else 无数据且 maxWaitMs > 0
        R->>R: 加入 DelayedFetch
        Note over R: 等待新数据或超时
        R->>R: 超时或新数据到达
        R-->>A: FetchResult
        A-->>C: FetchResponse
    end

public class MetadataRequestData {
    private List<MetadataRequestTopic> topics;
    private boolean allowAutoTopicCreation;
    private boolean includeTopicAuthorizedOperations;
}

public class MetadataRequestTopic {
    private Uuid topicId;
    private String name;
}

public class MetadataResponseData {
    private int throttleTimeMs;
    private List<MetadataResponseBroker> brokers;
    private String clusterId;
    private int controllerId;
    private List<MetadataResponseTopic> topics;
}

public class MetadataResponseBroker {
    private int nodeId;
    private String host;
    private int port;
    private String rack;
}

public class MetadataResponseTopic {
    private short errorCode;
    private String name;
    private Uuid topicId;
    private boolean isInternal;
    private List<MetadataResponsePartition> partitions;
    private int topicAuthorizedOperations;
}

public class MetadataResponsePartition {
    private short errorCode;
    private int partitionIndex;
    private int leaderId;
    private int leaderEpoch;
    private List<Integer> replicaNodes;
    private List<Integer> isrNodes;
    private List<Integer> offlineReplicas;
}

classDiagram
    class Partition {
        +TopicPartition topicPartition
        +ReplicaManager replicaManager
        +UnifiedLog log
        +ReplicationQuotaManager quotaManager
        
        +Option~LeaderAndIsr~ leaderReplicaIdOpt
        +Set~Int~ inSyncReplicaIds
        +Set~Int~ assignedReplicas
        
        +makeLeader()
        +makeFollower()
        +appendRecordsToLeader()
        +fetchOffsetSnapshot()
        +maybeIncrementLeaderHW()
        +maybeShrinkIsr()
    }
    
    class UnifiedLog {
        +File dir
        +LogConfig config
        +LogSegments segments
        +ProducerStateManager producerStateManager
        
        +appendAsLeader()
        +appendAsFollower()
        +read()
        +roll()
        +flush()
    }
    
    class ReplicaManager {
        +Map~TopicPartition, HostedPartition~ allPartitions
        +ReplicaFetcherManager replicaFetcherManager
        +DelayedOperationPurgatory~DelayedProduce~ delayedProducePurgatory
        
        +appendRecords()
        +fetchMessages()
        +becomeLeaderOrFollower()
        +stopReplicas()
    }
    
    Partition --> UnifiedLog : contains
    ReplicaManager --> Partition : manages

def makeLeader(
    partitionState: LeaderAndIsr,
    highWatermarkCheckpoints: OffsetCheckpoints,
    topicId: Option[Uuid]
): Boolean = {
    inWriteLock(leaderIsrUpdateLock) {
        // 更新 leader epoch
        val newLeaderEpoch = partitionState.leaderEpoch
        
        // 更新 ISR
        val newInSyncReplicaIds = partitionState.isr.toSet
        
        // 恢复或创建日志
        val log = createLogIfNotExists(
            isNew = false,
            isFutureReplica = false,
            highWatermarkCheckpoints,
            topicId
        )
        
        // 更新状态
        this.leaderReplicaIdOpt = Some(localBrokerId)
        this.leaderEpoch = newLeaderEpoch
        this.inSyncReplicaIds = newInSyncReplicaIds
        
        // 初始化 high watermark
        val hw = log.highWatermark
        log.maybeIncrementHighWatermark(log.logEndOffsetMetadata)
        
        true
    }
}

def appendRecordsToLeader(
    records: MemoryRecords,
    origin: AppendOrigin,
    requiredAcks: Int,
    requestLocal: RequestLocal
): LogAppendInfo = {
    // 验证是 Leader
    if (!isLeader) {
        throw new NotLeaderOrFollowerException(
            s"Leader not local for partition $topicPartition"
        )
    }
    
    val log = localLogOrException
    val minIsr = log.config.minInSyncReplicas
    
    // 检查 ISR 大小
    if (inSyncReplicaIds.size < minIsr && requiredAcks == -1) {
        throw new NotEnoughReplicasException(
            s"Number of insync replicas for partition $topicPartition is " +
            s"[${inSyncReplicaIds.size}], below required minimum [$minIsr]"
        )
    }
    
    // 追加到日志
    val info = log.appendAsLeader(records, leaderEpoch, origin, requestLocal)
    
    // 更新 high watermark
    maybeIncrementLeaderHW(log)
    
    info
}

private def maybeIncrementLeaderHW(
    leaderLog: UnifiedLog,
    currentTimeMs: Long = time.milliseconds()
): Boolean = {
    // 收集所有 ISR 的 LEO
    val leaderLogEndOffset = leaderLog.logEndOffsetMetadata
    val allLogEndOffsets = inSyncReplicaIds.map { replicaId =>
        if (replicaId == localBrokerId) {
            leaderLogEndOffset
        } else {
            getReplicaOrException(replicaId).logEndOffsetMetadata
        }
    }
    
    // 计算新的 HW（ISR 中最小的 LEO）
    val newHighWatermark = allLogEndOffsets.min
    
    val oldHighWatermark = leaderLog.highWatermark
    if (newHighWatermark.messageOffset > oldHighWatermark) {
        leaderLog.updateHighWatermark(newHighWatermark)
        true
    } else {
        false
    }
}

classDiagram
    class Replica {
        +Int brokerId
        +TopicPartition topicPartition
        +Time time
        
        -Long _logEndOffset
        -Long _logStartOffset
        -Long lastCaughtUpTimeMs
        -Long lastFetchLeaderLogEndOffset
        -Long lastFetchTimeMs
        
        +updateFetchState()
        +resetLastCaughtUpTime()
    }

def updateFetchState(
    followerFetchOffsetMetadata: LogOffsetMetadata,
    followerStartOffset: Long,
    followerFetchTimeMs: Long,
    leaderEndOffset: Long
): Unit = {
    _logEndOffset = followerFetchOffsetMetadata.messageOffset
    _logStartOffset = followerStartOffset
    lastFetchTimeMs = followerFetchTimeMs
    lastFetchLeaderLogEndOffset = leaderEndOffset
    
    // 如果追上了 Leader，更新时间
    if (followerFetchOffsetMetadata.messageOffset >= leaderEndOffset) {
        lastCaughtUpTimeMs = followerFetchTimeMs
    }
}

classDiagram
    class HostedPartition {
        <<sealed trait>>
    }
    
    class Online {
        +Partition partition
    }
    
    class Offline {
    }
    
    class None {
    }
    
    HostedPartition <|-- Online
    HostedPartition <|-- Offline
    HostedPartition <|-- None

classDiagram
    class Request {
        +Int processor
        +RequestContext context
        +Long startTimeNanos
        +MemoryPool memoryPool
        +ByteBuffer buffer
        +RequestHeader header
        +RequestAndSize bodyAndSize
        
        +body~T~()
        +sizeOfBodyInBytes()
        +release()
    }
    
    class RequestContext {
        +RequestHeader header
        +String connectionId
        +InetAddress clientAddress
        +KafkaPrincipal principal
        +ListenerName listenerName
        +SecurityProtocol securityProtocol
        +ClientInformation clientInformation
    }
    
    class RequestHeader {
        +ApiKeys apiKey
        +short apiVersion
        +String clientId
        +int correlationId
    }
    
    Request --> RequestContext
    Request --> RequestHeader

classDiagram
    class DelayedOperation {
        <<abstract>>
        +Long delayMs
        +Option~Throwable~ error
        
        +tryComplete()
        +onExpiration()
        +onComplete()
        +forceComplete()
        +isCompleted()
    }
    
    class DelayedProduce {
        +Int requiredAcks
        +ReplicaManager replicaManager
        +Map~TopicPartition, PartitionResponse~ responseCallback
        
        +tryComplete()
        +onExpiration()
    }
    
    class DelayedFetch {
        +FetchParams fetchParams
        +Map~TopicPartition, FetchRequest.PartitionData~ fetchPartitionStatus
        +ReplicaManager replicaManager
        
        +tryComplete()
        +onExpiration()
    }
    
    DelayedOperation <|-- DelayedProduce
    DelayedOperation <|-- DelayedFetch

abstract class DelayedOperation(
    delayMs: Long,
    lockOpt: Option[Lock] = None
) extends TimerTask with Logging {
    
    // 尝试完成操作
    def tryComplete(): Boolean
    
    // 超时时调用
    def onExpiration(): Unit
    
    // 完成时调用（无论是否超时）
    def onComplete(): Unit
    
    // 强制完成
    def forceComplete(): Boolean = {
        if (completed.compareAndSet(false, true)) {
            onComplete()
            true
        } else {
            false
        }
    }
}

// DelayedProduce 实现
class DelayedProduce(
    delayMs: Long,
    requiredAcks: Int,
    replicaManager: ReplicaManager,
    responseCallback: Map[TopicPartition, PartitionResponse] => Unit
) extends DelayedOperation(delayMs) {
    
    override def tryComplete(): Boolean = {
        // 检查所有分区是否满足条件
        val isSatisfied = partitionStatus.forall { case (tp, status) =>
            // 检查 HW 是否达到写入的 offset
            val partition = replicaManager.getPartition(tp)
            partition.exists(_.highWatermark >= status.requiredOffset)
        }
        
        if (isSatisfied) {
            forceComplete()
        } else {
            false
        }
    }
    
    override def onExpiration(): Unit = {
        // 超时：返回当前状态
        responseCallback(partitionStatus)
    }
    
    override def onComplete(): Unit = {
        // 完成：返回结果
        responseCallback(partitionStatus)
    }
}

classDiagram
    class FetchDataInfo {
        +LogOffsetMetadata fetchOffsetMetadata
        +MemoryRecords records
        +boolean firstEntryIncomplete
        +Option~AbortedTxn~ abortedTransactions
        
        +toSend(maxBytes: Int)
    }
    
    class LogOffsetMetadata {
        +long messageOffset
        +long segmentBaseOffset
        +int relativePositionInSegment
    }
    
    FetchDataInfo --> LogOffsetMetadata

def toSend(maxBytes: Int): FetchDataInfo = {
    if (records.sizeInBytes <= maxBytes) {
        this
    } else {
        // 截断到 maxBytes
        val truncated = records.downConvert(maxBytes)
        copy(records = truncated)
    }
}

sequenceDiagram
    autonumber
    participant Main as Kafka Main
    participant KRS as KafkaRaftServer
    participant SS as SharedServer
    participant CS as ControllerServer
    participant BS as BrokerServer
    participant RM as ReplicaManager
    participant LM as LogManager
    
    Main->>KRS: startup()
    
    Note over KRS: Phase 1: 加载配置
    KRS->>KRS: loadConfig()
    
    Note over KRS: Phase 2: 初始化 SharedServer
    KRS->>SS: startup()
    SS->>SS: 创建 MetadataCache
    SS->>SS: 创建 KRaftManager
    SS->>SS: 启动 SocketServer
    
    Note over KRS: Phase 3: 启动 Controller（如果配置）
    alt process.roles 包含 Controller
        KRS->>CS: startup()
        CS->>CS: 创建 QuorumController
        CS->>CS: 启动 Raft 复制
        CS-->>KRS: Controller 就绪
    end
    
    Note over KRS: Phase 4: 启动 Broker（如果配置）
    alt process.roles 包含 Broker
        KRS->>BS: startup()
        
        BS->>LM: LogManager.apply()
        LM->>LM: 加载所有日志目录
        LM->>LM: 恢复未关闭的日志
        LM-->>BS: LogManager 就绪
        
        BS->>RM: new ReplicaManager()
        RM->>RM: 初始化 allPartitions
        RM->>RM: 启动 ReplicaFetcherManager
        RM-->>BS: ReplicaManager 就绪
        
        BS->>BS: 启动 KafkaApis
        BS->>BS: 启动 RequestHandlerPool
        
        BS-->>KRS: Broker 就绪
    end
    
    Note over KRS: Phase 5: 注册 Broker
    KRS->>CS: registerBroker(brokerId, listeners)
    CS->>CS: 写入 BrokerRegistrationRecord
    CS-->>KRS: 注册成功
    
    Note over KRS: Phase 6: 启动完成
    KRS->>KRS: 打印 "Kafka Server started"

sequenceDiagram
    autonumber
    participant P as Producer
    participant NS as NetworkServer
    participant RH as RequestHandler
    participant KA as KafkaApis
    participant RM as ReplicaManager
    participant Part as Partition
    participant Log as UnifiedLog
    participant Seg as LogSegment
    participant FR as FileRecords
    
    P->>NS: ProduceRequest
    NS->>NS: 读取完整请求
    NS->>RH: 放入请求队列
    
    RH->>KA: handleProduceRequest()
    
    Note over KA: 1. 验证阶段
    KA->>KA: 验证 Topic 授权
    KA->>KA: 验证 Topic 存在
    KA->>KA: 验证分区有效
    
    Note over KA: 2. 准备阶段
    KA->>KA: 解析 ProduceRequest
    KA->>KA: 按分区组织数据
    
    Note over KA: 3. 追加阶段
    KA->>RM: appendRecords(entriesPerPartition)
    
    loop 每个分区
        RM->>Part: appendRecordsToLeader()
        
        alt 不是 Leader
            Part-->>RM: NotLeaderOrFollowerException
        else 是 Leader
            Note over Part: 检查 ISR 大小
            alt ISR < minISR && acks=-1
                Part-->>RM: NotEnoughReplicasException
            else ISR 足够
                Part->>Log: appendAsLeader()
                
                Log->>Log: 分析 RecordBatch
                Log->>Log: 验证 sequence number
                Log->>Log: 分配 offset
                
                Log->>Seg: append()
                Seg->>FR: append(MemoryRecords)
                FR->>FR: 写入磁盘
                FR-->>Seg: written bytes
                
                Seg->>Seg: 更新 OffsetIndex
                Seg->>Seg: 更新 TimeIndex
                Seg-->>Log: LogAppendInfo
                
                Log->>Log: 更新 LEO
                Log-->>Part: LogAppendInfo
                
                Part->>Part: maybeIncrementLeaderHW()
                Part-->>RM: LogAppendInfo
            end
        end
    end
    
    Note over RM: 4. 等待确认阶段
    alt acks = 0
        RM-->>KA: 立即返回
    else acks = 1
        RM-->>KA: Leader 写入完成
    else acks = -1
        RM->>RM: 创建 DelayedProduce
        RM->>RM: 等待 ISR 确认
        Note over RM: （见 DelayedProduce 流程）
    end
    
    Note over KA: 5. 响应阶段
    KA->>KA: 构建 ProduceResponse
    KA->>NS: sendResponse()
    NS->>P: ProduceResponse