Operator has been terminated issue with RowsFetchSpec and Spring DatabaseClient

[sephiroth-j]

When a mapping function via map is used on a GenericExecuteSpec, the onErrorDropped hook is triggered twice and the following error is logged. This does not happen when fetch is used with its default row mapper that maps the result to a Map<String,Object>. The query is successfully executed in both cases.

Error

2023-09-19T11:00:49.267+02:00 ERROR 4268 --- [onPool-worker-1] reactor.core.publisher.Operators         : Operator called default onErrorDropped

reactor.core.Exceptions$StaticThrowable: Operator has been terminated

2023-09-19T11:00:49.271+02:00 ERROR 4268 --- [onPool-worker-1] reactor.core.publisher.Operators         : Operator called default onErrorDropped

reactor.core.Exceptions$StaticThrowable: Operator has been terminated

Example

@DataR2dbcTest(properties =
{
	"spring.r2dbc.url=r2dbc:oracle://testdb",
	"spring.r2dbc.username=sa",
	"spring.r2dbc.password=sa",
})
class DatabaseClientTest
{
	@Autowired
	private DatabaseClient dbClient;

	@Test
	void testMethod()
	{
		dbClient.sql("select '1234' value from dual")
				.map(readable -> readable.get(0, String.class)).first()// onErrorDropped: Operator has been terminated
//				.fetch().one().map(row -> row.get("value")).cast(String.class)// no onErrorDropped
				.as(StepVerifier::create)
				.expectNext("1234")
				.verifyComplete();
	}
}

spring-oracle-r2dbc-issue.zip

[sephiroth-j]

Hm, it seems that the mapping function is not the problem. It's one() vs first() that makes the difference. first() causes the error, while one() does not.

It boils down to all().next() vs. all().buffer(2).next().
https://github.com/spring-projects/spring-framework/blob/73a1f4f82e029f5963469af32f9cb6395129af53/spring-r2dbc/src/main/java/org/springframework/r2dbc/core/DefaultFetchSpec.java#L60-L79

[Michael-A-McMahon]

Hi, @sephiroth-j. Thank you for reporting this.

I think I've got a fix: #134
Tested with your code and those log messages went away.

In a nutshell: Oracle R2DBC will handle resource clean up the way Project Reactor intends it to: Flux.usingWhen(...)

[sephiroth-j]

Hey, thanks for the quick solution. Until the release with the fix is available, I can use one() in combination with a "limit 1" clause as a workaround ... which probably makes more sense anyway, since I'm only interested in the first row anyway.

[sephiroth-j]

Hi @Michael-A-McMahon, can you give an estimated date for the release of this fix?

[Michael-A-McMahon]

Hi @sephiroth-j, I'll check with my team about a new release. I think we are due for one. I can't estimate a date right now, but I'll update this thread once I know more.

[Michael-A-McMahon]

Going to get a few more changes in this week, and then get our release process started. Next release could happen by the end of next week, but there's no guarantee.