4 Comments
User's avatar
Boris's avatar

Hello Markus, this is a great article! Could you please elaborate if this solution incrementally fetching from PostgreSQL?

For example,

Postgres

-> fetch 1000

-> emit 1000

-> fetch next 1000

...

or

Postgres

-> fetch all rows

-> emit as Multi

I am thinking of two aspects. First, memory usage (it makes sense not to keep the entire result in memory, but do it in portions). Second, if new data arrives to the database, then "fetch all rows" will not bring the new data after the "fetch all rows" is done.

Thank you very much!

yborgess's avatar

Hi I Markus, thanks for these tutorials, I'm learning a lot Quarkus with them.

I was unable to run the example as it is:

First, the The import.sql files uses a sequence that is not created by by a PanacheEntity, see "hibernate_sequence", this sequence is not created when using the record pattern

I tried to make it work by using custom annotations for the Fruit ID and extending from PanacheEntityBase instead:

@Entity

public class Fruit extends PanacheEntityBase {

@Id

@SequenceGenerator(name = "fruitSequence", sequenceName = "hibernate_sequence", allocationSize = 1)

@GeneratedValue(strategy = GenerationType.SEQUENCE, generator = "fruitSequence")

public Long id;

....

}

But it also does not work, it throws the following error in the logs:

2025-07-17 10:21:48,195 ERROR [org.jbo.res.rea.ser.han.PublisherResponseHandler] (vert.x-eventloop-thread-1) Exception in SSE server handling, impossible to send it to client: java.lang.IllegalStateException: This method is normally automatically overridden in subclasses: did you forget to annotate your entity with @Entity?

at io.quarkus.hibernate.reactive.panache.common.runtime.AbstractJpaOperations.implementationInjectionMissing(AbstractJpaOperations.java:318)

at io.quarkus.hibernate.reactive.panache.PanacheEntityBase.listAll(PanacheEntityBase.java:390)

at io.quarkus.hibernate.reactive.panache.Panache.lambda$withSession$0(Panache.java:29)

at io.smallrye.context.impl.wrappers.SlowContextualFunction.apply(SlowContextualFunction.java:21)

at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.performInnerSubscription(UniOnItemTransformToUni.java:68)

at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.onItem(UniOnItemTransformToUni.java:57)

at io.smallrye.mutiny.operators.uni.UniOnItemConsume$UniOnItemComsumeProcessor.onItem(UniOnItemConsume.java:43)

at io.smallrye.mutiny.operators.uni.UniOnItemTransform$UniOnItemTransformProcessor.onItem(UniOnItemTransform.java:43)

at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.onItem(UniOnItemTransformToUni.java:60)

at io.smallrye.mutiny.operators.uni.UniOperatorProcessor.onItem(UniOperatorProcessor.java:47)

at io.smallrye.mutiny.operators.uni.UniOnCancellationCall$UniOnCancellationCallProcessor.onItem(UniOnCancellationCall.java:52)

at io.smallrye.mutiny.operators.uni.builders.UniCreateFromItemSupplier.subscribe(UniCreateFromItemSupplier.java:29)

at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:35)

at io.smallrye.mutiny.operators.uni.UniOnCancellationCall.subscribe(UniOnCancellationCall.java:27)

at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:35)

at io.smallrye.mutiny.operators.uni.UniOnFailureFlatMap.subscribe(UniOnFailureFlatMap.java:31)

at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:35)

at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.performInnerSubscription(UniOnItemTransformToUni.java:81)

at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.onItem(UniOnItemTransformToUni.java:57)

at io.smallrye.mutiny.operators.uni.UniOperatorProcessor.onItem(UniOperatorProcessor.java:47)

at io.smallrye.mutiny.operators.uni.builders.UniCreateFromCompletionStage$CompletionStageUniSubscription.forwardResult(UniCreateFromCompletionStage.java:63)

at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)

at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)

at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)

at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147)

at org.hibernate.reactive.pool.impl.SqlClientPool.lambda$completionStage$2(SqlClientPool.java:173)

at io.vertx.core.impl.future.FutureImpl$4.onSuccess(FutureImpl.java:176)

at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:66)

at io.vertx.core.impl.future.FutureImpl.tryComplete(FutureImpl.java:259)

at io.vertx.core.impl.future.Mapping.onSuccess(Mapping.java:40)

at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:66)

at io.vertx.core.impl.future.FutureImpl.tryComplete(FutureImpl.java:259)

at io.vertx.core.impl.future.Mapping.onSuccess(Mapping.java:40)

at io.vertx.core.impl.future.FutureBase.lambda$emitSuccess$0(FutureBase.java:60)

at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)

at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)

at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)

at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:569)

at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)

at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)

at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)

at java.base/java.lang.Thread.run(Thread.java:840)

I only made it work by switching from record pattern to repository pattern

Markus Eisele's avatar

Hey! That is weird. I did add a link to the repository with a working example now:

https://github.com/myfear/ejq_substack_articles/tree/main/reactive-streaming-example

And I did also add the necessary sequence. That I did indeed skip. Thanks for pointing it out. Give it a shot and lmk if you still get the exception.

yborgess's avatar

> Hey! That is weird.

Indeed. I can reproduce it with the code taken directly from your repository by using this method:

public Multi<Fruit> streamFruits() {

return Panache.withSession(Fruit::listAll)

.onItem().transformToMulti(Multi.createFrom()::iterable)

.onItem().castTo(Fruit.class);

}

Which, at my eyes, should be equivalent to the one that you used which does not generates the issue:

public Multi<Fruit> streamFruits() {

return Panache.withSession(() -> Fruit.listAll())

.onItem().transformToMulti(Multi.createFrom()::iterable)

.onItem().castTo(Fruit.class);

}

I'm passing an static method reference to the Panache.withSession method and you are using a lambda. Maybe there is a subtle detail here I don't see.