[Kafka] Kafka Manager API를 사용하여 Kafka 클러스터 모니터링하는 법: 카프카 클러스터의 고가용성확보

카프카 클러스터에서 고가용성을 확보하기 위해서는

여러 대의 브로커 노드를 이용한 복제와 리더-팔로워 구조를 활용해야 한다.

이를 통해 카프카 클러스터 전체에서 데이터의 손실 없이 안정적으로 메시지를 처리할 수 있다.

 

먼저, 카프카 클러스터에서 데이터를 안정적으로 처리하기 위해서는

여러 대의 브로커 노드를 이용한 복제가 필요하다.

이를 위해 데이터를 복제할 브로커 노드를 미리 설정해야 하고

이때, 브로커 노드의 수는 홀수로 설정하는 것이 좋다.

이유는 브로커 노드 수가 짝수일 경우, 브로커 노드의 다수결로 리더 노드를 선출하는데,

만약 다수결을 가릴 수 없는 경우 브로커 노드 간에 충돌이 발생하여

데이터 손실이 발생할 수 있기 때문이다.

 

또한, 카프카 클러스터에서 데이터의 안정적인 처리를 위해서는

리더-팔로워 구조를 사용해야 합니다. 리더 노드는 쓰기 요청을 처리하고,

팔로워 노드는 리더 노드로부터 데이터를 동기화하여 읽기 요청을 처리한다.

 

이때, 리더 노드가 다운되면 자동으로 팔로워 노드 중 하나가 리더 노드로 승격된다.

이를 통해 데이터의 손실 없이 안정적으로 메시지를 처리할 수 있다.

그러나, 카프카 클러스터에서 고가용성을 확보하더라도, 여러 가지 문제가 발생할 수 있다.

 

예를 들어, 브로커 노드의 수가 부족하거나, 브로커 노드 간의 네트워크 연결이 불안정할 경우

데이터 전송 속도가 느려지거나 데이터 전송이 실패할 수 있다.

 

이러한 문제를 해결하기 위해서는 카프카 클러스터의 모니터링 및 튜닝을 진행해야한다.

클러스터의 고가용성을 확보하는 과정에서 발생할 수 있는 문제와 해결책에 대해 알아보자.

 

1. Broker 장애 시 복구

• 주키퍼에 등록된 브로커 정보를 활용하여 클러스터에 연결된 브로커들 중 정상 동작 중인 브로커를 찾아서 메시지 전송
• 예제 코드: kafka-topics.sh --zookeeper <zookeeper-server>:<port> --create --topic <topic> --replication-factor <replication-factor> --partitions <partition-count>
• 주의할 점: 클러스터에 복구된 브로커가 추가될 때 해당 브로커의 리더 파티션이 자동으로 재분배되는데, 이 때 메시지 전송이 중단될 수 있으므로 적절한 대응이 필요함
  
  2. Zookeeper 장애 시 복구

• 다른 Zookeeper 서버에서 클러스터 정보를 가져와서 정상 동작 중인 브로커에 전달하여 메시지 전송
• 예제 코드: kafka-topics.sh --bootstrap-server <bootstrap-server>:<port> --create --topic <topic> --replication-factor <replication-factor> --partitions <partition-count>
• 주의할 점: Zookeeper 서버가 다운되면 클러스터의 일부 브로커도 동작을 멈출 수 있으므로 이에 대한 대응이 필요함
•  
• 
  
  3. 파티션 리더 장애 시 복구

• ISR(In-Sync Replicas) 리스트에 등록된 복제본 중에서 리더 역할을 수행할 브로커를 선출하여 메시지 전송
• 예제 코드: kafka-topics.sh --bootstrap-server <bootstrap-server>:<port> --describe --topic <topic>
• 주의할 점: 리더 복제본의 브로커가 장애를 겪었을 경우 ISR 리스트에 등록된 다른 복제본 중에서 새로운 리더를 선출해야 하므로 장애 대응 프로세스를 적절히 설정해야함

 

 

 

 

 

먼저 주의해야 할 점으로는,

카프카 클러스터의 고가용성을 확보하기 위해서는 여러 대의 브로커를 사용해야한다.

하지만 브로커 수가 증가할수록 클러스터 구성이 복잡해지고 유지보수 비용이 증가하는 등의 단점이 있다.

따라서 브로커의 수를 적절히 조절해야 한다. 

 

카프카 클러스터의 상태를 확인하는 스크립트를 작성하는 것이 좋은데

예를 들어, 주키퍼와 브로커의 상태를 모니터링하고 이상이 발생하면 알림을 보내는 스크립트를 작성할 수 있다.

이렇게 작성한 스크립트를 정기적으로 실행하면 클러스터의 상태를 실시간으로 모니터링하여

이상이 발생하면 빠르게 대응할 수 있다.

 

Kafka Manager API가 정상적으로 동작하지 않을 때

다음은 주키퍼와 브로커의 상태를 모니터링하고 이상이 발생하면 알림을 보내는 스크립트의 자바 예제 코드이다. 

import java.io.IOException;

public class KafkaClusterStatusChecker {
    private static final String ZK_HOST = "localhost:2181";
    private static final String BROKER_LIST = "localhost:9092";

    public static void main(String[] args) throws IOException {
        // Check the status of ZooKeeper
        Process zkProcess = new ProcessBuilder("echo", "stat").start();
        Process ncZkProcess = new ProcessBuilder("nc", "-vz", ZK_HOST)
                .redirectInput(ProcessBuilder.Redirect.from(zkProcess.getInputStream()))
                .redirectError(ProcessBuilder.Redirect.INHERIT)
                .start();

        int zkResult = ncZkProcess.waitFor();
        if (zkResult == 0) {
            System.out.println("ZooKeeper is up and running");
        } else {
            System.out.println("ZooKeeper is down");
        }

        // Check the status of Kafka brokers
        Process brokerProcess = new ProcessBuilder("echo", "dump").start();
        Process ncBrokerProcess = new ProcessBuilder("nc", "-vz", BROKER_LIST)
                .redirectInput(ProcessBuilder.Redirect.from(brokerProcess.getInputStream()))
                .redirectError(ProcessBuilder.Redirect.INHERIT)
                .start();

        int brokerResult = ncBrokerProcess.waitFor();
        if (brokerResult == 0) {
            System.out.println("Kafka brokers are up and running");
        } else {
            System.out.println("Kafka brokers are down");
        }
    }
}

1. 먼저 ZooKeeper의 상태를 확인합니다. 이를 위해, 'echo stat' 명령어를 실행하고, 'nc' (netcat) 명령어를 사용하여 ZooKeeper 호스트의 포트에 연결-> 이 과정에서 'redirectInput' 및 'redirectError' 메소드를 사용하여 표준입력 및 에러를 처리
2. 'nc' 명령어는 성공적으로 호스트의 포트에 연결할 경우 0을 반환하고, 연결하지 못할 경우 다른 값을 반환한다. 'waitFor' 메소드를 사용하여 결과 값을 받고, 결과 값이 0일 경우 ZooKeeper가 정상적으로 실행 중임을 알 수 있다.
3. 마찬가지로, Kafka 브로커의 상태도 'echo dump' 명령어와 'nc' 명령어를 사용하여 확인한다.

이 방법은 Kafka Manager API가 정상적으로 동작하지 않을 때 유용하다...?
또한, 'nc' 명령어를 사용하여 포트에 연결하는 방법은 네트워크 포트의 상태를 확인할 때

자주 사용되는 방법 중 하나이다...

 

 

또한, 카프카 클러스터의 고가용성을 확보하기 위해서는 복제본 수를 적절히 설정해야한다.

복제본 수를 늘리면 데이터의 안전성은 높아지지만, 전송 지연이 발생할 수 있다.

따라서 데이터의 중요도와 전송 속도 등을 고려하여 복제본 수를 설정해야 한다.

 

 

Kafka Manager API가 사용해서 모니터링하는 법 

아래 코드에서는 Kafka Manager API를 사용하여 클러스터의 상태를 확인하고,

이상이 발생할 경우 Slack으로 알림을 보내도록 구현되어 있다.

이러한 모니터링 도구를 이용하여 카프카 클러스터의 상태를 지속적으로 모니터링하고,

문제가 발생할 경우 빠르게 대응함으로써 클러스터의 안정성을 유지할 수 있다.

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.http.HttpResponse.BodyHandlers;

import org.json.JSONObject;

public class KafkaManagerAPIClient {
private static final String KM_API_ENDPOINT = "";
private static final String SLACK_WEBHOOK_URL = "<https://hooks.slack.com/services/XXXXXXXXX/XXXXXXXXX/XXXXXXXXXXXXXXXXXXXXXXXX>";

public static void main(String[] args) {

    // Kafka Manager API call to get cluster status
    String clusterName = "";
    HttpRequest request = HttpRequest.newBuilder()
            .uri(URI.create(KM_API_ENDPOINT + "/clusters/" + clusterName + "/topics"))
            .build();

    HttpClient client = HttpClient.newHttpClient();
    HttpResponse response;

    try {
        response = client.send(request, BodyHandlers.ofString());
        if (response.statusCode() == 200) {
            // Cluster is up and running
            System.out.println("Kafka cluster is up and running");
        } else {
            // Cluster is down
            System.out.println("Kafka cluster is down");

            // Send notification to Slack
            JSONObject message = new JSONObject();
            message.put("text", "Kafka cluster is down. Status code: " + response.statusCode());

            HttpRequest slackRequest = HttpRequest.newBuilder()
                    .uri(URI.create(SLACK_WEBHOOK_URL))
                    .POST(HttpRequest.BodyPublishers.ofString(message.toString()))
                    .build();
            client.send(slackRequest, BodyHandlers.ofString());
        }
    } catch (Exception e) {
        e.printStackTrace();
    }
}

 

마지막으로, 카프카 클러스터의 고가용성을 확보하기 위해서는 데이터 백업과 복원에 대한 계획을 수립해야 한다.

예를 들어, 카프카 클러스터의 데이터를 주기적으로 백업하고, 데이터를 복원하는

테스트를 정기적으로 수행해야한다. 이를 통해 데이터 손실의 위험을 최소화하고, 안정적으로 메시지를 처리할 수 있다.