각각의 이벤트만 봐서는 안 되는 작업이 있다. 한 시간 동안의 타입별 이벤트 수, 조인, 합계, 평균. 이럴 때 추적해야 하는 정보를 상태(state)라고 한다.
상태가 유실되는 지점
이벤트 개수를 세는 해시 테이블을 애플리케이션 변수에 두면 될 것 같다. 그런데 애플리케이션이 정지하거나 크래시하면 상태가 유실되고 결과가 달라진다.
최신 상태를 보존하면서 재시작할 때 복구되게 만드는 것, 그것이 상태 관리의 전부다.
로컬 상태와 외부 상태
로컬 상태는 애플리케이션에 내장되어 구동되는 인메모리 데이터베이스에 둔다. 매우 빠른 대신 사용 가능한 메모리 크기의 제한을 받는다. 외부 상태는 카산드라 같은 NoSQL 시스템에 둔다. 크기 제한이 없고 여러 인스턴스가 접근할 수 있는 대신 지연이 늘고 복잡도와 가용성 문제가 따라온다.
스트림 처리의 많은 디자인 패턴이 데이터를 분할해서 한정된 로컬 상태로 처리 가능한 서브스트림으로 만드는 데 초점을 둔다. 로컬 상태의 빠름을 얻으면서 크기 제한을 우회하는 방법이다. 외부 상태를 쓰더라도 내용물을 로컬에 캐싱해서 외부와 가능한 한 통신하지 않게 만드는 것이 보통이고, 그러면 내부와 외부를 일관되게 유지하는 것이 새로운 과제로 남는다.
로컬 상태로 충분한 근거는 파티셔너에 있다. 같은 항목의 모든 이벤트가 같은 파티션에 쓰이도록 하면, 각 파티션은 하나의 컨슈머만 읽으므로 각 인스턴스는 자기가 맡은 항목들에 대한 상태만 들고 있으면 된다. 그룹별 집계가 공유 상태 없이 성립하는 이유이고, 카프카에서 상태 있는 처리가 확장되는 방식이다.
메모리, 영속성, 리밸런싱
로컬 상태를 보유하는 순간 애플리케이션이 복잡해진다. 고려할 것이 셋이다.
메모리부터 본다. 로컬 상태는 인스턴스의 가용 메모리 안에 들어가는 것이 이상적이다. 디스크에 넘기는 저장소도 있지만 성능에 상당한 영향을 준다.
영속성은 인스턴스가 종료돼도 상태가 유실되지 않아야 한다는 요구다. 카프카 스트림즈는 RocksDB로 인메모리 저장과 디스크 영속화를 같이 하는데, 진짜 안전장치는 그것이 아니다. 로컬 상태의 모든 변경 사항이 카프카 토픽으로도 보내진다. 이것이 체인지 로그다. 노드에 장애가 나도 카프카 토픽에서 이벤트를 읽어오면 로컬 상태가 복구된다. 이 토픽이 무한정 자라지 않도록 로그 압착을 쓰고, 각 키의 최신 값만 남으므로 언제든 처음부터 읽어 현재 상태를 만들 수 있다.
리밸런싱이 셋째다. 파티션이 다른 컨슈머에게 재할당되면, 파티션을 잃은 인스턴스는 마지막 상태를 저장해서 새로 받은 인스턴스가 복구할 수 있게 해야 한다.
복구 시간과 스탠바이 레플리카
복구에 걸리는 시간이 그대로 가용성 손실이 된다. 장애가 난 스레드의 태스크를 다른 스레드가 넘겨받을 때 가장 먼저 하는 일이 상태 복구다. 내부 토픽을 다시 읽어 상태 저장소를 갱신하는 동안은 처리가 진행되지 않으므로 가용성이 줄고 출력이 뒤처진다. 줄이는 방법은 압착이 시작되기까지의 지연을 낮추고(토픽 단위로는 min.compaction.lag.ms, 브로커 단위로는 log.cleaner.min.compaction.lag.ms), 세그먼트 크기를 기본 1GB에서 100MB 정도로 낮추는 것이다.
더 나은 방법도 있다. 현재 작동 중인 태스크를 따라가기만 하는 스탠바이 레플리카를 다른 서버에 두면, 장애가 났을 때 이미 거의 최신 상태를 보유하고 있으므로 중단 시간이 거의 없이 처리를 재개한다.
참고
원본은 복구 시간을 줄이는 설정으로 compaction.lag.ms를 적었는데 그런 이름의 설정은 없다. 토픽 단위 설정은 min.compaction.lag.ms(기본 0), 브로커 단위 설정은 log.cleaner.min.compaction.lag.ms(기본 0)다. 둘 다 메시지가 압착 대상이 되기까지 로그에 머무는 최소 시간을 정한다. Kafka Topic Configs