Duplicated Context in StartupEvent Observers

[knutwannheden]

Description

In #28017 the scheduler was changed so that scheduled methods run in the context of a duplicated context. Now I have encountered similar problems in the context of a StartupEvent observer. I am thus wondering whether it would again make sense to change Quarkus, so that a duplicated context (and thus context locals) are available.

Implementation ideas

No response

[knutwannheden]

/cc @cescoffier

[cescoffier]

@mkouba I think we discussed it at some point, but I can't remember the outcome.

[mkouba]

#27623 and #27854 are related. I think that we should introduce some way (annotation?) to wrap the notification like this.

[quarkus-bot]

/cc @manovotn, @mkouba

[cescoffier]

@mkouba fancy to propose the feature?

[mkouba]

@mkouba fancy to propose the feature?

I'm still not sure how this should/could work. One of the questions is: do we want to execute the logic asynchronously, i.e. not necessarily during the startup sequence? Otherwise, we'd have to block and wait (and this should be fine in the context of StartupEvent).

[cescoffier]

That's what I would have done: call all the observers and join them before going further.

Maybe instead of join, it might be easier to invoke them sequentially.

[mkouba]

In theory, we could support this for any observer. I.e. something along the lines of: an observer method annotated with @NonBlocking is executed on the event loop (vertx duplicated context more precisely). OTOH it might be a bit confusing.

@manovotn @Ladicek WDYT?

[Ladicek]

I've got this feeling that you would need the observer method to return a Uni (or a CompletionStage), so that the observer can actually do something non-blocking. At which point, this looks dangerously close to #29191.

[mkouba]

I've got this feeling that you would need the observer method to return a Uni (or a CompletionStage), so that the observer can actually do something non-blocking. At which point, this looks dangerously close to #29191.

Not really. This is mostly about executing the observer method on the event loop; not about asynchronous invocation or participating in a reactive stream. In other words, a user just needs to execute the observer method on the event loop but block the current thread (and wait for the result).

   void onStart(@Observes StartupEvent event, Vertx vertx, Mutiny.SessionFactory factory) {
        Context context = VertxContext.getOrCreateDuplicatedContext(vertx);
        VertxContextSafetyToggle.setContextSafe(context, true);
        CompletableFuture<Void> fu = new CompletableFuture<>();
        context.runOnContext(new Handler<Void>() {
            @Override
            public void handle(Void event) {
                // do something with the session factory
               fu.complete(null); 
            }
        });
       fu.get(); // wait for the result
    }

Could be replaced with something like:

   @NonBlocking
   void onStart(@Observes StartupEvent event, Mutiny.SessionFactory factory) {
      // do something with the session factory
   }

[Ladicek]

It actually can't be replaced the way you mention without significantly limiting what the observer method can do. In particular, it cannot start any asynchronous action, because there's no way to signal its completion.

[mkouba]

In particular, it cannot start any asynchronous action, because there's no way to signal its completion.

Why? You can just subscribe to the Uni returned from the session factory for example...

[mkouba]

OTOH I agree that returning Uni or CompletionStage from the observer method would be more convenient. And that's what you can do with a @Scheduled method by the way.

[Ladicek]

You can subscribe to the Uni, but you can't wait for it to complete (as you're on the event loop) and you can't return it so that the original thread waits for it. So you can do "fire and forget" kind of thing, yes. I don't think that's good, as you lose the ability to rely on startup tasks being finished when the application finishes startup, but you're right that it's possible.

[mkouba]

You can subscribe to the Uni, but you can't wait for it to complete (as you're on the event loop) and you can't return it so that the original thread waits for it. So you can do "fire and forget" kind of thing, yes. I don't think that's good, as you lose the ability to rely on startup tasks being finished when the application finishes startup, but you're right that it's possible.

You're right.

[mkouba]

So it seems that the only reasonable thing we could do is:

1. Require such an observer to return an async type; which goes against Enforce validation of return type of observer methods jakartaee/cdi#636
2. Provide a util method to wrap the execution; which is not very convenient

public class VertxContexts {
   public static void runOnVertxContext(Supplier<Uni<Void>> uniSupplier) {
     Vertx vertx = lookupVertx();
     Context context = VertxContext.getOrCreateDuplicatedContext(vertx);
     VertxContextSafetyToggle.setContextSafe(context, true);
     CompletableFuture<Void> fu = new CompletableFuture<>();
     context.runOnContext(new Handler<Void>() {
         @Override
         public void handle(Void event) {
             uniSupplier.get().subscribe().with(i -> fu.complete(null), e -> fu.completeExceptionally(e))
         }
     });
    fu.get();
   }
}

and then:

void onStart(@Observes StartupEvent event, Mutiny.SessionFactory factory) {
   VertxContexts.runOnVertxContext(() -> {
      // do something with the session factory and return Uni<Void>
   });
}

Am I missing something?

[Ladicek]

I think the 2nd option is sensible in the short term.

Longer term, I think we could add support for async observers returning Uni (or CompletionStage), but this use case would require firing the startup event asynchronously (in addition to synchronously), which is probably a more interesting topic...

[mkouba]

I think the 2nd option is sensible in the short term.

Ok, I'll try to prepare something...