Raft 알고리즘 in Kafka

배경

최근 스터디에서 raft 알고리즘에 대해 논의하는 시간이 있었다. 이 raft 알고리즘의 활용 사례가 궁금해서 찾아보던 중 kraft라는 걸 발견했다

kraft는 kafka 클러스터에서 메타데이터 관리를 주키퍼 앙상블에서 kafka 클러스터 자체로 옮기면서 생겨난 용어이다

kafka 클러스터에서 n개의 브로커가 메타데이터 관리를 위한 controller로 선택이 되고 이 중 1개는 leader, 나머지는 follower가 된다. 그리고 처음에 leader를 선출하거나, leader에 문제가 생겨서 새로운 leader를 선출해야 하는 경우 raft 알고리즘을 활용해서 새로운 leader를 선출한다. 이를 통해 일부 컨트롤러에 장애가 나도 kafka 클러스터의 메타데이터를 안정적으로 보관하고, 클러스터가 계속 동작 가능하게 한다

 

Raft 알고리즘이란?

kafka의 raft 알고리즘에 대해 알아보기 전 일반적인 raft 알고리즘의 절차에 대해 알아보겠다. 아래 내용은 wiki의 내용을 정리한 것이다

1. 클러스터의 모든 노드는 Follower, Candidate, Leader 중 하나의 상태를 가짐
2. 시스템이 시작되면 모든 노드는 Follower 상태임
3. 일정 시간 동안 리더의 heartbeat를 받지 못하면 Follower → Candidate로 전환하여 선거를 시작
   1. Follower마다 heartbeat를 기다리는 timeout이 다르기 때문에, 모든 Follower들이 동시에 Candidate로 전환되지는 않음
4. Candidate는 자신의 term을 증가시키고, 자신에게 투표함
5. 모든 다른 노드에 RequestVote 요청을 보냄
6. 과반수 이상의 노드에게 투표를 받으면 Leader로 승격
7. 누군가가 과반수 투표를 먼저 얻으면 다른 Candidate는 Follower로 돌아감
8. 동률이거나 실패 시, 타임아웃 후 다시 시도

 

Kraft 개념 알아보기

kraft는 kafka 자체에서 raft 알고리즘을 구현해서 controller quorum을 구성해서 관리하기 위한 기능으로, zookeeper 없이 메타데이터를 직접 복제하고 합의할 수 있게 만든 구조이다

kraft 기능은 kafka 2.8.0 버전에서 미리 보기 기능으로 처음 추가됐고, 3.3.0 버전에서 정식으로 도입됐다. zookeeper를 이용한 컨트롤러 관리 기능은 점차 deprecated 될 예정이라고 한다

 

핵심 구조:

1. Controller Quorum

• 여러 개의 브로커 중 일부가 controller 역할을 수행하도록 구성됨
• 이 controller 들은 Raft 알고리즘을 통해 합의(consensus)를 이룸
• leader controller가 모든 메타데이터 업데이트를 주도하고, 나머지 follower가 복제

2. Metadata Log

• 메타데이터는 Raft 기반 log에 append-only로 기록됨
• 이 log는 controller quorum 간에 복제되며, 모든 브로커는 이 로그를 읽어 메타데이터 상태를 동기화함
• Kafka의 기존 메타데이터 구조는 **상태 기반(stateful)**이었지만, Raft 도입으로 event sourcing 스타일로 변경됨

3. KRaft 브로커

• ZooKeeper 없이 독립적으로 동작 가능
• process.roles=broker,controller로 설정해 하나의 프로세스에서 역할 분리 가능

참고 링크

 

코드를 통해 살펴보는 Kafka의 Raft 알고리즘 구현

지금부터는 kafka에서 구현한 raft 알고리즘에 대해 살펴보겠다. kafka에서는 이를 kraft라고 부르고 있다

아래 코드는 kafka 3.9 버전의 코드를 가져왔다. 코드의 양이 방대하기 때문에 raft 알고리즘에서 핵심이라고 생각되는 부분만 가져와서 살펴보겠다. 일부 잘 못된 내용이 있을 수 있으며, kafka에서는 이런 방식으로 raft 알고리즘을 구현했구나~ 정도의 참고용으로 봐주면 되겠다

 

아래에서 사용하는 epoch이란 단어는 raft 알고리즘의 term과 비슷한 의미를 가진다

Leader/Follower에서 Candidate로 변하는 경우

먼저 기존 리더에 문제가 생겨서 Leader→ Candidate|Unattched로 전환 부분의 코드를 살펴보겠다. KafkaRaftClient의 pollResigned 메서드의 일부를 발췌했다

long endQuorumBackoffMs = maybeSendRequests(
            currentTimeMs,
            partitionState
                .lastVoterSet()
                .voterNodes(state.unackedVoters().stream(), channel.listenerName()),
            () -> buildEndQuorumEpochRequest(state)
        );
    
---------------------------------

else if (state.hasElectionTimeoutExpired(currentTimeMs)) {
            if (quorum.isVoter()) {
                transitionToCandidate(currentTimeMs);
            } else {
                // It is possible that the old leader is not a voter in the new voter set.
                // In that case increase the epoch and transition to unattached. The epoch needs
                // to be increased to avoid FETCH responses with the leader being this replica.
                transitionToUnattached(quorum.epoch() + 1);
            }

 

kafka 클러스터의 컨트롤러들은 주기적으로 polling 하며 자신의 상태를 체크한다. 그리고 자신의 상태에 따라 적절한 handler를 실행하는데, pollResigned 메서드는 kraft에서 Leader가 더 이상 Leader 역할을 수행할 수 없을 때 수행되는 메서드이다

위 코드에서 첫 번째 부분은 Leader가 더 이상 Leader 역할을 수행할 수 없을 때 다른 노드들에게 EndQuorumEpochRequest 요청을 보내는 코드이다. 이를 통해 새로운 Leader를 뽑기 위한 재선거를 유도한다

위 코드의 두 번째 부분은 Leader가 자신의 상태 변환을 하는 코드이다. 더 이상 Leader가 아닌 컨트롤러는 두 가지 상태로 변환이 가능하다

1. 자신이 아직 voter라면 상태를 Leader -> Candidate로 변환한다
2. 더 이상 voter가 아니라면 상태를 Leader -> Unattacehd로 변환한다

kraft에서 voter는 투표권을 가진 노드로 리더 선출과 로그 복제 합의 과정에 참여한다. [2]의 케이스는 더 이상 voter가 아니기 때문에 합의 과정에 곧바로 참여할 수 없고 그래서 Unattached로 상태가 변경되는 것이다

Leader로부터 EndQuorumEpochRequest 요청을 받은 Follower 컨트롤러들은 요청 데이터를 검증 후 Follwer/Unattached 상태로 변경한다. Unattached는 현재 컨트롤러가 가진 epoch보다 더 큰 epoch을 발견했으나 Leader를 명확히 알 수 없는 경우에 가지는 상태이다

 

이번엔 일정 시간 동안 리더의 heartbeat를 받지 못하면 Follower → Candidate로 전환하여 선거를 시작 부분의 코드를 살펴보겠다. KafkaRaftClient의 pollFlollowerAsVoter 메서드의 일부를 발췌했다

else if (state.hasFetchTimeoutExpired(currentTimeMs)) {
    logger.info("Become candidate due to fetch timeout");
    transitionToCandidate(currentTimeMs);
    backoffMs = 0;
}

컨트롤러의 현재 상태가 Follwer인데 Leader에게 전송한 fetch 요청에 대한 응답이 timeout 때까지 오지 않으면 현재 상태를 Candidate로 변경하는 코드이다

Kraft에서 fetch 요청이란, Follower가 메타데이터를 최신화하기 위해 Leader에 정보를 요청하는 것이다. Leader가 Follwer에게 최신 메타데이터를 전달해 주는 방식이 아닌, Follower가 직접 요청해서 가져가는 방식이다

 

위 코드에서 transitionToCandidate 메서드의 내부를 좀 더 살펴보겠다. 아래는 QuorumState의 transitionToCandidate 메서드 일부를 발췌한 것이다

epoch을 1개 증가시키고, Candidate로 전환된 상태를 durableTransitionTo 메서드를 통해 파일에 영구적으로 저장한다. 그리고 electionTimouMs만큼 다른 컨트롤러에서 투표 결과가 오기를 기다린다

int retries = isCandidate() ? candidateStateOrThrow().retries() + 1 : 1;
int newEpoch = epoch() + 1;
int electionTimeoutMs = randomElectionTimeoutMs();

durableTransitionTo(new CandidateState(
    time,
    localIdOrThrow(),
    localDirectoryId,
    newEpoch,
    partitionState.lastVoterSet(),
    state.highWatermark(),
    retries,
    electionTimeoutMs,
    logContext
));

 

Candidate로부터 투표 요청(VoteRequestData)을 받은 경우

지금부터 Candidate 컨트롤러로부터 투표 요청을 받은 경우 어떻게 처리하는지 알아보겠다. KafkaRaftClient의 handleVoteRequest 메서드의 일부를 발췌했다

---- 생략
 VoteRequestData.PartitionData partitionRequest =
            request.topics().get(0).partitions().get(0);


int lastEpoch = partitionRequest.lastOffsetEpoch();
long lastEpochEndOffset = partitionRequest.lastOffset();
        
Optional<Errors> errorOpt = validateVoterOnlyRequest(candidateId, candidateEpoch);
if (errorOpt.isPresent()) {
    return buildVoteResponse(
        requestMetadata.listenerName(),
        requestMetadata.apiVersion(),
        errorOpt.get(),
        false
    );
}

if (candidateEpoch > quorum.epoch()) {
    transitionToUnattached(candidateEpoch);
}

---- 생략

 OffsetAndEpoch lastEpochEndOffsetAndEpoch = new OffsetAndEpoch(lastEpochEndOffset, lastEpoch);
        ReplicaKey candidateKey = ReplicaKey.of(
            candidateId,
            partitionRequest.candidateDirectoryId()
        );
        boolean voteGranted = quorum.canGrantVote(
            candidateKey,
            lastEpochEndOffsetAndEpoch.compareTo(endOffset()) >= 0
        );

        if (voteGranted && quorum.isUnattachedNotVoted()) {
            transitionToUnattachedVoted(candidateKey, candidateEpoch);
        }

1. validateVoterOnlyRequest 메서드 내부에서는 Candidate 컨트롤러의 epoch이 자신의 epoch보다 큰지 확인한다. 왜냐하면 자신보다 epoch이 큰(더 최신의) Candidate에게만 투표할 수 있기 때문이다
2. Candidate epoch이 자신의 epoch보다 큰 경우에는 자신의 상태를 Unattached로 변경한다
3. Candidate의 마지막 epoch 및 로그의 offset과 자신의 epoch 및 offset을 비교한다. 위 코드에서 lastEpochEndOffsetAndEpoch.compareTo(endOffset()) 부분이 해당 내용이다. Candidate의 epoch 및 offset이 더 최신이면 투표를 할 수 있는 상태이며 quorum.canGrantVote 메서드의 결과로 true가 리턴된다
4. 투표할 수 있는 상태이고, 현재 투표를 한 적이 없다면 transitionToUnattachedVoted 메서드를 통해 투표 완료 상태로 수정한다. 투표 완료 상태로 수정이란 현재 epoch을 Candidate의 epoch로 수정해서 최신화해 준다는 의미이다

 

배운 점

1. raft 알고리즘에 대해 좀 더 친숙해질 수 있는 경험이었다
2. kafka에서 raft 알고리즘을 구현하기 위해 다양한 상태를 정의하고 사용했는데, 여러 상태 정의가 필요한 경우의 코드 스타일을 구경할 수 있었다
3. 처음부터 코드를 보며 이해하다 보니 시간이 오래 걸렸는데, 전체적인 구조를 정리한 문서가 있는지 먼저 찾아보는 것도 좋을 것 같다. 큰 구조만 알면 코드를 따라가는 건 어려운 일이 아닐 것 같다