I am specifically seeing this in integration tests where I do a blocking call, and I can reliably reproduce this under high CPU load. The use of EmitterProcessor in ReactorNettyClient seems to be issue, but I am not 100% sure.
reactor.core.Exceptions$OverflowException: Backpressure overflow during Sinks.Many#emitNext
at reactor.core.Exceptions.failWithOverflow(Exceptions.java:233) ~[reactor-core-3.4.3.jar:3.4.3]
Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Assembly trace from producer [reactor.core.publisher.MonoFlatMapMany] :
reactor.core.publisher.Mono.flatMapMany
io.r2dbc.mssql.client.ReactorNettyClient.exchange(ReactorNettyClient.java:591)
Error has been observed at the following site(s):
|_ Mono.flatMapMany ⇢ at io.r2dbc.mssql.client.ReactorNettyClient.exchange(ReactorNettyClient.java:591)
|_ Flux.handle ⇢ at io.r2dbc.mssql.client.ReactorNettyClient.exchange(ReactorNettyClient.java:591)
|_ Flux.doAfterTerminate ⇢ at io.r2dbc.mssql.client.ReactorNettyClient.exchange(ReactorNettyClient.java:601)
|_ Flux.doOnCancel ⇢ at io.r2dbc.mssql.client.ReactorNettyClient.exchange(ReactorNettyClient.java:601)
|_ Flux.doOnSubscribe ⇢ at io.r2dbc.mssql.QueryMessageFlow.exchange(QueryMessageFlow.java:53)
|_ Flux.handle ⇢ at io.r2dbc.mssql.QueryMessageFlow.exchange(QueryMessageFlow.java:54)
|_ ⇢ at io.r2dbc.mssql.util.Operators.discardOnCancel(Operators.java:43)
|_ Flux.transform ⇢ at io.r2dbc.mssql.SimpleMssqlStatement.lambda$execute$0(SimpleMssqlStatement.java:135)
|_ Flux.doOnDiscard ⇢ at io.r2dbc.mssql.SimpleMssqlStatement.lambda$execute$0(SimpleMssqlStatement.java:135)
|_ Flux.windowUntil ⇢ at io.r2dbc.mssql.SimpleMssqlStatement.createResultStream(SimpleMssqlStatement.java:147)
|_ Flux.map ⇢ at io.r2dbc.mssql.SimpleMssqlStatement.createResultStream(SimpleMssqlStatement.java:148)
|_ Flux.defer ⇢ at io.r2dbc.mssql.SimpleMssqlStatement.execute(SimpleMssqlStatement.java:113)
|_ Flux.from ⇢ at org.springframework.r2dbc.core.DefaultDatabaseClient$DefaultGenericExecuteSpec.lambda$execute$3(DefaultDatabaseClient.java:375)
|_ Flux.cast ⇢ at org.springframework.r2dbc.core.DefaultDatabaseClient$DefaultGenericExecuteSpec.lambda$execute$3(DefaultDatabaseClient.java:376)
|_ checkpoint ⇢ SQL "
select * from something_with_more_than_250_rows
" [DatabaseClient]
|_ Flux.flatMap ⇢ at org.springframework.r2dbc.core.DefaultFetchSpec.lambda$all$2(DefaultFetchSpec.java:89)
|_ Flux.usingWhen ⇢ at org.springframework.r2dbc.core.DefaultDatabaseClient.inConnectionMany(DefaultDatabaseClient.java:134)
|_ Flux.onErrorMap ⇢ at org.springframework.r2dbc.core.DefaultDatabaseClient.inConnectionMany(DefaultDatabaseClient.java:146)
|_ ⇢ at org.springframework.r2dbc.core.DefaultFetchSpec.all(DefaultFetchSpec.java:87)
|_ Flux.subscriberContext ⇢ at kotlinx.coroutines.reactor.ReactorContextInjector.injectCoroutineContex
|_ Mono.create ⇢ at kotlinx.coroutines.reactor.MonoKt.monoInternal(Mono.kt:50)
|_ ⇢ at kotlinx.coroutines.reactor.MonoKt.mono(Mono.kt:32)
|_ Mono.subscribeOn ⇢ at company.TestSchedulerKt$runBlockingTest$1.invoke(TestScheduler.kt:18)
|_ Mono.publishOn ⇢ at company.TestSchedulerKt$runBlockingTest$1.invoke(TestScheduler.kt:19) at reactor.core.Exceptions.failWithOverflow(Exceptions.java:233) ~[reactor-core-3.4.3.jar:3.4.3]
at reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:46) ~[reactor-core-3.4.3.jar:3.4.3]
at reactor.core.publisher.EmitterProcessor.onNext(EmitterProcessor.java:265) ~[reactor-core-3.4.3.jar:3.4.3]
at io.r2dbc.mssql.client.ReactorNettyClient$1.next(ReactorNettyClient.java:244) ~[r2dbc-mssql-0.8.5.RELEASE.jar:0.8.5.RELEASE]
...
9 replies
rollbackTransaction doesn't assert that the connection isn't closed in r2dbc-pool? https://github.com/r2dbc/r2dbc-pool/blob/1c54e515e151ca082b766713960197085ed3c8c5/src/main/java/io/r2dbc/pool/PooledConnection.java#L138
I'm wondering what's the status of supporting execution of stored procedures?
It's a feature which blocks adoption for us.
Hi
i am using r2dbc with maria db and it works well. When i try to add BlockHound.install(); in main
i get the error :
Any idea how to fix it
021-04-21 12:34:39.593 ERROR 7175 --- [actor-tcp-nio-2] o.m.r.message.flow.AuthenticationFlow : Authentication failed
io.r2dbc.spi.R2dbcNonTransientResourceException: Connection unexpectedly closed
at org.mariadb.r2dbc.client.ClientBase.closedServlet(ClientBase.java:248) ~[r2dbc-mariadb-1.0.0.jar:1.0.0]
at reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:250) ~[reactor-core-3.4.5.jar:3.4.5]
at reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:259) ~[reactor-core-3.4.5.jar:3.4.5]
at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:142) ~[reactor-core-3.4.5.jar:3.4.5]
at reactor.netty.channel.FluxReceive.onInboundComplete(FluxReceive.java:401) ~[reactor-netty-core-1.0.6.jar:1.0.6]
at reactor.netty.channel.ChannelOperations.onInboundComplete(ChannelOperations.java:416) ~[reactor-netty-core-1.0.6.jar:1.0.6]
at reactor.netty.channel.ChannelOperations.terminate(ChannelOperations.java:470) ~[reactor-netty-core-1.0.6.jar:1.0.6]
at reactor.netty.channel.ChannelOperations.onInboundClose(ChannelOperations.java:431) ~[reactor-netty-core-1.0.6.jar:1.0.6]
at reactor.netty.channel.ChannelOperationsHandler.channelInactive(ChannelOperationsHandler.java:74) ~[reactor-netty-core-1.0.6.jar:1.0.6]
JSON typed column. Is there any work-around to do this?
@mp911de I think I experience connection leak similar to what is decribed here r2dbc/r2dbc-pool#54.
Background
I've run locally performance tests which do ~25 requests/second using local PostgreSQL server. After finishing of the tests I observe connection leak.
Same happens when tests are executed against production environment, but we use PgBouncer there, which also shows a connection leak
Observation for a single application instance running locally
These are our connection pool settings
CONNECTION_POOL_INITIAL_SIZE=0
CONNECTION_POOL_MAX_SIZE=21
CONNECTION_POOL_MAX_IDLE_TIME=1000
CONNECTION_POOL_MAX_CREATE_CONNECTION_TIME=10000
CONNECTION_POOL_ACQUIRE_RETRY=3
CONNECTION_POOL_MAX_ACQUIRE_TIME=7000Metrics exposed to Prometheus
r2dbc_pool_allocated_connections{name="connectionFactory",} 4.0
r2dbc_pool_idle_connections{name="connectionFactory",} 4.0
r2dbc_pool_acquired_connections{name="connectionFactory",} 0.0
r2dbc_pool_max_allocated_connections{name="connectionFactory",} 21.0
r2dbc_pool_pending_connections{name="connectionFactory",} 0.0
r2dbc_pool_max_pending_connections{name="connectionFactory",} 2.147483647E9I get from JMX (which is same a Prometheus)
AcquiredSize 0
AllocatedSize 4
IdleSize 4
MaxAllocatedSize 21
MaxPendingAcquireSize 2147483647
PendingAcquireSize 0I get from PostgreSQL server
active_conn | idle_conn
------------------------+----------------------
0 | 21
(1 row)Versions:
org.springframework.boot - 2.2.8.RELEASE
r2dbc-pool - 0.9.0.M1
Regards
@jarpz The SQL Server driver was built by the open source community as proof of concept for R2DBC with SQL Server. It is by no means aligned feature-wise with Microsoft's JDBC driver. You might want to reach out to your Microsoft sales rep so Microsoft can get involved into R2DBC driver development so you can use commercial features.
Thanks.
This https://github.com/r2dbc/.github/blob/main/CONTRIBUTING.adoc#submit-a-pull-request also states that
You must have the right to submit code changes. Is it only required to add Signed-off-by in commit message or I should do something in addition?
You must have the right to submit code changes is a legal requirement that basically means that you must be authorized to submit code. If you write the code on your own, then likely it's your own decision whether you authorize yourself. Moreover, if you write code on behalf of your company/employer/… and the code is owned by someone else, then you need permission from them to submit code changes. By adding Signed-off-by you express that all of that is given. You can read more about DCO at https://en.wikipedia.org/wiki/Developer_Certificate_of_Origin
Hello Guys, I am trying to read composite data type from R2DBC driver-
My table looks like below
CREATE TABLE test_udt
(
id uuid,
changes change_request_changes
)
CREATE TYPE change_request_changes AS
(
field_name text,
old_value text,
new_value text
);
I am trying to read the above table data using below code but it fails for decoding error
public Mono<TestUdt> findAll() {
return client.sql("select * from test_udt").map(row -> {
Change change = new Change(row.get("(changes).field_name", String.class), row.get("(changes).old_value", String.class), row.get("(changes).new_value", String.class));
return new TestUdt(row.get("id", UUID.class), null);
}
).first();
}
Please help me understand this. Thanks
Hello!
I try to insert a list of items to PostgreSQL database via R2DBC, but it does not complete. Configurations with example are below.
Could anyone get an advise why this code does not hit a database?
private Flux<OnlineChartItem> batchInsert() {
return Mono.from(chartInputConnectionFactory.create())
.log("start#1")
.flatMapMany(connection -> {
Statement statement = connection.createStatement(INSERT_INTO_ONLINE_CHARTS);
for (OnlineChartItem item : items) {
this.prepareInsertStatement(statement, item).add();
}
return Flux.from(statement.execute());
})
.thenMany(Flux.fromIterable(items));
}
@Bean
public ConnectionFactory chartInputConnectionFactory(R2dbcProperties r2dbcProperties) {
ConnectionFactoryOptions factoryOptions = ConnectionFactoryOptions.parse("r2dbc:pool:postgresql://localhost:5432/my_db")
.mutate()
.option(ConnectionFactoryOptions.USER, r2dbcProperties.getUsername())
.option(ConnectionFactoryOptions.PASSWORD, r2dbcProperties.getPassword())
.build();
ConnectionFactory connectionFactory = ConnectionFactories.get(factoryOptions);
R2dbcProperties.Pool pool = r2dbcProperties.getPool();
ConnectionPoolConfiguration configuration = ConnectionPoolConfiguration.builder()
.connectionFactory(connectionFactory)
.maxSize(20)
.initialSize(10)
.acquireRetry(Integer.MAX_VALUE)
.maxAcquireTime(Duration.ofSeconds(2))
.validationQuery("select 1")
.build();
return new ConnectionPool(configuration);
}
Hi everyone, im suddenly start getting mysql connection exceptions some how related to SSL:
dev.miku.r2dbc.mysql.client.MySqlConnectionClosedException: Connection unexpectedly closed
********
javax.net.ssl.SSLHandshakeException: No appropriate protocol (protocol is disabled or cipher suites are inappropriate)Any idea how to fix it ?
