Can't inject a Repository bean extending RxJavaCrudRepository

[aliabdikk]

I have a Micronaut project, and in a part of it I'm using reactive programming to perform a IO-heavy job that fetches a lot of data from database, of course from different threads. I was using the JpaRepository like normal DB connections, and it was working OK.

After migrating to Micronaut 4 (and consequently Hibernate 6), my CLI job started to fail with java.lang.IllegalStateException: Illegal pop() with non-matching JdbcValuesSourceProcessingState. After digging a bit, I fount out the culprit could be that the Hibernate transaction is being used in a multi-threaded context in parallel, which should not be.

One option is trying to acquire a new transaction on each operation through TransactionDefinition.Propagation.REQUIRES_NEW, and that worked well.

But to me the cleanest way is to actually use the reactive repository implementation provided by Micronaut data. So I switched to use RxJavaCrudRepository, but the moment I change my repository interface from extending JpaRepository to RxJavaCrudRepository, it fails to start with the following message:

No bean of type [com.my.project.data.MyReactiveRepository] exists.
Make sure the bean is not disabled by bean requirements (enable trace logging for
'io.micronaut.context.condition' to check) and if the bean is enabled then ensure the 
class is declared a bean and annotation processing is enabled (for Java and Kotlin 
the 'micronaut-inject-java' dependency should be configured as an annotation processor).

So apparently Micronaut is not instantiating a bean when the interface is extending RxJavaCrudRepository.

I did the following steps to replace my normal repository with a reactive one:

1. Added io.micronaut.data:micronaut-data-hibernate-reactive to the dependancies.
2. Configured a reactive JPA datasource, like this (I already had a read-only and read-write datasource)

   datasources:
   default: # Read-Only datasource
   url: ${JDBC_URL_RO}
   username: ${JDBC_USER}
   password: ${JDBC_PASSWORD}
   dialect: POSTGRES
   read-only: true
   rx: # Reactive datasource (Read-Only)
   url: ${JDBC_URL_RO}
   username: ${JDBC_USER}
   password: ${JDBC_PASSWORD}
   dialect: POSTGRES
   read-only: true
   reactive: true
   rw: # Read-Write datasource
   url: ${JDBC_URL}
   username: ${JDBC_USER}
   password: ${JDBC_PASSWORD}
   dialect: POSTGRES

3. Define my repository like this

   import io.micronaut.data.repository.reactive.RxJavaCrudRepository;
   ...
   @Repository("rx")
   interface MyReactiveRepository
   extends RxJavaCrudRepository<Entity, EntityId> {
   
   @Query("SELECT e FROM Entity e WHERE e.field IN :field")
   Single<List<Entity>> getAllByCodes(List<String> field);
   }

4. Inject the repository in my job and use it:

   public class MyJob {
   private final MyReactiveRepository repository;
   
   public MyJob(MyReactiveRepository repository) {
       this.repository = repository;
   }
   
   // Just a minimised version to make the point, not the actual implementation
   public void run() {
       Flowable.fromIterable(batchOfData)
           .buffer(bufferSize)
           .flatMap(batch -> repository.getAllByCodes(batch).toFlowable())
           .parallel(entryGenerationParallelism)
           .runOn(io)
           .blockingGet();
   }
   }

The above implementation starts successfully when using JpaRepository, so I assume I'm missing some configuration to make Micronaut instantiate the bean, but after many investigations, I couldn't find anything.

Any help will be appreciated :)

[dstepanov]

Please create a sample app. Make sure you have RxJava on the classpath. It's recommended to use Reactor sure it will work with RxJava