Using runSubscriptionOn(Infrastructure.getDefaultExecutor()) on Quarkus Mutiny leadding RejectedExecutionException

[lkyhfx]

The code bellow create a simple Multi using runSubscriptionOn but throw RejectedExecutionException.

Multi.createFrom().emitter(em -> {
                while (...) {
                    em.emit(...);
                }
                em.complete();
            }).runSubscriptionOn(Infrastructure.getDefaultExecutor()).map(o -> (String) o);

Dependencies in pom.xml

  <dependencies>
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-mutiny</artifactId>
    </dependency>
    <dependency>
      <groupId>io.smallrye.reactive</groupId>
      <artifactId>smallrye-mutiny-vertx-web-client</artifactId>
      <version>3.5.0</version>
    </dependency>
    <dependency>
      <groupId>org.apache.camel.quarkus</groupId>
      <artifactId>camel-quarkus-tika</artifactId>
    </dependency>
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-reactive-pg-client</artifactId>
    </dependency>
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-arc</artifactId>
    </dependency>
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-vertx</artifactId>
    </dependency>
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-junit5</artifactId>
      <scope>test</scope>
    </dependency>
  </dependencies>

[cescoffier]

That's because the executor is getting shutdown when the application stops.

[lkyhfx]

Thanks for reply! The simple code bellow can reproduce this problem, any hint to debug the code?

Simple.java

@QuarkusMain
public class Simple implements QuarkusApplication {

    @Inject
    WebPage webPage;


    @Override
    public int run(String... args) throws Exception {
        var pattern = Pattern.compile(".*quick-start.html");
        webPage.getMulti("https://spark.apache.org/docs/3.4.1/", pattern)
            .subscribe().with(
                item -> System.out.println(item),
                failure -> failure.printStackTrace(),
                () -> System.out.println("Completed")
            );
        return 0;

    }
}

WebPage.java

@ApplicationScoped
public class WebPage {
    private final WebClient client;


    @Inject
    public WebPage(Vertx vertx) {
        this.client = WebClient.create(vertx);
    }

    public Multi<String> getMulti(String url, Pattern pattern) {
        return client.getAbs(url).send()
            .onItem().transformToMulti(bufferHttpResponse -> {
            return getUrls(bufferHttpResponse.bodyAsString(), pattern);
        });
    }

    public Multi<String> getUrls(String html, Pattern pattern) {
        var matcher = pattern.matcher(html);
        return Multi.createFrom().emitter(em -> {
                while (matcher.find()) {
                    em.emit(matcher.group());
                }
                em.complete();
            }).runSubscriptionOn(Infrastructure.getDefaultExecutor()).map(o -> (String) o);
    }
}

It just visit a url and extract all match string and return a Multi<String>, finally print the result to stdout.

[cescoffier]

Your application stops before Mutiny has the chance to do anything (leading to the rejection).
Use asyncExit and waitForExit as indicated in https://quarkus.io/guides/lifecycle#the-main-method