Kafka Streams State

들어가며

 카프카 스트림즈에 대한 개발 도중 아래와 같은 로그를 확인했습니다.

2019-12-09 17:22:50.108  INFO 51383 --- [-StreamThread-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=privacyMaskingStream-b09c662d-26ce-4a51-969f-3372161deb7c-StreamThread-1-restore-consumer, groupId=null] Unsubscribed all topics or patterns and assigned partitions
2019-12-09 17:22:50.108  INFO 51383 --- [-StreamThread-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=privacyMaskingStream-b09c662d-26ce-4a51-969f-3372161deb7c-StreamThread-1-restore-consumer, groupId=null] Unsubscribed all topics or patterns and assigned partitions
2019-12-09 17:22:50.108  INFO 51383 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [privacyMaskingStream-b09c662d-26ce-4a51-969f-3372161deb7c-StreamThread-1] State transition from PARTITIONS_ASSIGNED to RUNNING
2019-12-09 17:22:50.108  INFO 51383 --- [-StreamThread-1] org.apache.kafka.streams.KafkaStreams    : stream-client [privacyMaskingStream-b09c662d-26ce-4a51-969f-3372161deb7c] State transition from REBALANCING to RUNNING
State transition from REBALANCING to RUNNING
2019-12-09 17:22:50.117  INFO 51383 --- [-StreamThread-1] o.a.k.c.c.internals.SubscriptionState    : [Consumer clientId=privacyMaskingStream-b09c662d-26ce-4a51-969f-3372161deb7c-StreamThread-1-consumer, groupId=privacyMaskingStream] Resetting offset for partition target-topic-0 to offset 0.
2019-12-09 17:23:01.964  INFO 51383 --- [extShutdownHook] org.apache.kafka.streams.KafkaStreams    : stream-client [privacyMaskingStream-b09c662d-26ce-4a51-969f-3372161deb7c] State transition from RUNNING to PENDING_SHUTDOWN
State transition from RUNNING to PENDING_SHUTDOWN
2019-12-09 17:23:01.964  INFO 51383 --- [ms-close-thread] o.a.k.s.p.internals.StreamThread         : stream-thread [privacyMaskingStream-b09c662d-26ce-4a51-969f-3372161deb7c-StreamThread-1] Informed to shut down

...
2019-12-09 17:23:02.001  INFO 51383 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [privacyMaskingStream-b09c662d-26ce-4a51-969f-3372161deb7c-StreamThread-1] Shutdown complete
2019-12-09 17:23:02.003  INFO 51383 --- [ms-close-thread] org.apache.kafka.streams.KafkaStreams    : stream-client [privacyMaskingStream-b09c662d-26ce-4a51-969f-3372161deb7c] State transition from PENDING_SHUTDOWN to NOT_RUNNING
State transition from PENDING_SHUTDOWN to NOT_RUNNING
2019-12-09 17:23:02.003  INFO 51383 --- [extShutdownHook] org.apache.kafka.streams.KafkaStreams    : stream-client [privacyMaskingStream-b09c662d-26ce-4a51-969f-3372161deb7c] Streams client stopped completely

Process finished with exit code 130 (interrupted by signal 2: SIGINT)

 위 로그에서 제 눈에 띈 부분은 State transition from REBALANCING to RUNNING과 같이 스트림 애플리케이션 인스턴스의 상태 변화를 나타내는 부분입니다. 저는 스트림즈 인스턴스의 각 상태와 특징에 대해 궁금했습니다. 본 정리 글은 스트림즈 인스턴스 상태에 대해 정리한 글입니다.

Kafka Streams State

Kafka Streams Instance State

org.apche.kafka.KafkaStreams.State (enum)는 카프카 스트림즈 인스턴스의 상태, 줄여서 스트림즈의 상태를 나타내는 enum 클래스입니다. 스트림즈의 상태는 아래 상태 중 하나를 가집니다.

• CREATED : 스트림즈의 초기 상태.
• REBALANCING : 현재 스트림즈가 input 으로 받고 있는 source partition 에 대하여 소유권 할당이 진행되는 단계.
• RUNNING : 정상 구동 상태
• PENDING_SHUTDOWN : 종료 요청에 따라 스트림즈 인스턴스를 종료하기 위한 대기 상태
• NOT_RUNNING) : 종료된 상태
• ERROR : 에러 발생 상태

State Transit

스트림즈의 상태는 상황에 따라 다른 상태로 변경됩니다. 상태 별로 변경되는 상황은 다음과 같습니다.

Before	After	Trigger Statement
Created	Rebalancing	정상적인 인스턴스 실행 후
Running	Rebalancing	스트림 스레드 중 하나라도 PARTITION_REVOKED 상태가 된 경우
Rebalancing	Running	모든 스레드가 RUNNING 상태인 경우
Created, Error, Rebalancing, Running	Pending Shutdown	종료 요청이 호출된 경우
Rebalancing, Running	Error	글로벌 스트림 스레드가 죽거나,  모든 스트림 스레드가 죽었을 경우  (혹은 둘 다)
Pending Shutdown	Not Running	종료된 경우

좀 더 보기 쉽게 그림으로 표현하면 다음과 같습니다.

악필입니다만;;

 가장 유의해야할 부분은 ERROR 상태에서는 종료 외의 상태 변경이 불가능하다는 것입니다.

마무리

이번 글을 통해 카프카 스트림즈의 상태에 대해서 간략히 알아봤습니다.

참고

• KafkaStreams.State (kafka 2.3.0 API)