I am trying to bring together transactional approach in operations that is happening on different external resources. Starting processing in a kafka consumer, I need to run CRUD operations on mongodb and redis. As transactional approach in Quarkus doesn't encompass all technologies in a common library, I was only able to come up with nesting withTransactional methods in each other.

emitter.withTransaction(emitter -> { 
    return reactiveRedis.withTransaction(tx -> { 
        return Panache.withTransaction(()-> {
               return redisOps.save(key, String , tx)
                      .mongoOps.save(myObject)
                      .onItem().transform(v->emmiter.send(myObject);});
    });
});

This is working as expected, Quarkus rollbacks all activities if an exception happens. However, as the architecture is layered and segregated into common libraries, I don't call database operations right in kafka client, but I need to pass tx object to libraries that I use for Redis CRUD operations. I find that is problematic because calling redisOps is nested in service methods rather than in kafka client. When I dug in Redis's withTransaction method, I see ReactiveTransactionRedisDataSourceImpl is instantiated rather than injecting it.

My question is whether there is a programmatic way to start the transaction in kafka client and inject the ReactiveTransactionRedisDataSource wherever I call CRUD methods or not? Or are there any other ways to make it better and neater instead of passing tx from method to method?