Existe-t-il un moyen standard de réinitialiser les décalages lors de l'utilisation de l'annotation @KafkaListener? - apache-kafka, printemps-kafka

J'utilise l'interface KafkaListener dans mon application Spring Boot, qui fonctionne parfaitement. Les décalages sont stockés par Kafka elle-même.

Maintenant, disons qu'un des consommateurs d'un sujet déploie une nouvelle version et visse 2 heures de messages. Ensuite, ils corrigent l'application et souhaitent démarrer la nouvelle version avec un décalage par rapport à il y a deux heures.

Je pouvais utiliser le paramètre consumer.seek () avec un précédent appel consommateur.offsetsForTimes (), mais cela n’est direct que lorsque j’utilise le mécanisme de scrutation, et non lorsque Spring est utilisé.

J’ai essayé d’utiliser ConsumerRebalanceListener, mais pour régler les compensations, il faut que le consommateur soit actif, ce que je n’ai pas lors de l’instanciation de ConsumerRebalanceListener

Dois-je passer à la scrutation? Puis-je contacter le consommateur avant que mon KafkaListener ne soit appelé pour la première fois, mais après que le consommateur se soit vu attribuer une partition?

Réponses:

Pas en même temps; le prochain Version 2.0 introduit ConsumerAwareMessageListener, qui permet de passer le Consumer comme argument dans le @KafkaListener méthode.