Блог

В org.springframework.kafka.listener.AbstractMessageListenerContainer, при запуске контейнера метод checkTopics проверяет, существует ли подписанная тема на брокере, используя потребителя Kafka, который создается в блоке try with resources .

Когда потребитель закрыт, закрытие каскадируется до многих закрываемых связанных объектов, в частности десериализаторов ключей и значений (см. org.apache .kafka.clients.consumer.KafkaConsumer). В приложении Spring десериализаторы обычно объявляются как компоненты, поэтому на фабрике каждого типа есть только один экземпляр, и хотя большинство десериализаторов реализуют close как no-op, я сталкивался со случаями, когда закрытие десериализатора делает его непригодным для использования с этого момента.

Мне кажется, что, хотя закрытие потребителя звучит разумно, учитывая, что Spring запускает несколько экземпляров, а созданный здесь — просто одноразовый, каскад до компонентов десериализатора является нежелательным последствием, которое, возможно, не было замечено при написании AbstractMessageListenerContainer .

Существует обходной путь — при создании KafkaListenerContainerFactory просто вызовите

 factory.getContainerProperties().setMissingTopicsFatal(false);
  

но это удаляет проверку безопасности для существования темы и кажется чем-то вроде взлома. Действительно ли закрытие потребителя правильно в AbstractMessageListenerContainer?

Мы можем и будем использовать AdminClient вместо a, Consumer чтобы проверить, существуют ли темы.

Однако это не панацея, поскольку stop() ping контейнера также закрывает Consumer (ы), поэтому та же проблема будет существовать при перезапуске контейнера (ов).

В подобных случаях лучше позволить Kafka управлять жизненным циклом десериализатора вместо объявления его как компонента.