Reactive and Non-Blocking with Kafka

[ghost]

Trying to return the value from the listener to the producer and perform some business logic with a reactive non-blocking approach, but never returns the values from the Kafka listener

Listener

@KafkaListener(offsetReset = OffsetReset.EARLIEST)
public class ProductListener {
    @Topic(ProductTopicConstants.GET_FREE_TEXT_SEARCH)
    public Flowable<Product> findByFreeText(String text) {
        LOG.info(String.format("Listening value = %s", text));
        return Flowable.just(new Product("This is the test", 0,"This is test description"));
    }
}

Producer interface

@Topic(ProductTopicConstants.GET_FREE_TEXT_SEARCH)
Flowable<Product> findFreeText(String text);

Implementation

@Override
    public Flowable<ProductViewModel> findFreeText(String text) {
        LOG.info("Manager --> Finding all the products");
        ProductViewModel model = new ProductViewModel();
        IProductProducer kafkaProducer = applicationContext.getBean(IProductProducer.class);
        return kafkaProducer.findFreeText(text).subscribe(item -> {
            System.out.println(item);
        });
    }

Never received the return value from the listener, the breakpoint never hit the for-loop

[graemerocher]

I am unsure why you are invoking the consumer directly, it is unclear from your code

[ghost]

What is the best approach to return the value from the consumer to the producer? I was following the micronaut documentation

Flowable Value and Return Type
Flowable<Book> sendBooks(
    @KafkaKey String author,
    Flowable<Book> book
);

Here is the repo of the code https://github.com/anandjaisy/MicronautKafka/blob/master/src/main/java/fete/bird/service/producer/manager/ProductManager.java