Handling InvalidStateStoreException: “the state store may have migrated to another instance”?

Spring Application 에서 Stream을 materialize 하기 위해서는 StateStore를 bean으로 접근 가능하게 해야한다. (xxxxRepository를 @Repository로 사용하는것과 같게 생각하면 된다)

그러기 위해서는 https://kafka.apache.org/10/javadoc/index.html?org/apache/kafka/streams/KafkaStreams.html 에서 store method를 이용해서 (최소한) KeyValueStore 형태의instance를 가지고 있어야 한다. 당연히 이 코드는 Spring Bean 내부에서 만들것이고, 그러면 제목과 같은 에러 메시지를 50프로 이상 볼 수 있다.

confluent의 FAQ에서 해답을 찾을수 있는데  KafkaStream의 state와 StateStore를 만드는 타이밍의 문제이다.
문제를 해결하기 위해서는 ApplicationReadyEvent 에서 처리를 하거나, 아니면 KafkaStream이  Running 상태로 바뀌기를 기다리거나 해야하는데, confluent 문서처럼 retry 형태로 구현하기로 했다.

sleep도 찝찝하지만 while true는 더 찝찝하기에 retry를 추가하였고, 그 횟수동안 store를 못 만든다면 그냥 Application 자체가 실행이 안되도록 RuntimeException을 일으키도록 하였다.