안녕하세요, 이번에는 NestJS나 타입스크립트를 벗어나서 카프카 삽질을 좀 해보려 합니다.
이번 주제는 카프카가 메시지를 어떻게 받아서 저장까지 하는지, 그 흐름을 한 번 따라가보려고 하는데요.
카프카의 Design 문서를 보면 아래와 같은 주제가 있습니다.
카프카는 메시지를 저장하고 캐싱할 때 파일시스템에 크게 의존하고 있으며, 카프카 서버 자체에서 어떠한 캐싱도 하지 않는다는 내용인데요.
JVM 위에서 객체를 캐싱하는 경우(in-memory), 이에 대한 메모리 오버헤드가 데이터 크기의 2배가 될 때도 있으며, 가비지 컬렉션에 의해 발생하는 Stop-the-world 이슈 등 여러 문제가 발생할 수 있습니다. 여기에 더해, 굳이 OS에서 잘 관리하고 있는 페이지 캐시가 있음에도 한 번 더 캐싱할 필요가 없다는 것입니다.
음 그래요 알겠는데. 어떻게 JVM 위에서 바로 OS cache를 제어하는 거죠? 정말이라면 어떻게 했을까요?
이번에는 그게 궁금해서 시작한 삽질입니다.
한 번 시작해봅시다.
이번에 살펴본 코드는 4.1.1 버전을 기준으로 합니다.
어떤 흐름으로?
이번 글은 네트워크로부터 메시지를 받아서 디스크에 저장하기까지 전 과정을 다룹니다.
아시겠지만 카프카에서는 자바 외에도 스칼라 코드가 있는데요. 스칼라를 살펴보는 글은 아니기 때문에 문법에 관한 내용은 생략합니다. 저도 스칼라 문법에 대해서는 모르기도 하구요. 그래도 자바 해보셨다면 충분히 이해 가능한 범주 내에 있는 것 같습니다.
추가로, 카프카는 당연하겠지만 모니터링이나 디버깅을 위해 로그가 정말 여기저기 많이 심어져 있는데요. 이번 주제에 핵심적인 부분은 아니기 때문에 모두 제거하였습니다. 원본 코드는 각 코드 블럭 아래 글에 첨부되어 있는 링크를 타고 들어가시면 보실 수 있으니 참고 부탁드립니다.
네트워크에서 온 메시지 받기
카프카 클라이언트와 서버는 TCP로 메시지를 주고 받는데요. 이하에서는 카프카 브로커가 TCP 서버를 열고, 클라이언트와 연결한 후 메시지를 받아 요청 정보를 생성하여 큐에 넣어주기까지의 과정을 다룹니다.
Acceptor@start
def start(): Unit = synchronized {
try {
if (!shouldRun.get()) {
throw new ClosedChannelException()
}
if (serverChannel == null) {
serverChannel = openServerSocket(endPoint.host, endPoint.port, listenBacklogSize)
}
processors.foreach(_.start())
thread.start()
startedFuture.complete(null)
started.set(true)
} catch { /* ... */ }
}카프카 서버 소켓을 여는 곳입니다. 카프카 서버가 켜질 때 호출되는 곳으로, 여기서 시작하는 thread는 Acceptor 자기 자신입니다. 해당 클래스는 외부로부터 연결 요청을 받아 소켓을 생성하는 역할을 합니다.
Acceptor가 쓰레드라고 했고, 쓰레드가 시작되었으니 run이 호출되겠죠?
Acceptor@run
override def run(): Unit = {
serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)
try {
while (shouldRun.get()) {
try {
acceptNewConnections()
closeThrottledConnections()
}
catch { /* ... */ }
}
} finally {
closeAll()
}
}커넥션을 받는 곳인데요. 연결 받고 처리하고 연결 받고 처리하고 연결 받고 처리하고 무한히 도는 역할 밖에 안 합니다. 사실 요청 받는 곳은 어떤 서버던 크게 다르진 않은 것 같습니다.
개인적으로 처음 보는 부분은 첫 줄입니다.
serverChannel은 Java NIO의 ServerSocketChannel 객체입니다. ServerSocketChannel은 연결을 대기하고(listen) 있다가 클라이언트의 연결 요청을 받아 SocketChannel을 만들어주는 역할을 합니다.
여기서 해당 라인을 해석해보면, serverChannel 채널을 nioSelector에 등록하고 새 연결을 accept할 수 있는 상태가 되었을 때 이를 감지하도록 설정합니다.
메시지 처리에 관한 내용을 봐야 하니 acceptNewConnections로 가봅시다.
Acceptor@acceptNewConnections
private def acceptNewConnections(): Unit = {
val ready = nioSelector.select(500)
if (ready > 0) {
val keys = nioSelector.selectedKeys()
val iter = keys.iterator()
while (iter.hasNext && shouldRun.get()) {
try {
val key = iter.next
iter.remove()
if (key.isAcceptable) {
accept(key).foreach { socketChannel => // (1)
var retriesLeft = synchronized(processors.length)
var processor: Processor = null
do {
retriesLeft -= 1
processor = synchronized {
currentProcessorIndex = currentProcessorIndex % processors.length
processors(currentProcessorIndex)
}
currentProcessorIndex += 1
} while (!assignNewConnection(socketChannel, processor, retriesLeft == 0)) // (2), (3)
}
} else
throw new IllegalStateException("Unrecognized key state for acceptor thread.")
} catch { /* ... */ }
}
}
}앞서 nioSelector에서 accept 직전의 소켓을 여기서 받아옵니다.
먼저 nioSelector에 accept 직전의 소켓이 존재하는지 확인합니다. select 메서드를 호출할 때 넘겨준 500 값은 대기 시간이며, 500ms를 블로킹으로 대기합니다.
(1) - 소켓이 존재한다면, 해당 소켓들 전체에 대해 루프를 돌며 accept 합니다. accept(key).forEach가 되어 있어서 컬렉션에 대한 처리로 보일 수 있는데, accept 메서드가 Optional을 반환하기 때문에 존재할 때만 해당 콜백 내 코드를 실행한다고 보시면 됩니다.
(2) - 해당 소켓을 서버가 갖고 있는 모든 Processor들에 대해 할당을 시도합니다(assignNewConnection).
(3) - 만일 모든 Processor들에 대해 할당을 실패하는 경우, 즉 retriesLeft가 0인 경우, 세 번째 파라미터(=mayBlock)을 true로 넘겨 해당 소켓에 대해서는 처리 가능할 때까지 블로킹되도록 하고 프로세서에 할당합니다.
여기서 말하는 Processor는 TCP로 들어온 메시지를 처리하는 역할을 하는 별도 쓰레드입니다. 아래에서 살펴 볼 예정입니다.
다른 건 프로세서 할당 관련 내용이고, 다음 단계로 넘어가는 건 assignNewConnection으로 보이니 저기로 가봅시다.
Acceptor@assignNewConnection
private def assignNewConnection(socketChannel: SocketChannel, processor: Processor, mayBlock: Boolean): Boolean = {
if (processor.accept(socketChannel, mayBlock, blockedPercentMeter)) {
true
} else
false
}해당 프로세서가 소켓을 할당 받을 수 있는 상태인지 확인합니다. 내용은 크게 없으니 accept로 가봅시다.
Processor@accept
def accept(socketChannel: SocketChannel,
mayBlock: Boolean,
acceptorBlockedPercentMeter: com.yammer.metrics.core.Meter): Boolean = {
val accepted = {
if (newConnections.offer(socketChannel))
true
else if (mayBlock) {
val startNs = time.nanoseconds
newConnections.put(socketChannel)
true
} else
false
}
if (accepted)
wakeup()
accepted
}새 연결을 받을 수 있는 여유 공간이 큐(newConnections)에 있으면 true, 아니면 false를 반환합니다. 만약 mayBlock이 true인 경우 새 연결을 받을 수 없더라도 일단 큐에 넣어둡니다.
큐에 등록해두면 프로세서 쓰레드가 처리하게 됩니다. 얘도 쓰레드니까 run을 봅시다.
Processor@run
override def run(): Unit = {
try {
while (shouldRun.get()) {
try {
// setup any new connections that have been queued up
configureNewConnections()
// register any new responses for writing
processNewResponses()
poll()
processCompletedReceives()
processCompletedSends()
processDisconnected()
closeExcessConnections()
} catch { /* ... */ }
}
} finally { /* ... */ }
}카프카 클라이언트에서 전송하는 데이터는 TCP 스트림 위로 전달되기 때문에, 한 번의 read에서 카프카 요청 전체가 들어온다는 보장이 없습니다. Produce 요청 하나에 여러 레코드가 들어있을 수도 있고, 이 요청 바이트 스트림이 여러 TCP segment로 나뉘어 도착할 수도 있습니다.
즉, 서버가 받은 하나의 TCP segment만으로는 완성된 카프카 요청이 아닐 수 있다는 것입니다. 이 때문에 Processor는 받은 바이트들을 누적하다가 비로소 완성된 카프카 요청이 만들어지면 다음 계층으로 넘기게 됩니다.
정리하면, Processor는 여러 TCP 연결로부터 들어오는 바이트 스트림을 누적하고, 완성된 카프카 요청을 별도 요청 단위로 파싱하여 다음 단계로 넘겨주는 non-blocking 이벤트 루프 쓰레드라고 볼 수 있습니다.
그러한 맥락에서 완성된 요청을 다음 계층으로 넘기는 포인트는 processCompletedReceives에 있습니다.
Processor@processCompletedReceives
private def processCompletedReceives(): Unit = {
selector.completedReceives.forEach { receive =>
try {
openOrClosingChannel(receive.source) match {
case Some(channel) =>
val header = parseRequestHeader(apiVersionManager, receive.payload)
if (
header.apiKey == ApiKeys.SASL_HANDSHAKE
&& channel.maybeBeginServerReauthentication(receive, () => time.nanoseconds())
) { /* ... */ }
else {
val nowNanos = time.nanoseconds()
if (channel.serverAuthenticationSessionExpired(nowNanos)) {
/* ... */
} else {
val connectionId = receive.source
val context = new RequestContext(/* ... */)
val req = new RequestChannel.Request(/* ... */)
if (header.apiKey == ApiKeys.API_VERSIONS) { /* ... */ }
requestChannel.sendRequest(req)
selector.mute(connectionId)
}
}
case None =>
// This should never happen since completed receives are processed immediately after `poll()`
throw new IllegalStateException(s"Channel ${receive.source} removed from selector before processing completed receive")
}
} catch { /* ... */ }
}
selector.clearCompletedReceives()
}받은 메시지에 대한 컨텍스트를 넘기기 위해 RequestContext 객체를 생성하고, 실제로 내부에서 처리하기 위한 객체로 RequestChannel.Request 객체를 생성하고 요청 처리 큐(requestChannel)로 던집니다. 그런 다음 해당 채널에 대한 동시 처리를 방지하기 위해 read를 차단합니다. (mute)
이렇게 요청 객체를 만드는 것은 마치 http 처리를 위해 express가 Request 객체를 만드는 것과 동일한 맥락이라고도 볼 수 있습니다.
참고로 요청 처리 큐(requestChannel)는 SocketServer 내에서 하나의 객체로 관리합니다. 이때SocketServer는 BrokerServer와 ControllerServer가 하나만 생성하여 갖고 있는데요.
즉, requestChannel은 브로커 서버 하나 당 하나만 만들어지고 여러 Processor가 공유하는 요청 처리 큐입니다.
(참고) BrokerServer는 알겠는데, ControllerServer는 뭐에요?
KRaft Controller는 기존에 Zookeeper가 해주었던 관리 역할을 대체하기 위해 등장했습니다. ControllerServer는 이 Kraft Controller의 구현체입니다.
추가로 KRaft 영역을 Control Plane, Broker 영역을 Data Plane이라고 부르는데요. 앞서 살펴본 Acceptor의 구현체 이름이 동일한 맥락에서 Acceptor의 구현체 이름이 DataPlaneAcceptor입니다.
요청 처리 및 라우팅
이렇게 만들어진 요청 객체를 다음 단계에 맞게 가공하고, 적절한 핸들러에게 요청을 넘기는 단계입니다.
KafkaRequestHandler@run
def run(): Unit = {
threadRequestChannel.set(requestChannel)
while (!stopped) {
val startSelectTime = time.nanoseconds
val req = requestChannel.receiveRequest(300)
req match {
case RequestChannel.ShutdownRequest =>
/* ... */
return
case callback: RequestChannel.CallbackRequest =>
/* ... */
case request: RequestChannel.Request =>
try {
request.requestDequeueTimeNanos = endTime
threadCurrentRequest.set(request)
apis.handle(request, requestLocal)
} catch {
/* ... */
} finally {
/* ... */
}
case RequestChannel.WakeupRequest =>
/* ... */
case null => // continue
}
}
completeShutdown()
}갑자기 KafkaRequestHandler라는 클래스가 튀어나왔습니다.
바로 앞 부분에서 요청 처리 큐(requestChannel)에다가 요청 객체를 만들어서 넣어줬었죠? 여기서 해당 큐를 폴링하고, 불러온 요청 객체를 그 유형에 맞게 분기하여 적절한 메서드를 호출해줍니다.
KafkaRequestHandler도 마찬가지로 루프를 도는 쓰레드이고, 쓰레드 풀(KafkaRequestHandlerPool)에서 관리됩니다.
아까 processCompletedReceives 메서드 구현체에서 RequestChannel.Request 객체를 만들어줬으니 해당 분기를 탑니다. 그럼 handle 메서드가 호출되겠죠? apis 객체는 KafkaApis 클래스의 객체이니, 해당 메서드로 가봅시다.
KafkaApis@handle
override def handle(request: RequestChannel.Request, requestLocal: RequestLocal): Unit = {
def handleError(e: Throwable): Unit = {
error(s"Unexpected error handling request ${request.requestDesc(true)} " +
s"with context ${request.context}", e)
requestHelper.handleError(request, e)
}
try {
if (!apiVersionManager.isApiEnabled(request.header.apiKey, request.header.apiVersion)) {
// The socket server will reject APIs which are not exposed in this scope and close the connection
// before handing them to the request handler, so this path should not be exercised in practice
throw new IllegalStateException(s"API ${request.header.apiKey} with version ${request.header.apiVersion} is not enabled")
}
request.header.apiKey match {
case ApiKeys.PRODUCE => handleProduceRequest(request, requestLocal)
case ApiKeys.FETCH => handleFetchRequest(request)
case ApiKeys.LIST_OFFSETS => handleListOffsetRequest(request)
case ApiKeys.METADATA => handleTopicMetadataRequest(request)
case ApiKeys.OFFSET_COMMIT => handleOffsetCommitRequest(request, requestLocal).exceptionally(handleError)
case ApiKeys.OFFSET_FETCH => handleOffsetFetchRequest(request).exceptionally(handleError)
case ApiKeys.FIND_COORDINATOR => handleFindCoordinatorRequest(request)
case ApiKeys.JOIN_GROUP => handleJoinGroupRequest(request, requestLocal).exceptionally(handleError)
case ApiKeys.HEARTBEAT => handleHeartbeatRequest(request).exceptionally(handleError)
case ApiKeys.LEAVE_GROUP => handleLeaveGroupRequest(request).exceptionally(handleError)
/* 아무튼 뭐가 많음 */
}
} catch {
/* ... */
} finally {
/* ... */
}
}여기서 무슨 요청인지에 따라 분기되어서 알맞는 메서드를 호출합니다. switch 문으로 이루어진 분기가 상당히 큰데요. case 문만 79개입니다. 오프셋 커밋하는 곳도 보이고, 하트비트 보내는 것도 있고..
우리는 이 수많은 분기 중에서 메시지를 받아서 저장하는 것을 봐야 하니 ApiKeys.PRODUCE를 봐야 합니다. 그럼 handleProduceRequest겠네요.
(참고) Api Key 관련
KafkaApis@handle 구현을 살펴보면 Api Key라는 표현이 있습니다. 흔히 우리가 생각하는 Api Key는 요청자에 대한 인증(Authentication)을 위해 헤더에 같이 붙여주는 값이지만, 카프카에서의 Api Key는 다릅니다.
이름 그대로 API에 대한 key 값인 것인데요. handle 메서드에 있는 switch-case문은 결국 매핑 작업이기 때문에, 다르게 생각해보면 Api Key를 key로, 대응되는 메서드를 value로 하는 Map이라고도 볼 수 있습니다.
그러한 맥락에서의 Key로 보고, Api Key라고 이름 지어진 것으로 개인적으로는 추정하고 있습니다.
Api Key에 대한 명세는 카프카의 공식 문서 중 프로토콜 문서에서 확인할 수 있습니다. Api Key는 클라이언트에서 메시지를 보낼 때 결정하며, 메시지의 헤더에 해당 값이 위치하게 됩니다.
Request Header v1 => { request_api_key request_api_version correlation_id client_id }
request_api_key => INT16
request_api_version => INT16
correlation_id => INT32
client_id => NULLABLE_STRINGRequest Header v2 => { request_api_key request_api_version correlation_id client_id }
request_api_key => INT16
request_api_version => INT16
correlation_id => INT32
client_id => NULLABLE_STRING기본적으로 카프카는 TCP로 소통하기 때문에 바이트 스트림으로 값이 들어옵니다. 따라서 헤더 부분을 읽을 때 순서대로 읽어 헤더를 파싱해주어야 하는데요.
public static RequestHeader parse(ByteBuffer buffer) {
short apiKeyId = -1;
try {
int bufferStartPositionForHeader = buffer.position();
apiKeyId = buffer.getShort();
short apiVersion = buffer.getShort();
ApiKeys apiKey = ApiKeys.forId(apiKeyId);
/* ... */
final RequestHeader header = new RequestHeader(headerData, headerVersion)
header.size = Math.max(buffer.position() - bufferStartPositionForHeader, 0);
return header;
} catch { /* ... */ }
}v1과 v2에서 모두 api key가 가장 처음에 있는 INT16으로 정의되어 있기 때문에, 헤더를 파싱하는 코드를 살펴보면 Api Key 값을 getShort()로 가져오게 됩니다.
KafkaApis@handleProduceRequest
def handleProduceRequest(request: RequestChannel.Request, requestLocal: RequestLocal): Unit = {
// 해당 메서드로 들어왔다는 것은 `PRODUCE` 요청이라는 것이므로
// `ProduceRequest`로 타입 캐스팅합니다.
val produceRequest = request.body[ProduceRequest]
if (RequestUtils.hasTransactionalRecords(produceRequest)) { /* ... */ }
// 에러 발생 시 이에 대한 응답을 위해 (토픽/파티션 - 응답) 간 Map을 정의합니다.
val unauthorizedTopicResponses = mutable.Map[TopicIdPartition, PartitionResponse]()
val nonExistingTopicResponses = mutable.Map[TopicIdPartition, PartitionResponse]()
val invalidRequestResponses = mutable.Map[TopicIdPartition, PartitionResponse]()
// 문제가 없어 다음으로 넘길 수 있는 (토픽/파티션 - 해당 파티션에 대한 레코드들) 간
// Map을 정의합니다.
val authorizedRequestInfo = mutable.Map[TopicIdPartition, MemoryRecords]()
// 일단 실제로 존재하는 토픽, 파티션에 대한 데이터인 것들을 잠시 넣어두는 곳입니다.
val topicIdToPartitionData = new mutable.ArrayBuffer[(TopicIdPartition, ProduceRequestData.PartitionProduceData)]
produceRequest.data.topicData.forEach { topic =>
topic.partitionData.forEach { partition =>
// 토픽의 이름과 ID를 식별합니다.
val (topicName, topicId) = if (topic.topicId().equals(Uuid.ZERO_UUID)) {
(topic.name(), metadataCache.getTopicId(topic.name()))
} else {
(metadataCache.getTopicName(topic.topicId).orElse(topic.name), topic.topicId())
}
// 토픽의 존재 여부를 식별합니다.
// 존재하지 않는 토픽이라면 응답과 함께 `nonExistingTopicResponses`에 추가합니다.
// 존재하는 토픽이라면 `topicIdToPartitionData`에 추가합니다.
val topicPartition = new TopicPartition(topicName, partition.index())
if (topicName.isEmpty && request.header.apiVersion > 12)
nonExistingTopicResponses += new TopicIdPartition(topicId, topicPartition) -> new PartitionResponse(Errors.UNKNOWN_TOPIC_ID)
else
topicIdToPartitionData += new TopicIdPartition(topicId, topicPartition) -> partition
}
}
// cache the result to avoid redundant authorization calls
val authorizedTopics = authHelper.filterByAuthorized(request.context, WRITE, TOPIC, topicIdToPartitionData)(_._1.topic)
topicIdToPartitionData.foreach { case (topicIdPartition, partition) =>
val memoryRecords = partition.records.asInstanceOf[MemoryRecords]
if (!authorizedTopics.contains(topicIdPartition.topic))
unauthorizedTopicResponses += topicIdPartition -> new PartitionResponse(Errors.TOPIC_AUTHORIZATION_FAILED)
else if (!metadataCache.contains(topicIdPartition.topicPartition))
nonExistingTopicResponses += topicIdPartition -> new PartitionResponse(Errors.UNKNOWN_TOPIC_OR_PARTITION)
else
try {
ProduceRequest.validateRecords(request.header.apiVersion, memoryRecords)
// 이것저것 검증 후에, 문제가 없는 토픽파티션 - 레코드 쌍만
// `authorizedRequestInfo`에다가 넣습니다.
authorizedRequestInfo += (topicIdPartition -> memoryRecords)
} catch {
case e: ApiException =>
invalidRequestResponses += topicIdPartition -> new PartitionResponse(Errors.forException(e))
}
}
@nowarn("cat=deprecation")
def sendResponseCallback(responseStatus: Map[TopicIdPartition, PartitionResponse]): Unit = {
/* ... */
}
def processingStatsCallback(processingStats: ProduceResponseStats): Unit = {
/* ... */
}
if (authorizedRequestInfo.isEmpty)
sendResponseCallback(Map.empty)
else {
val internalTopicsAllowed = request.header.clientId == "__admin_client"
val transactionSupportedOperation = AddPartitionsToTxnManager.produceRequestVersionToTransactionSupportedOperation(request.header.apiVersion())
// 이렇게 만들어진 `authorizedRequestInfo`는 `handleProduceAppend`의
// `entriesPerPartition` 파라미터로 넘어가게 됩니다.
replicaManager.handleProduceAppend(
timeout = produceRequest.timeout.toLong,
requiredAcks = produceRequest.acks,
internalTopicsAllowed = internalTopicsAllowed,
transactionalId = produceRequest.transactionalId,
entriesPerPartition = authorizedRequestInfo,
responseCallback = sendResponseCallback,
recordValidationStatsCallback = processingStatsCallback,
requestLocal = requestLocal,
transactionSupportedOperation = transactionSupportedOperation)
produceRequest.clearPartitionRecords()
}
}들어온 요청에 대해,
1. 각 데이터별로 토픽과 파티션을 식별하고
2. 각 파티션 별 레코드를 MemoryRecords 형태로 정의하여
3. handleProduceAppend로 넘깁니다.
구현체 코드가 상당히 길기 때문에 읽기 어려울 수 있는데, 크게는 위의 순서대로 처리한다고 보면 될 것 같습니다. 사실 하나하나 까보면 그리 복잡한 코드는 아니라서 주석과 함께 찬찬히 읽어보셔도 좋을 것 같습니다.
handleProduceAppend로 넘긴 요청 정보는
1. ReplicaManager@handleProduceAppend (#)
2. ReplicaManager@appendRecords (#)
3. ReplicaManager@appendRecordsToLeader (#)
4. ReplicaManager@appendToLocalLog (#)
호출 체인을 타고 내려갑니다.
ReplicaManager@appendToLocalLog
private def appendToLocalLog(internalTopicsAllowed: Boolean,
origin: AppendOrigin,
entriesPerPartition: Map[TopicIdPartition, MemoryRecords],
requiredAcks: Short,
requestLocal: RequestLocal,
verificationGuards: Map[TopicPartition, VerificationGuard]):
Map[TopicIdPartition, LogAppendResult] = {
// ...
entriesPerPartition.map { case (topicIdPartition, records) =>
if (Topic.isInternal(topicIdPartition.topic) && !internalTopicsAllowed) {
/* ... */
} else {
try {
val partition = getPartitionOrException(topicIdPartition)
val info = partition.appendRecordsToLeader(records, origin, requiredAcks, requestLocal,
verificationGuards.getOrElse(topicIdPartition.topicPartition(), VerificationGuard.SENTINEL))
val numAppendedMessages = info.numMessages
(topicIdPartition, LogAppendResult(info, exception = None, hasCustomErrorMessage = false))
} catch { /* ... */ }
}
}
}여기서는 아까 저 멀리서 온 entriesPerPartition(authorizedRequestInfo)에서 각 파티션별로 반복하면서 partition.appendRecordsToLeader를 호출합니다.
Partition@appendRecordsToLeader
def appendRecordsToLeader(
records: MemoryRecords,
origin: AppendOrigin,
requiredAcks: Int,
requestLocal: RequestLocal,
verificationGuard: VerificationGuard = VerificationGuard.SENTINEL
): LogAppendInfo = {
val (info, leaderHWIncremented) = inReadLock(leaderIsrUpdateLock) {
leaderLogIfLocal match {
case Some(leaderLog) =>
val minIsr = effectiveMinIsr(leaderLog)
val inSyncSize = partitionState.isr.size
// Avoid writing to leader if there are not enough insync replicas to make it safe
if (inSyncSize < minIsr && requiredAcks == -1) {
throw new NotEnoughReplicasException(s"The size of the current ISR : $inSyncSize " +
s"is insufficient to satisfy the min.isr requirement of $minIsr for partition $topicPartition, " +
s"live replica(s) broker.id are : $inSyncReplicaIds")
}
val info = leaderLog.appendAsLeader(records, this.leaderEpoch, origin, requestLocal, verificationGuard)
(info, maybeIncrementLeaderHW(leaderLog))
case None =>
throw new NotLeaderOrFollowerException("Leader not local for partition %s on broker %d"
.format(topicPartition, localBrokerId))
}
}
info.copy(if (leaderHWIncremented) LeaderHwChange.INCREASED else LeaderHwChange.SAME)
}현재 브로커가 해당 파티션에 대해 leader가 맞는지 확인하고, 리더가 맞다면 현재 ISR이 충족이 되는지를 확인합니다. leaderLogIfLocal은 해당 브로커 서버가 leader일 때만 leaderLog를 반환하기 때문에, 이를 통해서 leader인지 확인이 가능합니다.
해당 브로커가 leader라면, ISR을 확인한 후 문제가 있다면 예외를 던지고 아니라면 로그를 남기게 되는데요. 여기서 카프카가 신뢰성을 어떻게 달성했는지 확인할 수 있습니다.
ISR은 In-Sync Replicas의 약자입니다. 그대로 해석하면 ISR은 leader와 sync 상태에 있는 Replica인데요. 카프카에서는 'Sync 상태'에 있다는 것은, leader가 메시지를 기록한 후 주어진 시간[1] 내에 follower가 메시지를 기록한다는 것을 말합니다. 뒤집으면, 주어진 시간 내에 follower가 메시지를 기록하지 못한다면 leader는 해당 브로커를 ISR 목록에서 빼게 됩니다(참고).
다시 말해, In-Sync Replica는 leader의 상황을 잘 동기화하고 있는 Replica라고 볼 수 있습니다.
다시 코드로 돌아가면, 현재 ISR 수가 최소 ISR 수[2]보다 적고, requiredAcks가 -1(all)인 경우에는 메시지를 기록하지 않고 예외를 던집니다. 이는 leader가 ISR의 모든 ACK를 기다린다는 뜻인데요[3]. 즉, 최소 N개 이상의 모든 ISR이 정상적으로 복제가 되었는지를 확인할 때까지 leader가 대기합니다.
이는 다르게 말하면 최소 N개는 복제가 되어야 안심을 할 수 있다는 뜻이 되기도 하는데요. 따라서 설정된 최소 ISR 수보다 적은 수가 남아있는 상황에서 전체 ACK를 기다리는 설정을 하는 경우, 애초에 위 조건을 달성할 수 없기 때문에 예외를 던지게 됩니다.
모든 조건이 문제가 없다면 leaderLog.appendAsLeader를 호출합니다.
참고:
[1] 해당 시간은 replica.lag.time.max.ms Config로 제어할 수 있습니다.
[2] 최소 ISR 수는 min.insync.replicas로 제어할 수 있습니다.
[3] 해당 내용은 길어서 별도로 뺐습니다. 바로 아래를 참고해주세요.
Leader가 Follower가 따라왔음을 확인하는 방법
본문에서는 acks라는 이름 때문에 Leader가 ISR의 모든 ACK를 기다린다고 표현하였는데, 그렇다고 Leader가 Follower에게 sync 요청을 보내는 건 아니고, Follower도 offset을 따라간 뒤에 복제 완료 ACK를 보내는 것도 아닙니다.
Follower는 Leader에게 FetchRequest을 반복적으로 보내는데, 이 요청에 자신이 필요한 offset 값(fetch_offset)을 함께 보내줍니다. 이는 다르게 말하면 해당 Follower는 fetch 요청에 포함된 offset 직전의 메시지는 모두 갖고 있다는 뜻으로 해석할 수도 있습니다. 이를 통해 Leader는 Follower가 어디까지 따라왔는지 파악할 수 있습니다. 이러한 Follower offset 정보는 모두 Leader에 저장됩니다.
또한, 각 브로커 서버는 자신이 Leader인 파티션에 대해 HighWatermark(HW) 값을 갖고 있습니다. HighWatermark는 단순하게 말하면 ISR 전체가 공통으로 가지고 있다고 볼 수 있는 가장 큰 offset이며, 다르게 말하면 Leader가 안전하게 commit된 것으로 간주할 수 있는 offset 입니다.
위 정보를 조합하면, Leader는 각 ISR의 offset 정보를 알고 있고, 이를 기반으로 HighWatermark를 조정합니다. 즉, HighWatermark 값은 모든 ISR이 최소 해당 값까지는 갖고 있다는 것을 나타냅니다.
이것이 바로 'Leader가 ISR의 모든 ACK를 기다린다'의 구현체인 것입니다.
이렇게 대기하다가 HighWatermark 값이 이번 Produce의 마지막 offset 이상으로 올라가면, Leader 브로커 서버는 비로소 Producer(client)에게 응답을 보내주게 됩니다.
여기서 또 발생할 수 있는 문제가 있는데요. Leader에 메시지를 기록하기 전에 최소 ISR을 확인하였지만, 실제로 ISR들이 offset을 따라가는 시점은 그 이후입니다. 최소 ISR을 확인할 때는 문제가 없었지만 이후에 Replica에 문제가 생겨 HighWatermark가 필요한 값까지 달성하지 못할 수 있습니다.
이 경우 주어진 timeout 시간만큼 대기하다가 실패 처리 되는데, 일반적인 timeout 오류와 마찬가지로Producer(client) 입장에서는 해당 메시지들이 실제로 기록이 되었을지 안 되었을지 알 수 없고, 최악의 경우에는 해당 메시지가 소실될 수도 있습니다.
이제 진짜 기록하자!
이제 다 확인했으니 Leader에 로그를 기록하는 곳으로 가봅시다.
UnifiedLog@appendAsLeader
public LogAppendInfo appendAsLeader(MemoryRecords records,
int leaderEpoch,
AppendOrigin origin,
RequestLocal requestLocal,
VerificationGuard verificationGuard) {
boolean validateAndAssignOffsets = origin != AppendOrigin.RAFT_LEADER;
return append(records, origin, validateAndAssignOffsets, leaderEpoch, Optional.of(requestLocal), verificationGuard, false, RecordBatch.CURRENT_MAGIC_VALUE);
}스칼라 코드만 보다가 오랜만에 보는 자바 코드입니다. 아무튼 append로 다시 던집니다.
UnifiedLog@append
private LogAppendInfo append(
MemoryRecords records,
AppendOrigin origin,
boolean validateAndAssignOffsets,
int leaderEpoch,
Optional<RequestLocal> requestLocal,
VerificationGuard verificationGuard,
boolean ignoreRecordSize,
byte toMagic
) {
LogAppendInfo appendInfo = analyzeAndValidateRecords(records, origin, ignoreRecordSize, !validateAndAssignOffsets, leaderEpoch);
if (appendInfo.validBytes() <= 0) {
return appendInfo;
} else {
final MemoryRecords trimmedRecords = trimInvalidBytes(records, appendInfo);
synchronized (lock) {
return maybeHandleIOException(
() -> "Error while appending records to " + topicPartition() + " in dir " + dir().getParent(),
() -> {
MemoryRecords validRecords = trimmedRecords;
localLog.checkIfMemoryMappedBufferClosed();
if (validateAndAssignOffsets) {
// Leader log의 마지막 offset을 가져와서, 이번에 기록할 메시지들의
// 첫 번째 offset으로 등록합니다.
PrimitiveRef.LongRef offset = PrimitiveRef.ofLong(localLog.logEndOffset());
appendInfo.setFirstOffset(offset.value);
// 유효성 확인 이것저것 하고, 메시지들에 오프셋을 부여합니다.
// 이때 `offset` 값 또한 증가하게 됩니다.
LogValidator validator = new LogValidator(validRecords, /* ... */);
LogValidator.ValidationResult validateAndOffsetAssignResult = validator.validateMessagesAndAssignOffsets(offset, /* ... */);
validRecords = validateAndOffsetAssignResult.validatedRecords;
// 기록한 메시지들의 최대 timestamp 및 마지막 offset을 기록합니다.
// 위에서는 `offset` 값이 이번에 기록될 메시지들의 첫 오프셋을 가리키고 있었지만,
// `validateMessagesAndAssignOffsets` 메서드를 거치며 마지막 오프셋 + 1을
// 가리키게 되므로 `offset.value - 1`이 메시지들의 마지막 offset이 됩니다.
appendInfo.setMaxTimestamp(validateAndOffsetAssignResult.maxTimestampMs);
appendInfo.setLastOffset(offset.value - 1);
} else {
/* ... */
}
LogSegment segment = maybeRoll(validRecords.sizeInBytes(), appendInfo);
LogOffsetMetadata logOffsetMetadata = new LogOffsetMetadata(
appendInfo.firstOrLastOffsetOfFirstBatch(),
segment.baseOffset(),
segment.size());
AnalyzeAndValidateProducerStateResult result = analyzeAndValidateProducerState(
logOffsetMetadata, validRecords, origin, verificationGuard);
if (result.maybeDuplicate.isPresent()) {
/* ... */
} else {
// Leader의 log에 메시지를 기록합니다.
localLog.append(appendInfo.lastOffset(), validRecords);
updateHighWatermarkWithLogEndOffset();
// 슬슬 flush 할 때가 되면 flush 해줍니다.
if (localLog.unflushedMessages() >= config().flushInterval) flush(false);
}
return appendInfo;
}
);
}
}
}코드가 상당히 긴데, maybeHandleIOException의 두 번째 콜백 함수만 봐주시면 될 것 같습니다.
먼저, 이번에 기록되는 파티션의 마지막 오프셋을 가져와서 이번 append 내용의 시작 오프셋으로 기록합니다. 유효성 확인을 마친 후에, 현재 기록될 메시지의 마지막 오프셋을 기록합니다. 즉, 실제로 오프셋을 할당하는 과정입니다.
그런 다음 이것저것하고 localLog.append를 통해 실제로 기록을 진행합니다. 여기서 중요한 지점은, 기록은 진행하지만 실제로 디스크에 작성하는 것(flush)은 특정 조건 하에서만 한다는 것인데요.
기본적으로 카프카는 OS의 페이지 캐시를 신뢰하는 상황입니다. 즉 OS 페이지 캐시를 자신의 캐시로 사용합니다. 캐시가 있는 경우 바로 디스크에 flush하지 않듯이, 카프카도 마찬가지로 OS 페이지 캐시에만 기록해두고 실제로 물리 디스크에 flush하는 건 나중에 하도록 구현되어 있습니다.
그럼 카프카가 죽으면 데이터가 소실될 우려가 생길 수 있는데, 이를 위해 앞서 ISR 개수 확인을 했습니다. 즉, 브로커 하나가 죽어도 다른 Replica 또한 자신의 페이지 캐시에 기록하기 때문에, 굳이 비싼 fsync를 매번 할 필요가 없게 됩니다. 이것이 카프카가 브로커 서버 자체적으로 캐시를 구현하지 않고도 빠르고 신뢰할 수 있는 서비스를 구축한 근간입니다.
마지막으로 append랑 flush 보러 갑시다.
LocalLog@append
public void append(long lastOffset, MemoryRecords records) throws IOException {
segments.activeSegment().append(lastOffset, records);
updateLogEndOffset(lastOffset + 1);
}활성 상태의 세그먼트(=대상 로그 파일)에 append 합니다.
LogSegment@append
public void append(long largestOffset,
MemoryRecords records) throws IOException {
if (records.sizeInBytes() > 0) {
// ...
long appendedBytes = log.append(records);
// ...
}
}자신이 갖고 있는 로그(=파일)에 append 합니다.
FileRecords@append
public int append(MemoryRecords records) throws IOException {
// ...
int written = records.writeFullyTo(channel);
size.getAndAdd(written);
return written;
}메시지를 해당 파일이 갖고 있는 FileChannel에 기록합니다.
MemoryRecords@writeFullyTo
public int writeFullyTo(GatheringByteChannel channel) throws IOException {
buffer.mark();
int written = 0;
while (written < sizeInBytes())
written += channel.write(buffer);
buffer.reset();
return written;
}여기가 진짜진짜로 데이터를 쓰는 곳입니다.
FileChannel(GatheringByteChannel)에 데이터를 씁니다. FileChannel은 java의 NIO API로, 파일에 연결된 바이트 채널 구현체입니다. 여기서 write 하면 JNI를 통해 OS의 write / pwrite system call을 일으킵니다.
여기부턴 자바의 구현이기 때문에 OpenJDK 구현체를 살펴봅시다.
private static int writeFromNativeBuffer(FileDescriptor fd, ByteBuffer bb,
long position, boolean directIO,
boolean async, int alignment,
NativeDispatcher nd)
throws IOException
{
// ...
try {
if (position != -1) {
written = nd.pwrite(fd, bufferAddress(bb) + pos, rem, position);
} else {
written = nd.write(fd, bufferAddress(bb) + pos, rem);
}
} finally {
releaseScope(bb);
}
// ...
}write를 호출하면 이렇게저렇게 호출해서 writeFromNativeBuffer를 호출하게 되는데, 여기서 NativeDispatcher의 pwrite와 write를 호출합니다.
int write(FileDescriptor fd, long address, int len) throws IOException {
return write0(fd, address, len);
}
int pwrite(FileDescriptor fd, long address, int len, long position)
throws IOException
{
return pwrite0(fd, address, len, position);
}
static native int write0(FileDescriptor fd, long address, int len)
throws IOException;
static native int pwrite0(FileDescriptor fd, long address, int len,
long position) throws IOException;UnixFileDispatcherImpl.java#L64-L72, UnixFileDispatcherImpl.java#L187-L191
리눅스의 경우 UnixFileDispatcherImpl을 그대로 상속하고 있기 때문에(참고), 해당 클래스를 보면 여기서부터는 네이티브 구현체를 호출하고 있습니다.
JNIEXPORT jint JNICALL
Java_sun_nio_ch_UnixFileDispatcherImpl_write0(JNIEnv *env, jclass clazz,
jobject fdo, jlong address, jint len)
{
jint fd = fdval(env, fdo);
void *buf = (void *)jlong_to_ptr(address);
return convertReturnVal(env, write(fd, buf, len), JNI_FALSE);
}
JNIEXPORT jint JNICALL
Java_sun_nio_ch_UnixFileDispatcherImpl_pwrite0(JNIEnv *env, jclass clazz, jobject fdo,
jlong address, jint len, jlong offset)
{
jint fd = fdval(env, fdo);
void *buf = (void *)jlong_to_ptr(address);
return convertReturnVal(env, pwrite(fd, buf, len, offset), JNI_FALSE);
}네이티브 구현체를 살펴보면, 결국 write와 pwrite를 호출하는 모습을 볼 수 있습니다.
A successful return from write() does not make any guarantee that data has been committed to disk. On some filesystems, including NFS, it does not even guarantee that space has successfully been reserved for the data. In this case, some errors might be delayed until a future write(), fsync(2), or even close(2). The only way to be sure is to call fsync(2) after you are done writing all your data.
write의 리눅스 매뉴얼을 살펴보면, write 함수의 성공적인 반환(종료)은 데이터가 실제로 디스크에 기록되었다(commit)는 것을 보장하진 않으며, 보장을 할 유일한 방법은 fsync를 호출하는 것 뿐이라고 작성되어 있습니다.
즉, 메시지를 기록한다고 해서 실제로 디스크에 기록까지 보장되는 것은 아니며, 일반적으로는 OS 페이지 캐시에 기록해두었다 나중에 디스크로 flush된다고 볼 수 있습니다. 그럼 진짜 디스크에는 언제 쓸까요?
FileRecords@flush
public void flush() throws IOException {
channel.force(true);
}UnifiedLog@append 코드에서 flush를 호출하는 모습을 보았는데, 이때 FileChannel에 대해 강제로 flush합니다. 이 과정에서 OS 페이지 캐시에만 있던 내용이 실제 디스크에 기록됩니다(fsync).
마무리
지금까지 카프카 브로커 서버가 TCP 서버를 열고, 네트워크로부터 메시지를 받아 자신의 OS 페이지 캐시에 기록하거나 파일 시스템에 flush하기까지, 전체 과정을 실제 코드와 함께 살펴보았습니다.
내용을 요약해보자면, 크게 Acceptor, Processor, KafkaRequestHandler 쓰레드가 있습니다. 각 쓰레드의 역할을 정리해보면 다음과 같습니다.
Acceptor는 네트워크로부터 TCP 연결을 받아 소켓을 Processor의 큐(newConnections)에 할당하는 역할Processor는 자신에게 할당된 여러 소켓으로부터 바이트 스트림을 받아 누적하고, 온전한 카프카 요청이 만들어지면 요청 처리 큐(requestChannel)에 넣어주는 역할KafkaRequestHandler는 요청 처리 큐로부터 요청을 읽어 실제로 처리하는 역할
또한, 카프카 설계 문서의 Don't fear the filesystem! 섹션에서 확인했듯이 카프카는 로그 저장을 위해 브로커 내부에 별도의 어플리케이션 레밸 캐시를 두지 않고, OS의 페이지 캐시와 파일 시스템 동작을 적극적으로 활용합니다.
저번에 카프카 한 번 공부해보겠다고 냅다 카프카의 설계 문서를 읽었다가 저 Don't fear the filesystem! 한 문장에 꽂혀서 시작한 삽질이었는데, 어쩌다보니 카프카의 신뢰성 기반까지 확인하게 되었습니다.
다음엔 또 어디에 꽂혀서 어디를 삽질할지는 모르겠지만.
아무튼 읽어주셔서 감사합니다!
