Kafka 브로커의 네트워크 레이어

                     ┌─────────────────────────────────────────┐
                     │            SocketServer                 │
                     │                                         │
  클라이언트 ───────►│  Acceptor (1개, Boss 역할)               │
                     │    │                                    │
                     │    ├──► Processor 0 ──► RequestChannel  │
                     │    ├──► Processor 1 ──► RequestChannel ─┼──► RequestHandler
                     │    └──► Processor 2 ──► RequestChannel  │    Pool (Worker)
                     │         (Worker 역할)                    │
                     └─────────────────────────────────────────┘

// Kafka Acceptor — 단순화된 구조
// ServerSocketChannel을 열고 새 연결을 받아들이는 역할
class Acceptor(val endPoint: EndPoint,
               val processors: Array[Processor]) extends Runnable {

  val serverChannel = ServerSocketChannel.open()
  serverChannel.bind(new InetSocketAddress(endPoint.host, endPoint.port))
  serverChannel.configureBlocking(false)

  // Selector에 ACCEPT 이벤트 등록
  val acceptSelector = Selector.open()
  serverChannel.register(acceptSelector, SelectionKey.OP_ACCEPT)

  override def run(): Unit = {
    while (isRunning) {
      // 새 연결 대기
      val ready = acceptSelector.select(500)
      if (ready > 0) {
        val keys = acceptSelector.selectedKeys().iterator()
        while (keys.hasNext) {
          val key = keys.next()
          keys.remove()

          if (key.isAcceptable) {
            accept(key)
          }
        }
      }
    }
  }

  // 새 연결을 라운드 로빈으로 Processor에 분배
  private def accept(key: SelectionKey): Unit = {
    val serverSocketChannel = key.channel().asInstanceOf[ServerSocketChannel]
    val socketChannel = serverSocketChannel.accept()
    socketChannel.configureBlocking(false)
    // 다음 Processor에 할당
    val processor = processors(currentProcessorIndex)
    processor.accept(socketChannel)
    currentProcessorIndex = (currentProcessorIndex + 1) % processors.length
  }
}

// Kafka Processor — 단순화된 구조
// 개별 연결의 I/O를 처리하고, 요청을 디코딩하며 응답을 인코딩하는 역할
class Processor(val id: Int,
                val requestChannel: RequestChannel) extends Runnable {

  // Processor마다 독립적인 Selector
  val selector = new KSelector(...)  // Kafka의 자체 Selector 래퍼

  // Acceptor로부터 새 연결을 받는 큐
  private val newConnections = new ConcurrentLinkedQueue[SocketChannel]()

  override def run(): Unit = {
    while (isRunning) {
      // 1. 새 연결 등록
      configureNewConnections()

      // 2. I/O 이벤트 처리
      selector.poll(300)

      // 3. 완성된 요청을 RequestChannel에 전달
      processCompletedReceives()

      // 4. 완성된 응답을 클라이언트에 전송
      processCompletedSends()
    }
  }

  // Acceptor가 넘긴 새 연결을 Selector에 등록
  private def configureNewConnections(): Unit = {
    while (!newConnections.isEmpty) {
      val channel = newConnections.poll()
      // OP_READ 등록 — 클라이언트가 보낸 데이터를 읽을 준비
      selector.register(channel)
    }
  }

  // 완성된 요청을 디코딩하여 RequestChannel로 전달
  private def processCompletedReceives(): Unit = {
    for (receive <- selector.completedReceives) {
      // Kafka 프로토콜에 따라 헤더 파싱
      val header = RequestHeader.parse(receive.payload)
      // RequestChannel에 요청 전달 → RequestHandler가 처리
      val request = new Request(processor = id, header, receive.payload)
      requestChannel.sendRequest(request)
    }
  }
}

// Netty Worker EventLoop — 동일한 패턴을 프레임워크가 처리
// 1. 새 연결 등록  → Boss가 channel.register(workerEventLoop)
// 2. I/O 폴링     → EventLoop.run()에서 selector.select()
// 3. 요청 디코딩  → Pipeline의 Decoder(ByteToMessageDecoder)
// 4. 응답 인코딩  → Pipeline의 Encoder(MessageToByteEncoder)

// Kafka RequestHandler — 단순화된 구조
// 비즈니스 로직을 처리하는 워커 스레드
class KafkaRequestHandler(val id: Int,
                          val requestChannel: RequestChannel,
                          val apis: KafkaApis) extends Runnable {

  override def run(): Unit = {
    while (isRunning) {
      // RequestChannel에서 요청 꺼내기 (블로킹)
      val request = requestChannel.receiveRequest(300)

      if (request != null) {
        // KafkaApis로 비즈니스 로직 처리
        apis.handle(request)
      }
    }
  }
}

// KafkaApis — API 키에 따라 적절한 핸들러로 라우팅
class KafkaApis {
  def handle(request: Request): Unit = {
    request.header.apiKey match {
      case ApiKeys.PRODUCE    => handleProduceRequest(request)
      case ApiKeys.FETCH      => handleFetchRequest(request)
      case ApiKeys.METADATA   => handleMetadataRequest(request)
      case ApiKeys.JOIN_GROUP => handleJoinGroupRequest(request)
      // ... 수십 가지 API
    }
  }
}

// Netty에서 무거운 작업을 별도 스레드 풀로 위임하는 패턴
EventExecutorGroup businessGroup = new DefaultEventExecutorGroup(16);

pipeline.addLast(businessGroup, "handler", new MyBusinessHandler());
// → EventLoop가 블로킹되지 않음

클라이언트 요청
    │
    ▼
┌─────────┐  accept()   ┌────────────┐  poll()   ┌──────────────┐
│ Acceptor │ ──────────► │ Processor  │ ────────► │ RequestChannel│
│ (Boss)   │  라운드로빈  │ (Worker)   │  디코딩   │ (공유 큐)     │
└─────────┘             └────────────┘           └──────┬───────┘
                                                        │
                                                        ▼
                                                 ┌──────────────┐
                                                 │RequestHandler │
                                                 │  + KafkaApis  │
                                                 │ (비즈니스로직) │
                                                 └──────┬───────┘
                                                        │
                                                        ▼
┌─────────┐             ┌────────────┐           ┌──────────────┐
│클라이언트│ ◄─────────── │ Processor  │ ◄──────── │ ResponseQueue│
│ (응답)  │   전송       │ (인코딩)   │  응답 전달 │              │
└─────────┘             └────────────┘           └──────────────┘

Kafka 프로토콜 메시지 구조:
┌──────────┬────────┬──────────┬──────────────┐
│ 길이(4B) │ API키  │ 버전     │ 페이로드      │
│          │ (2B)   │ (2B)     │ (가변)        │
└──────────┴────────┴──────────┴──────────────┘

// Kafka의 제로카피 — FileChannel.transferTo() 직접 사용
// 로그 세그먼트 파일을 네트워크로 전송할 때
public long writeTo(TransferableChannel channel, long position, int length) {
    // OS의 sendfile() 시스템콜로 매핑
    // 데이터가 유저 공간을 거치지 않고 커널에서 직접 소켓으로 전달
    return fileChannel.transferTo(position, length, channel);
}

[Reactor 패턴의 핵심 구조]

                  ┌─ Reactor (accept) ─── 연결 수락
                  │
이벤트 루프 ──────┼─ Reactor (I/O)    ─── 읽기/쓰기 처리
                  │
                  └─ Worker Pool      ─── 비즈니스 로직