0. Table Of Content

• 0. Table Of Content
• 1. Server Gateway Interface가 왜 필요한가?
• 2. WSGI (Web Server Gateway Interface)
• 3. ASGI (Asynchronous Server Gateway Interface)
  • 3.1. 기존 WSGI에는 어떤 문제점이 있었는가?
  • 3.2. ASGI는 어떤 방식으로 WSGI의 문제점을 해결했는가?
  • 3.3. 간단한 ASGI Application 예제
• 4. Reference

1. Server Gateway Interface가 왜 필요한가?

일반적으로 우리가 보고 있는 웹 서비스는 브라우저를 통해서 흘러나온 웹서버의 내용들이다.
대부분의 어플리케이션의 경우 웹과 소통하는 미들웨어를 가장 널리 사용되는 tomcat, apache를 채택하여 사용하고 있다. 아쉽게도 Tomcat, Apache는 Java기반으로 만들어졌기 때문에, Python기반의 프레임워크에서는 가장 널리 사용되는 웹 서버를 사용하기 위해서는 중간에서 Java기반 미들웨어가 말하는 것을 해석해 줄 또다른 미들웨어가 필요하게 된 것이다. 물론, 파이썬 기반의 미들웨어를 사용해도 괜찮지만, 이미 검증된 것을 포기할만큼 매력이 없거나 큰 리스크를 동반해야하기 때문에 Apache, Tomcat을 그대로 사용하고 이를 중간에서 번역해주는 python framework 전용 미들웨어를 하나 만들게 되었다. 그것이 바로 Server Gateway Interface이다. Python Server Gateway Interface의 경우, 널리 사용되는 것이 WSGI와 ASGI가 있는데, 상세 설명은 다음 챕터부터 진행 하겠다.

2. WSGI (Web Server Gateway Interface)

기존 python 기반 framework가 Java Middleware와 통신하기 위해서는 Medusa(Python으로 작성된 middleware), mod_python(embed Python), CGI / FastCGI(invoke Python via a gateway protocol) 같은 API를 사용해야 했었다. 그러나 위에서 명시된 API들은 특정 요소만을 고려해서 제작된 API기 때문에, 해당 API에 맞는 부분만을 개발자가 바라보게 했기 때문에, 개발자들이 선호하는 특정 영역에만 시야가 한정되었다.
그러나, 범용으로 쓰일 수 있는 WSGI의 등장으로 위의 문제점들이 사라지게 되었으며 WSGI는 PEP3333에 정식으로 채용(?)이 되었다.

3. ASGI (Asynchronous Server Gateway Interface)

그러나 WSGI도 시간이 지나면서 문제점이 발생하기 시작했다.

3.1. 기존 WSGI에는 어떤 문제점이 있었는가?

WSGI가 개발 중일 당시, WSGI는 오직 웹개발을 위한 공통 기반을 제공하는 프로토콜을 만드는 것이었다. 이 덕분에 파이썬 기반 웹개발자는 프레임워크 세부사항에 신경 쓰지 않고 여러 프레임워크에서 쉽게 작업을 할 수 있었다. 그러나, WebSocket 개념이 웹 개발자 사이에서 인기를 얻기 시작했을 때, WSGI는 single, synchoronous callable한 특성을 가지고 있었기 때문에, 다음과 같은 특성을 지니고 있어 webSocket과는 맞지 않았다.

• HTTP는 Connection이 짧게 유지되는 특성을 지니고 있었기 때문에, Long-Polling HTTP와 WebSocket 같이 상대적으로 connection이 긴 특성을 지닌 Protocol과는 맞지 않았다.
• HTTP Request는 application내부에서 오직 하나의 path를 가질 수 있기 때문에, 여러개의 path를 통해 이벤트를 수신하는 WebSocket의 이벤트를 처리할 수 없었다.

3.2. ASGI는 어떤 방식으로 WSGI의 문제점을 해결했는가?

ASGI의 구성요소와 책임은 다음과 같다.

• 소켓을 종료하고 이를 connection에 매핑하는 프로토콜 서버
• 포로토콜 서버 내부에서 실행되는 어플리케이션 연결을 인스턴스화(per 1 connection) 하며, 이벤트 메시지의 처리

WSGI와 비슷하게 ASGI도 기능이 비슷한 것처럼 보인다. 그러나, 다음요소에서 차이가 난다.

• Connection의 lifetime과 protocol을 정의하는 Connection Scope
• Application으로 보내지는 Connection 동안 일어날 사건에 대한 명세 Event

ASGI Application은 단일, 비동기 callable 속성을 지니고 있다. 수신 요청에 대한 정보를 포함하는 scope를 accept하고, 클라이언트에 이벤트를 보내고 받을 수 있는 awaitable를 보내고 받을 수 있다. 이 덕분에, ASGI Application은 WSGI의 한계점을 뛰어넘는 수신 / 발신 event를 허영할 수 있다. 그 뿐만아니라 ASGI Application은 background coroutine을 허용하기 때문에 application은 요청을 처리하면서 background에서 다른 작업도 수행할 수 있게 되었다. (ex. 외부 event를 listening 하고 있는 redis queue 등)

ASGI Application을 통해서 보내거나 받는 모든 event는 Python Dictionary Type이다. 이러한 사전 정의된 event의 format은 ASGI Application가 쉽게 다른 웹 서버에서 다른 웹서버로 쉽게 전환할 수 있게 한다.

3.3. 간단한 ASGI Application 예제

async def application(scope, receive, send):
    event = await receive()    
    ....     
    await send({"type": "websocket.send", ...}

4. Reference

• https://www.technoarchsoftwares.com/blog/asgi/
• https://channels.readthedocs.io/en/stable/asgi.html
• https://asgi.readthedocs.io/en/latest/

1. 서론 및 얽힌 이야기

내부 시스템에 대해 지속적으로 healthCheck를 하는 Spring Boot기반 서비스를 개발한 적이 있다. DB에 등록된 시스템 리스트를 가져와 순차적으로 healthCheck를 실행하는 방식의 서비스이다.

그러나, 10/12 HealthCheck Target별로 HealthCheck 시점의 정합성이 맞지 않는 문제가 발생하였다. 

문제 상황 서술 전에 적용중인 시스템 아키텍처 및 HealthChecker 서비스에 대한 개요를 먼저 풀어본다.

• A Group의 WAS에 대해 배포를 진행하였다.
• 배포가 진행되면, 해당 인스턴스로의 직접적인 호출에 대해서는 동작하지 않게 된다.
• static file을 제외한 모든 경로에 대해서 reverse proxy를 통하여 데이터가 was로 전달된다.
• healthCheck를 하는 경로 /api/healthcheck는 was에 설계되어 있기 때문에 was가 동작하지 않으면 web도 동작하지 않는다.
• WAS에 APP이 deploy되기 까지 약 25초의 시간이 소모되며 WEB이 배포되기 까지 5초의 시간이 소모된다.
• 최초의 healthCheck가 실패하면, 5초 간격으로 추가 5회의 재시도를 진행하며 모두 실패하여야만 최종 실패로 간주된다.

위 서술된 환경에서 발생한 문제를 시간 순서대로 다이어그램과 함께 아래에 서술하겠다.

위 같이 같은 시점에서 healthCheck를 진행했으면 같은 시점의 결과가 출력되어야 하나, 순차적으로 실행하는 도중 healthCheck Error로 인해 재시도를 함으로써 다음 healthCheck가 미뤄졌다.

그 결과, WAS와 WEB의 healthCheck의 시점의 차이가 커져 시점에 대한 일관성이 맞지 않는 문제가 발생하게 되었다.

이 문제를 해결하기 위해 delay없이 최대한 모든 target에 대해 비슷한 시점에 실행하여야 했기때문에, thread를 이용함으로써 동시다발적으로 최대한 같은 시점에서 target healthCheck를 진행해보기로 하였다.

그 중 @Async를 활용한 Thread를 이용하여 개선작업을 하기로 하고 문제가 되는 코드를 추려낸 결과, 다음과 같았다.

@Service
public class MonitoringSystemService {
 
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
 
    @Autowired
    private MessageRepository messageRepo;
 
    @Autowired
    private MonitoringSystemRepository monitoringRepo;
 
 
 
    /*
     * 스케줄러 method
     *
     * @parameter
     * @return
     *          Type : void
     * @cycle period : 모든 target의 healthCheck가 끝난 후, 15초 대기
     */
 
    @Scheduled(fixedDelay = 15000)
    @Transactional
    public void scheduledMonitoringService() throws Exception {
 
 
        try {
 
            this.setPid();
 
            List<MonitoringSystem> systemInfoList = monitoringRepo.getAllMonitoringSystemList();
 
 
            //시스템별 상태 체크 및 톡 발송 실행
            for( MonitoringSystem systemInfo : systemInfoList) {
 
 
 
                switch(systemInfo.getMonFlagCd()) {
 
                    //RealTime 호출 (엔드포인트 호출형)
                    case "R":
                            logger.info("모니터링 유형 : 엔드포인트 호출");
                            logger.info("타겟 엔드포인트 : " + systemInfo.getMonSysUrl());
                            logger.info("타겟 엔드포인트 포트 : " + systemInfo.getMonSysPort());
                            logger.info("HttpRequest Method : " + systemInfo.getReqMtd());
                            boolean httpRequestResult = HealthChecker.run(systemInfo.getMonSysUrl(), systemInfo.getReqMtd());
 
 
                            systemInfo = this.sendMessage(systemInfo, httpRequestResult);
                            this.modifySystemInfo(systemInfo, httpRequestResult);
 
                        break;
 
                    //Static 호출 (직접적인 쿼리 실행)
                    case "S":
                        logger.info("모니터링 유형 : 직접적인 쿼리");
                        logger.info("실행 쿼리 : " + systemInfo.getMonSysQury());
 
                        boolean scheduledSystemStatus = monitoringRepo.getScheduledSystemStatus(systemInfo.getMonSysQury());
                        logger.info("쿼리 업데이트 정상 여부 : " + String.valueOf(scheduledSystemStatus));
 
                        systemInfo = this.sendMessage(systemInfo, scheduledSystemStatus);
                        this.modifySystemInfo(systemInfo, scheduledSystemStatus);
                        break;
                }
 
                logger.info("=======================================================================================================");
            }
 
        }catch (Exception e) {
            logger.error(e.getMessage());
        }
 
        logger.info("#################################################################################################스케줄러 끝난 시간 : " + getNowTime());
 
 
    }
 
 
}

public class Http5xxErrorRetryHandler implements ServiceUnavailableRetryStrategy {
 
    private final Set<Integer> retryableErrorCodes = new HashSet<>(Arrays.asList(500, 503, 502));
 
    private static final Logger logger = LoggerFactory.getLogger(Http5xxErrorRetryHandler.class);
 
    @Override
    public boolean retryRequest(HttpResponse response, int executionCount, HttpContext context) {
 
 
        int statusCode = response.getStatusLine().getStatusCode();
 
 
 
        if (executionCount > 5) {
            logger.error("재시도 5회 초과로 에러처리 합니다.");
            return false;
        }
 
        if (retryableErrorCodes.contains(statusCode)) {
            logger.warn(statusCode + "오류 발생");
            logger.info("재시도 횟수 : " + executionCount);
            return true;
        }
 
        return false;
    }
 
    @Override
    public long getRetryInterval() {
        return 5000;
    }
 
}

위 2개의 코드로 인해 최초 health check가 실패한 경우, 다음 healthCheck 대상으로 넘어가기까지 최소 30초 이상의 시간이 소모되기 때문에 동시성이 보장되지 않고 있는 상황이다.

2. 개선코드 및 작업내용

순차적으로 healthCheck를 하는 것이 아닌 thread를 이용하여 동시다발적으로 처리할 수 있도록 하기 위해 가장 먼저 thread를 실행시킬 executor를 생성해주어야 한다.

나는 아래와 같이 AsyncConfigurer를 이용하여 세부설정이 완료된 executor 객체를 bean으로 만들어 어디서든 이용 할 수 있도록 하였으며, 세부 설정은 아래 코드를 참조한다.

@Configuration
@EnableAsync
public class AsyncThreadConfig implements AsyncConfigurer{
 
 
 
    // 기본 thread 개수
    private static int THREAD_CORE_POOL_SIZE = 5;
 
    // 최대 thread 개수
    private int THREAD_MAX_POOL_SIZE = 10;
 
    // Thread Queue 사이즈
    private static int THREAD_QUEUE_CAPACITY = 5;
 
    private static String THREAD_NAME = "healthCheckExecutor";
 
    @Resource(name = "healthCheckExecutor")
    private ThreadPoolTaskExecutor healthCheckExecutor;
 
    @Override
    @Bean(name = "healthCheckExecutor")
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //THREAD_MAX_POOL_SIZE = monitoringSystemRepository.getMonitoringSystemCount();
 
        executor.setCorePoolSize(THREAD_CORE_POOL_SIZE);
        executor.setMaxPoolSize(THREAD_MAX_POOL_SIZE);
        executor.setQueueCapacity(THREAD_QUEUE_CAPACITY);
        executor.setBeanName(THREAD_NAME);
        executor.initialize();
        return executor;
    }
 
 
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        // TODO Auto-generated method stub
        return AsyncConfigurer.super.getAsyncUncaughtExceptionHandler();
    }
 
 
    public boolean isThreadPoolAvailable(int createCnt) {
 
        boolean threadStatus = true;
 
        if ((healthCheckExecutor.getActiveCount() + createCnt) > (THREAD_MAX_POOL_SIZE + THREAD_QUEUE_CAPACITY)) {
            threadStatus = false;
        }
 
        return threadStatus;
    }
 
 
    public boolean isThreadPoolAvailable() {
 
        boolean threadStatus = true;
 
        if ((healthCheckExecutor.getActiveCount()) > (THREAD_MAX_POOL_SIZE + THREAD_QUEUE_CAPACITY)) {
            threadStatus = false;
        }
 
        return threadStatus;
    }
}

위와 같이 executor를 등록하였으면, 이제 executor에 넣을 비동기프로세스를 작성할 차례이다.

실질적인 처리 로직이 담긴 부분에서 비동기 처리할 메소드에 @Async  Annotation을 선언하여 비동기 스레드를 통해 처리되도록 하였다.

다음과 같이 비동기식으로 처리할 메소드에 @Async Annotation과 함께 앞서 선언한 Executor Bean 이름을 명시해주어야 한다.

@Service
public class ThreadExecutorService {
 
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
    @Autowired
    private MessageRepository messageRepo;
 
    @Autowired
    private MonitoringSystemRepository monitoringRepo;
 
 
    @Async("healthCheckExecutor")
    public void threadExecutor(MonitoringSystem systemInfo) {
        //시스템별 상태 체크 및 톡 발송 실행
        this.setPid();
 
        try {
            switch(systemInfo.getMonFlagCd()) {
 
                //RealTime 호출 (엔드포인트 호출형)
                case "R":
                        logger.info("모니터링 유형 : 엔드포인트 호출");
                        logger.info("타겟 엔드포인트 : " + systemInfo.getMonSysUrl());
                        logger.info("타겟 엔드포인트 포트 : " + systemInfo.getMonSysPort());
                        logger.info("HttpRequest Method : " + systemInfo.getReqMtd());
                        boolean httpRequestResult = HealthChecker.run(systemInfo.getMonSysUrl(), systemInfo.getReqMtd());
 
 
                        systemInfo = this.sendMessage(systemInfo, httpRequestResult);
                        this.modifySystemInfo(systemInfo, httpRequestResult);
 
                    break;
 
                //Static 호출 (직접적인 쿼리 실행)
                case "S":
                    logger.info("모니터링 유형 : 직접적인 쿼리");
                    logger.info("실행 쿼리 : " + systemInfo.getMonSysQury());
 
                    boolean scheduledSystemStatus = monitoringRepo.getScheduledSystemStatus(systemInfo.getMonSysQury());
                    logger.info("---------------------------------------시스템 정상 여부 : " + String.valueOf(scheduledSystemStatus));
 
                    systemInfo = this.sendMessage(systemInfo, scheduledSystemStatus);
                    this.modifySystemInfo(systemInfo, scheduledSystemStatus);
                    break;
            }
        }catch (Exception e) {
            logger.error(e.getMessage());
        }
            logger.info("=======================================================================================================");
    }
}

이제 필요할때 마다 비동기식으로 처리되도록 선언한 메소드를 호출해보자.

앞서 선언한 AsyncThreadConfig와 비동기 프로세스에 대해 작성한 ThreadExecutorService를 IOC Container에서 불러온다.

Bean을 불러온 뒤, 다음과 같이 원하는 비동기 메소드를 호출한다.

@Service
public class HealthCheckSchedulerService {
 
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
 
    @Autowired
    private MonitoringSystemRepository monitoringRepo;
 
    @Autowired
    private ThreadExecutorService threadExecutorService;
 
    @Autowired
    private AsyncThreadConfig asyncConfig;
 
 
    @Scheduled(fixedDelay = 15000)
    @Transactional
    public void scheduledMonitoringService() throws Exception {
 
 
        try {
            List<MonitoringSystem> systemInfoList = monitoringRepo.getAllMonitoringSystemList();
 
            for( MonitoringSystem systemInfo : systemInfoList) {
 
                 try {
                        // 등록 가능 여부 체크
                        if (asyncConfig.isThreadPoolAvailable()) {
                            // task 사용
                            threadExecutorService.threadExecutor(systemInfo);
                        } else {
                            logger.info("Thread 한도 초과");
                        }
                    } catch (TaskRejectedException e) {
                        logger.info(e.getLocalizedMessage());
                    }
 
            }
 
        }catch (Exception e) {
            logger.error(e.getMessage());
        }
 
 
    }
 
}

3. 개선 결과

먼저 개선전 서비스 로그를 보면 하나의 스레드를 이용하여 순차적으로 실행하기 때문에, 중간에 내부 서비스가 장애가 날시, 그 다음 서비스에 대해 healthCheck하는 것이 재시도 한만큼 지연되고있다.

그러나 개선된 다음에는 재시도 및 여러개의 요청이 각기 다른 스레드를 통해 진행되는 것을 확인할 수 있었으며 시간의 정합성을 조금 더 개선할 수 있었다.

Table of content

• 1. What is the Apache Kafka
• 2. Benefits of Apache Kafka
• 3. Apache Kafka 아키텍처
  • 3.1. 주체 단위 아키텍처
    • 3.1.1. Producer
    • 3.1.2. Broker
    • 3.1.3. Zookeeper
  • 3.2. 데이터 단위의 아키텍처
    • 3.2.1. Topic
    • 3.2.2. Partition
    • 3.2.3. Offset
• 4. Failover 과정
• 5. 가장 이상적인 Kafka 아키텍처

1. What is the Apache Kafka

Apache Kafka는 대량의 데이터를 처리할 수 있으며, 엔드포인트간 메시지를 전달 할 수 있는 분산 발행-구독 메시징 시스템이다. Kafka 메시지는 스토리지에 유지되고 데이터 복제를 통해 데이터 손실을 방지 할 수 있다. ZooKeeper라는 동기화 서비스 기반의 시스템이며, 실시간 데이터 스트리밍 분석 툴인 Apache Storm과 Spark와 통합되어 자주 사용된다.

2. Benefits of Apache Kafka

• 신뢰성 : 데이터의 분산, 분할, 복제를 통해 데이터의 신뢰성을 보장한다.

• 확장성 : topic의 발행만으로 down time 없이 쉽게 확장이 가능하다

• 내구성 : 분산된 commit log를 통해 클러스터간 동기화를 하며, failure시 이 로그를 통해 빠른 복구가 가능하다.

• 퍼포먼스 : TB단위의 메시지가 시스템에 저장되어도 안정된 퍼포먼스를 제공한다.

3. Apache Kafka 아키텍처

3.1. 주체 단위 아키텍처

3.1.1. Producer

• 하나 이상의 topic으로 메시지를 발행하여 broker로 전송한다.

• 각 메시지는 key, value, timestamp로 이루어져 있다.

• 새로운 메시지를 publish 할 때, 몇개의 파티션으로 나눌 것인지 결정한다. (이미 발행 된 메시지에 대해서 도 가능)

출처 : https://kafka-python.readthedocs.io/ (Kafka-Python API Document Homepage)

3.1.2. Broker

(한 VM 안에 있는 broker 전체집합을 카프카라고 한다.)

• producer에게서 받은 메시지를 세부 설정(파티션, 오프셋 정책…)에 따라서 저장하는 역할

• 저장된 메시지는 정책에 따라 관리되며, consumer의 메시지 pull요청에 의해서만 메시지를 전송할 수 있다.

• 클러스터 내부에 대표 브로커를 선정하여 controller의 역할을 부여한다.

• Controller는 클러스터 내부 모든 브로커에 대해 partition을 관리한다.

• 1.0버전 부터는 offset 정보를 broker에서 관리한다.

3.1.3. Zookeeper

• 클러스터 내부의 브로커 간 통신하기 위해서는 zookeeper를 거쳐야 한다.

• VM 내부의 브로커들이 구동되기 위해서는 zookeeper가 반드시 선행으로 실행되어야 한다.

• 다른 VM의 Zookeeper와 논리적으로 결합되어 있다.

• 외부에서 들어온 요청에 대해서 해당 Leader Partition이 존재하는 브로커로 연결시켜준다.

3.2. 데이터 단위의 아키텍처

3.2.1. Topic

• 메시지가 분류되는 키워드.

• 쉽게 말하면 메시지의 주제를 의미한다.

• 하나의 토픽은 여러개의 파티션으로 구성되어 있다.

3.2.2. Partition

• 토픽 발행 시, partitioning factor * replication factor의 수만큼 partition이 생성된다.

• Partition에는 2가지 유형이 있다.

  • Leader Partition : 한 토픽의 구성에 대해 외부와 인터페이스를 담당하는 Partition이다. 데이터 조회시, 이 반드시 leader Partition을 통해 데이터를 얻을 수 있으며, Partition Factor의 수만큼 Leader Partition이 생성된다.

  • Follower Partition : Leader Partition의 failure에 대비한 예비 leader Partition이다. Leader가 failure시, 이들 중 하나가 자동으로 Leader로 승급된다.

  • Follower는 Leader의 변경사항을 pull받는 방식으로 synchronize한다. 이러한 방식을 카프카에서는 ISR(In Sync Replica)라고 명명한다.

  • Follower는 Leader에 대해 pull 방식으로 동기화를 하며, 리더는 이 동기화 요청을 일정시간 받지 못하게 되면 ISR에서 해당 partition을 제외시키게 된다.

• Partitioning 한만큼, 데이터 병렬처리가 이루어지기 때문에 처리속도가 빨라진다.

3개의 공을 한명의 포수가 받는 시간과 3명이 나누어서 받을 때 걸리는 시간을 비교해보자.

3.3.3. Offset

• Kafka의 최소 데이터 단위

• 1 Message publish = 1 Offset

• Producer의 Partitioning Algorithm에 따라 데이터가 위치할 Partition이 결정된다.

• 오프셋별로 commit log가 남아 이를 이용하여 Fail Over를 수행한다.

• 별도의 카프카 데이터 보존 정책에 의해 데이터를 유지한다.

4. Failover 과정

FailOver가 어떻게 이루어지는지에 대해 아키텍처 기반으로 설명하겠다.

다음 그림은 2개의 토픽에 대해 “replication factor = 3, partition factor = 2”의 설정으로 메시지가 발행되어 정상적으로 Broker서버에 저장되어 있는 상태이다.

위 아키텍처에서 Broker 0 과 Broker 3에서 장애가 나면 아키텍처는 다음 그림과 같아진다. 이 때, Topic1의 P0 Partition과 Topic2의 P1 Partition의 Leader Partition이 Broker가 장애가 나면서 더이상 Partition에 대해 접근을 할 수 없게 되었다.

하지만, Kafka에는 다음과 같은 규칙이 있다.

“모든 Partition은 Leader Partition이 될 수 있다.“

위 규칙에 의해 Kafka 내부 알고리즘에 의해 기존에 존재하던 Follower Partition이 새로운 Leader로 뽑히게 되어 다음 그림과 같이 장애에 대해 자동으로 Failover가 수행된다.

5. 가장 이상적인 Kafka 아키텍처

지금까지의 내용을 기억한다면, 데이터에 대한 접근은 오직 “Leader Partition“에 대해서만 접근이 이루어진다는 것을 알 수 있다. 그러므로, Leader Partition이 클러스터 내부의 broker에 골고루 분포가 된다면 한 broker로 트래픽이 몰려 timeout이나 다른 에러가 나는 상황을 방지 할 수 있다.

다시 말해, Leader Partition을 최대한 많은 브로커에 분산시켜 트래픽에 대한 병목 현상을 방지하며, failover에 대해서도 이를 유지할 수 있는 아키텍처가 이상적인 아키텍처이다.

또한 설계하기 나름이지만, offset 설계를 잘하여 데이터의 정합성까지 지켜준다면 가상 이상적인 아키텍처가 될 것이다.