Встроенное хранилище состояний миграции Kafka между двумя экземплярами с разными @StreamListener

Question

У меня есть приложение SpringBoot с двумя потоковыми процессорами, сопоставленными через Spring Cloud. Каждый процессор имеет собственный @StreamListener для разных тем. Один процессор записывает агрегированные данные в надежное хранилище состояний. Я столкнулся с проблемой в моем модульном тесте при получении данных через мой @Service (Сервис получает агрегированные данные из хранилища состояний). По какой-то причине время от времени ловлю исключение:

org.apache.kafka.streams.errors.InvalidStateStoreException: the state store, recently-played-store, may have migrated to another instance. at org.apache.kafka.streams.state.internals.QueryableStoreProvider.getStore(QueryableStoreProvider.java:60) at org.apache.kafka.streams.KafkaStreams.store(KafkaStreams.java:1043) at org.springframework.cloud.stream.binder.kafka.streams.QueryableStoreRegistry.getQueryableStoreType(QueryableStoreRegistry.java:47)

Когда я удаляю StreamListener с другого процессора, все работает нормально и стабильно.

Как связать хранилище состояний для точного экземпляра с надлежащим процессором?

Answer 1

Я нашел решение своей проблемы, может быть, это поможет кому-то еще. Я использовал не последнюю версию spring-cloud-stream-binder-kafka-streams. Моя версия была 2.0.0, и в ней есть ошибка https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/366. Это было исправлено только в версии 2.0.1