Estoy tratando de implementar el consumidor reactivo de kafka en mi aplicación de arranque Spring y estoy viendo estos ejemplos: https://github.com/reactor/reactor-kafka/blob/master/reactor-kafka-samples/src/main /java/reactor/kafka/samples/SampleScenarios.java
y parece que todavía no hay soporte para Spring en kafka reactivo
Entiendo cómo funcionan los oyentes de kafka en la API de kafka no reactiva en Spring: la solución más simple es configurar beans para ConcurrentKafkaListenerContainerFactory y ConsumerFactory, luego usar la anotación @KafkaListener y listo
Pero no estoy seguro de cómo usar Kafka reactivo en Spring en este momento.
Básicamente necesito un oyente para el tema. ¿Debo crear algún tipo de bucle o programador propio? O tal vez me estoy perdiendo algo. ¿Alguien puede compartir sus conocimientos y mejores prácticas?
Todavía no tengo una solución lista, pero estoy intentando esto (código Kotlin, Spring Boot). Alguien publicó parte de este fragmento de código aquí https://github.com/reactor/reactor-kafka/issues/100
@EventListener(ApplicationStartedEvent::class) fun onSomeEvent() { kafkaReceiver .receive() .doOnNext { record -> val myEvent = record.value() processMyEvent(myEvent).thenEmpty { record.receiverOffset().acknowledge() } } .doOnError { /* todo */ } .subscribe() }Busque otras preguntas de desbordamiento de pila. No hay mucho allí, pero tal vez te dé algunas ideas.