Issues with Returning Stream<> from Repository Method

[PT-88]

Expected Behavior

Expecting a stream to be returned from repository method and stay open until completed with it in the calling method.

Actual Behaviour

The stream closes before the calling method has completed using the stream.

Exception received: org.hibernate.resource.jdbc.internal.LogicalConnectionManagedImpl is closed

Steps To Reproduce

1. Set up repository

   public interface ProductRepository<T extends ProductDataAccessObject> extends CrudRepository<T, UUID> {
   @Nonnull
   Stream<T> findByPriceBetween(double min, double max);
   }

2. Set up caller method

   @Transactional
   protected void test(double min, double max) throws Exception {
       try (var products = this.products.findByPriceBetween(min, max)) {
           final var list = products.map(ProductDataAccessObject::entity).toList();
       }
   }

Environment Information

• JDK: 20
• MacOS

Example Application

No response

Version

4.1.1

[dstepanov]

Can you please provide a project that reproduces it?

[dstepanov]

Closing, because of missing reproducer

[tucker-eventpipe]

I've got a reproducer for you: https://github.com/tucker-eventpipe/sampleApp/tree/stream

On the stream branch in TestStartupService.groovy: https://github.com/tucker-eventpipe/sampleApp/blob/stream/src/main/groovy/com/mycompany/TestStartupService.groovy

I get the same error but I'm using Mongo, which suggests it's not dependent on the source. I can open up a separate issue if you like or you can reopen this one. This is the full stack trace:

13:39:52.992 [main] ERROR io.micronaut.runtime.Micronaut - Error starting Micronaut server: state should be: open
java.lang.IllegalStateException: state should be: open
    at com.mongodb.assertions.Assertions.isTrue(Assertions.java:109)
    at com.mongodb.internal.session.BaseClientSessionImpl.getServerSession(BaseClientSessionImpl.java:125)
    at com.mongodb.internal.session.ClientSessionContext.getSessionId(ClientSessionContext.java:48)
    at com.mongodb.internal.connection.ClusterClockAdvancingSessionContext.getSessionId(ClusterClockAdvancingSessionContext.java:50)
    at com.mongodb.internal.connection.CommandMessage.getExtraElements(CommandMessage.java:230)
    at com.mongodb.internal.connection.CommandMessage.encodeMessageBodyWithMetadata(CommandMessage.java:155)
    at com.mongodb.internal.connection.RequestMessage.encode(RequestMessage.java:136)
    at com.mongodb.internal.connection.CommandMessage.encode(CommandMessage.java:59)
    at com.mongodb.internal.connection.InternalStreamConnection.sendAndReceive(InternalStreamConnection.java:360)
    at com.mongodb.internal.connection.UsageTrackingInternalConnection.sendAndReceive(UsageTrackingInternalConnection.java:114)
    at com.mongodb.internal.connection.DefaultConnectionPool$PooledConnection.sendAndReceive(DefaultConnectionPool.java:765)
    at com.mongodb.internal.connection.CommandProtocolImpl.execute(CommandProtocolImpl.java:76)
    at com.mongodb.internal.connection.DefaultServer$DefaultServerProtocolExecutor.execute(DefaultServer.java:209)
    at com.mongodb.internal.connection.DefaultServerConnection.executeProtocol(DefaultServerConnection.java:115)
    at com.mongodb.internal.connection.DefaultServerConnection.command(DefaultServerConnection.java:83)
    at com.mongodb.internal.connection.DefaultServerConnection.command(DefaultServerConnection.java:74)
    at com.mongodb.internal.connection.DefaultServer$OperationCountTrackingConnection.command(DefaultServer.java:299)
    at com.mongodb.internal.operation.QueryBatchCursor.lambda$getMore$1(QueryBatchCursor.java:285)
    at com.mongodb.internal.operation.QueryBatchCursor$ResourceManager.executeWithConnection(QueryBatchCursor.java:514)
    at com.mongodb.internal.operation.QueryBatchCursor.getMore(QueryBatchCursor.java:282)
    at com.mongodb.internal.operation.QueryBatchCursor.doHasNext(QueryBatchCursor.java:162)
    at com.mongodb.internal.operation.QueryBatchCursor$ResourceManager.execute(QueryBatchCursor.java:407)
    at com.mongodb.internal.operation.QueryBatchCursor.hasNext(QueryBatchCursor.java:149)
    at com.mongodb.client.internal.MongoBatchCursorAdapter.hasNext(MongoBatchCursorAdapter.java:52)
    at io.micronaut.data.mongodb.operations.DefaultMongoRepositoryOperations$1.tryAdvance(DefaultMongoRepositoryOperations.java:266)
    at java.base/java.util.Spliterator.forEachRemaining(Spliterator.java:332)
    at java.base/java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:762)
    at org.codehaus.groovy.vmplugin.v8.IndyInterface.fromCache(IndyInterface.java:321)
    at com.mycompany.TestStartupService.onStartup(TestStartupService.groovy:31)
    at com.mycompany.$TestStartupService$Definition$Exec.dispatch(Unknown Source)
    at io.micronaut.context.AbstractExecutableMethodsDefinition$DispatchedExecutableMethod.invoke(AbstractExecutableMethodsDefinition.java:456)
    at io.micronaut.context.DefaultBeanContext$BeanExecutionHandle.invoke(DefaultBeanContext.java:3940)
    at io.micronaut.aop.chain.AdapterIntroduction.intercept(AdapterIntroduction.java:87)
    at io.micronaut.aop.chain.MethodInterceptorChain.proceed(MethodInterceptorChain.java:143)
    at com.mycompany.TestStartupService$ApplicationEventListener$onStartup1$Intercepted.onApplicationEvent(Unknown Source)
    at io.micronaut.context.event.ApplicationEventPublisherFactory.notifyEventListeners(ApplicationEventPublisherFactory.java:266)
    at io.micronaut.context.event.ApplicationEventPublisherFactory$2.publishEvent(ApplicationEventPublisherFactory.java:226)
    at io.micronaut.context.DefaultBeanContext.publishEvent(DefaultBeanContext.java:1831)
    at io.micronaut.context.DefaultBeanContext.start(DefaultBeanContext.java:360)
    at io.micronaut.context.DefaultApplicationContext.start(DefaultApplicationContext.java:215)
    at io.micronaut.runtime.Micronaut.start(Micronaut.java:75)
    at io.micronaut.runtime.Micronaut.run(Micronaut.java:334)
    at io.micronaut.runtime.Micronaut.run(Micronaut.java:320)
    at com.mycompany.Application.main(Application.groovy:10)