Skip to content

rowsUpdated() returns more than one item. #207

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Closed
max-grigoriev opened this issue Dec 2, 2019 · 5 comments · Fixed by #230
Closed

rowsUpdated() returns more than one item. #207

max-grigoriev opened this issue Dec 2, 2019 · 5 comments · Fixed by #230
Labels
status: in-progress An issue that is currently being worked on type: bug A general bug
Milestone
0.8.1.RELEASE

Comments

@max-grigoriev
Copy link

Bug Report

Versions

  • Driver: 0.8.0.RELEASE
  • Database: Postgres RDS 9.6.11
  • Java: Corretto-8.232.09.1
  • OS: Linux Debian

Current Behavior

Sometimes after insert, I receive Source emitted more than one item. I can't find a reproducible scenario. This issue comes from time to time in our logs.

Stack trace 
java.lang.IndexOutOfBoundsException: Source emitted more than one item
	at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129)
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly trace from producer [reactor.core.publisher.MonoSingle] :
	reactor.core.publisher.Flux.singleOrEmpty
	io.r2dbc.postgresql.PostgresqlResult.getRowsUpdated(PostgresqlResult.java:69)
Error has been observed at the following site(s):
	|_     Flux.singleOrEmpty ⇢ at io.r2dbc.postgresql.PostgresqlResult.getRowsUpdated(PostgresqlResult.java:69)
	|_            Mono.handle ⇢ at io.r2dbc.postgresql.PostgresqlResult.getRowsUpdated(PostgresqlResult.java:70)
	|_        Flux.concatWith ⇢ at io.r2dbc.proxy.callback.CallbackHandlerSupport.proceedExecution(CallbackHandlerSupport.java:195)
	|_               Flux.map ⇢ at io.r2dbc.proxy.callback.CallbackHandlerSupport.proceedExecution(CallbackHandlerSupport.java:196)
	|_      Flux.doOnComplete ⇢ at io.r2dbc.proxy.callback.CallbackHandlerSupport.proceedExecution(CallbackHandlerSupport.java:207)
	|_         Flux.doOnError ⇢ at io.r2dbc.proxy.callback.CallbackHandlerSupport.proceedExecution(CallbackHandlerSupport.java:214)
	|_         Flux.doFinally ⇢ at io.r2dbc.proxy.callback.CallbackHandlerSupport.proceedExecution(CallbackHandlerSupport.java:217)
	|_           Flux.flatMap ⇢ at org.springframework.data.r2dbc.core.DefaultDatabaseClient$ExecuteSpecSupport.lambda$exchange$3(DefaultDatabaseClient.java:394)
	|_              Flux.next ⇢ at org.springframework.data.r2dbc.core.DefaultDatabaseClient$ExecuteSpecSupport.lambda$exchange$3(DefaultDatabaseClient.java:394)
	|_                        ⇢ at org.springframework.data.r2dbc.core.DefaultSqlResult$3.apply(DefaultSqlResult.java:90)
	|_                        ⇢ at org.springframework.data.r2dbc.core.DefaultDatabaseClient.doInConnection(DefaultDatabaseClient.java:1438)
	|_                        ⇢ at org.springframework.data.r2dbc.core.DefaultDatabaseClient.lambda$inConnection$2(DefaultDatabaseClient.java:159)
	|_         Mono.usingWhen ⇢ at org.springframework.data.r2dbc.core.DefaultDatabaseClient.inConnection(DefaultDatabaseClient.java:154)
	|_        Mono.onErrorMap ⇢ at org.springframework.data.r2dbc.core.DefaultDatabaseClient.inConnection(DefaultDatabaseClient.java:161)
	|_                        ⇢ at org.springframework.data.r2dbc.core.DefaultFetchSpec.rowsUpdated(DefaultFetchSpec.java:89)
	|_                        ⇢ at org.springframework.data.r2dbc.core.DefaultSqlResult.rowsUpdated(DefaultSqlResult.java:148)

Table schema

Input Code 
-- your SQL here;
CREATE TABLE IF NOT EXISTS room_contact
            (
                id                     bigserial PRIMARY KEY,
                active                 boolean   NOT NULL DEFAULT true,
                owner_person_uid       bigint    NOT NULL,
                owner_org_uid          bigint    NOT NULL,
                person_uid             bigint    NOT NULL,
                person_org_uid         bigint    NOT NULL,
                person_first_last_name varchar(255),
                person_last_first_name varchar(255),
                origin_code            smallint  NOT NULL,
                count                  integer  NOT NULL DEFAULT 0,
                created                timestamp NOT NULL,
                last_contacted         timestamp NOT NULL,
                version                smallint  NOT NULL DEFAULT 0,
                CONSTRAINT room_contact_users_idx UNIQUE (owner_person_uid, owner_org_uid, person_uid, person_org_uid)
            );

Steps to reproduce

Input Code 
// your code here;
databaseClient
                        .execute("INSERT INTO room_contact(active, owner_person_uid, owner_org_uid, " +
                                " person_uid, person_org_uid, person_first_last_name, person_last_first_name, " +
                                " origin_code, count, created, last_contacted, version) VALUES($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) " +
                                " ON CONFLICT ON CONSTRAINT room_contact_users_idx DO UPDATE " +
                                " SET active=$13, person_first_last_name = $14, person_last_first_name=$15, count=$16, last_contacted=$17")
                        .bind(0, contact.isActive())
                        .bind(1, contact.getOwnerPersonUid())
                        .bind(2, contact.getOwnerOrgUid())
                        .bind(3, contact.getPersonUid())
                        .bind(4, contact.getPersonOrgUid())
                        .bind(5, contact.getPersonFirstLastName())
                        .bind(6, contact.getPersonLastFirstName())
                        .bind(7, contact.getOriginCode().getContactOriginCode())
                        .bind(8, contact.getCount())
                        .bind(9, contact.getCreated())
                        .bind(10, contact.getLastContacted())
                        .bind(11, contact.getVersion())
                        .bind(12, contact.isActive())
                        .bind(13, contact.getPersonFirstLastName())
                        .bind(14, contact.getPersonLastFirstName())
                        .bind(15, contact.getCount())
                        .bind(16, contact.getLastContacted())
                        .fetch()
                        .rowsUpdated()

Expected behavior/code

Possible Solution

Additional context
@mp911de mp911de added the type: bug A general bug label Dec 2, 2019
@mp911de
Copy link
Collaborator

mp911de commented Dec 2, 2019

Thanks for report. Can you attach a log on at least DEBUG level of the failing INSERT operation? It looks like INSERT … ON CONFLICT UPDATE generates two CommandComplete messages.

@Squiry
Copy link
Collaborator

Squiry commented Dec 2, 2019 •
edited
Loading

documentation

Therefore, an Execute phase is always terminated by the appearance of exactly one of these messages: CommandComplete, EmptyQueryResponse (if the portal was created from an empty query string), ErrorResponse, or PortalSuspended.

@max-grigoriev
Copy link
Author

Thanks for report. Can you attach a log on at least DEBUG level of the failing INSERT operation? It looks like INSERT … ON CONFLICT UPDATE generates two CommandComplete messages.

I can't enable debug level on this host. And I can't reproduce it on localhost. :(

@mp911de
Copy link
Collaborator

mp911de commented Dec 13, 2019

I tried to reproduce the issue with no luck. Either the INSERT command sends an unexpected response, or (that's what I assume) there are non-consumed frames on the protocol stack. Having a reliable reproducer would allow us to trace down the issue.

@mcgin
Copy link

mcgin commented Jan 15, 2020

I've managed to reproduce and here is the debug output from around the error from ReactorNettyClient

18948 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18948 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18948 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: CommandComplete{command=INSERT, rowId=0, rows=1}
18948 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7919', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_0'}
18949 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: CloseComplete{}
18949 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7919', type=PORTAL}
18949 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18949 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7919', rows=0}
18949 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7920', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_0'}
18949 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7920', type=PORTAL}
18949 [parallel-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7921', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_0'}
18949 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7920', rows=0}
18949 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7920', type=PORTAL}
18949 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7921', type=PORTAL}
18949 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18949 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7921', rows=0}
18949 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7919', type=PORTAL}
18949 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7921', type=PORTAL}
18949 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18949 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18949 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18950 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Response: ParseComplete{}
18950 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Response: ParseComplete{}
18950 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Response: BindComplete{}
18950 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Response: ParameterDescription{parameters=[2950]}
18950 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: BindComplete{}
18950 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Response: ParameterDescription{parameters=[2950]}
18950 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18951 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18951 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18951 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: CommandComplete{command=INSERT, rowId=0, rows=1}
18951 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18951 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Response: CommandComplete{command=INSERT, rowId=0, rows=1}
18951 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7922', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_1'}
18951 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: CloseComplete{}
18951 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7923', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_0'}
18951 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7922', type=PORTAL}
18951 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18951 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Response: CloseComplete{}
18951 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7923', type=PORTAL}
18951 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7922', rows=0}
18951 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7923', rows=0}
18951 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18951 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7923', type=PORTAL}
18951 [parallel-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7924', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_0'}
18951 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7922', type=PORTAL}
18951 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18951 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18951 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7924', type=PORTAL}
18951 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7925', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_0'}
18951 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7924', rows=0}
18951 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18951 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7924', type=PORTAL}
18951 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7925', type=PORTAL}
18951 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18951 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7925', rows=0}
18951 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7925', type=PORTAL}
18951 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18951 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18952 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Response: BindComplete{}
18952 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18952 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Response: CommandComplete{command=INSERT, rowId=0, rows=1}
18952 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Response: CloseComplete{}
18952 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18952 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: ParseComplete{}
18952 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Response: BindComplete{}
18952 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Response: BindComplete{}
18952 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: ParameterDescription{parameters=[2950]}
18952 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18952 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18952 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18952 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Response: CommandComplete{command=INSERT, rowId=0, rows=1}
18952 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Response: CommandComplete{command=INSERT, rowId=0, rows=1}
18952 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7926', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_1'}
18953 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Response: BindComplete{}
18953 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18953 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Response: CommandComplete{command=INSERT, rowId=0, rows=1}
18953 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Response: CloseComplete{}
18953 [parallel-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7927', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_1'}
18953 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18953 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7926', type=PORTAL}
18953 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7927', type=PORTAL}
18953 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7926', rows=0}
18953 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7927', rows=0}
18953 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7926', type=PORTAL}
18953 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7927', type=PORTAL}
18953 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18953 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18953 [parallel-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7928', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_1'}
18953 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18953 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7928', type=PORTAL}
18953 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7928', rows=0}
18953 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7928', type=PORTAL}
18953 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18953 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Response: BindComplete{}
18953 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18953 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Response: CommandComplete{command=INSERT, rowId=0, rows=1}
18953 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Response: CloseComplete{}
18953 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18954 [reactor-tcp-nio-5] ERROR email event insert error - Source emitted more than one item
java.lang.IndexOutOfBoundsException: Source emitted more than one item
	at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129)
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly trace from producer [reactor.core.publisher.MonoSingle] :
	reactor.core.publisher.Flux.singleOrEmpty
	io.r2dbc.postgresql.PostgresqlResult.getRowsUpdated(PostgresqlResult.java:68)
Error has been observed at the following site(s):
	|_ Flux.singleOrEmpty ⇢ at io.r2dbc.postgresql.PostgresqlResult.getRowsUpdated(PostgresqlResult.java:68)
	|_        Mono.handle ⇢ at io.r2dbc.postgresql.PostgresqlResult.getRowsUpdated(PostgresqlResult.java:69)
	|_    Flux.concatWith ⇢ at io.r2dbc.proxy.callback.CallbackHandlerSupport.proceedExecution(CallbackHandlerSupport.java:195)
	|_           Flux.map ⇢ at io.r2dbc.proxy.callback.CallbackHandlerSupport.proceedExecution(CallbackHandlerSupport.java:196)
	|_  Flux.doOnComplete ⇢ at io.r2dbc.proxy.callback.CallbackHandlerSupport.proceedExecution(CallbackHandlerSupport.java:207)
	|_     Flux.doOnError ⇢ at io.r2dbc.proxy.callback.CallbackHandlerSupport.proceedExecution(CallbackHandlerSupport.java:214)
	|_     Flux.doFinally ⇢ at io.r2dbc.proxy.callback.CallbackHandlerSupport.proceedExecution(CallbackHandlerSupport.java:217)
	|_       Flux.flatMap ⇢ at io.r2dbc.client.Update.execute(Update.java:115)
	|_          Flux.then ⇢ at io.r2dbc.client.util.ReactiveUtils.lambda$appendError$0(ReactiveUtils.java:49)
Stack trace:
		at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129)
		at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:121)
		at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.onNext(FluxFilterFuseable.java:112)
		at reactor.core.publisher.FluxPeekFuseable$PeekFuseableConditionalSubscriber.onNext(FluxPeekFuseable.java:495)
		at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drainRegular(FluxWindowPredicate.java:650)
		at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drain(FluxWindowPredicate.java:728)
		at reactor.core.publisher.FluxWindowPredicate$WindowFlux.onNext(FluxWindowPredicate.java:770)
		at reactor.core.publisher.FluxWindowPredicate$WindowPredicateMain.onNext(FluxWindowPredicate.java:249)
		at reactor.core.publisher.FluxHandleFuseable$HandleFuseableSubscriber.onNext(FluxHandleFuseable.java:178)
		at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber.onNext(FluxFilterFuseable.java:330)
		at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:242)
		at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:242)
		at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drainRegular(FluxWindowPredicate.java:650)
		at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drain(FluxWindowPredicate.java:728)
		at reactor.core.publisher.FluxWindowPredicate$WindowFlux.onNext(FluxWindowPredicate.java:770)
		at reactor.core.publisher.FluxWindowPredicate$WindowPredicateMain.onNext(FluxWindowPredicate.java:249)
		at reactor.core.publisher.FluxHandleFuseable$HandleFuseableSubscriber.onNext(FluxHandleFuseable.java:178)
		at reactor.core.publisher.FluxMapFuseable$MapFuseableConditionalSubscriber.onNext(FluxMapFuseable.java:287)
		at reactor.core.publisher.FluxMapFuseable$MapFuseableConditionalSubscriber.onNext(FluxMapFuseable.java:287)
		at reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:218)
		at reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:351)
		at reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:348)
		at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:89)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
		at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:326)
		at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:313)
		at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:427)
		at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:281)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
		at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1422)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
		at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:931)
		at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163)
		at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:700)
		at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:635)
		at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:552)
		at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:514)
		at io.netty.util.concurrent.SingleThreadEventExecutor$6.run(SingleThreadEventExecutor.java:1050)
		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:830)
18954 [reactor-tcp-nio-5] ERROR email event insert error - Error message: Source emitted more than one item
--
	at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129)
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly trace from producer [reactor.core.publisher.MonoSingle] :
	reactor.core.publisher.Flux.singleOrEmpty
	io.r2dbc.postgresql.PostgresqlResult.getRowsUpdated(PostgresqlResult.java:68)
Error has been observed at the following site(s):
	|_ Flux.singleOrEmpty ⇢ at io.r2dbc.postgresql.PostgresqlResult.getRowsUpdated(PostgresqlResult.java:68)
	|_        Mono.handle ⇢ at io.r2dbc.postgresql.PostgresqlResult.getRowsUpdated(PostgresqlResult.java:69)
	|_    Flux.concatWith ⇢ at io.r2dbc.proxy.callback.CallbackHandlerSupport.proceedExecution(CallbackHandlerSupport.java:195)
	|_           Flux.map ⇢ at io.r2dbc.proxy.callback.CallbackHandlerSupport.proceedExecution(CallbackHandlerSupport.java:196)
	|_  Flux.doOnComplete ⇢ at io.r2dbc.proxy.callback.CallbackHandlerSupport.proceedExecution(CallbackHandlerSupport.java:207)
	|_     Flux.doOnError ⇢ at io.r2dbc.proxy.callback.CallbackHandlerSupport.proceedExecution(CallbackHandlerSupport.java:214)
	|_     Flux.doFinally ⇢ at io.r2dbc.proxy.callback.CallbackHandlerSupport.proceedExecution(CallbackHandlerSupport.java:217)
	|_       Flux.flatMap ⇢ at io.r2dbc.client.Update.execute(Update.java:115)
	|_          Flux.then ⇢ at io.r2dbc.client.util.ReactiveUtils.lambda$appendError$0(ReactiveUtils.java:49)
Stack trace:
		at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129)
		at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:121)
		at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.onNext(FluxFilterFuseable.java:112)
		at reactor.core.publisher.FluxPeekFuseable$PeekFuseableConditionalSubscriber.onNext(FluxPeekFuseable.java:495)
		at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drainRegular(FluxWindowPredicate.java:650)
		at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drain(FluxWindowPredicate.java:728)
		at reactor.core.publisher.FluxWindowPredicate$WindowFlux.onNext(FluxWindowPredicate.java:770)
		at reactor.core.publisher.FluxWindowPredicate$WindowPredicateMain.onNext(FluxWindowPredicate.java:249)
		at reactor.core.publisher.FluxHandleFuseable$HandleFuseableSubscriber.onNext(FluxHandleFuseable.java:178)
		at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber.onNext(FluxFilterFuseable.java:330)
		at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:242)
		at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:242)
		at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drainRegular(FluxWindowPredicate.java:650)
		at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drain(FluxWindowPredicate.java:728)
		at reactor.core.publisher.FluxWindowPredicate$WindowFlux.onNext(FluxWindowPredicate.java:770)
		at reactor.core.publisher.FluxWindowPredicate$WindowPredicateMain.onNext(FluxWindowPredicate.java:249)
		at reactor.core.publisher.FluxHandleFuseable$HandleFuseableSubscriber.onNext(FluxHandleFuseable.java:178)
		at reactor.core.publisher.FluxMapFuseable$MapFuseableConditionalSubscriber.onNext(FluxMapFuseable.java:287)
		at reactor.core.publisher.FluxMapFuseable$MapFuseableConditionalSubscriber.onNext(FluxMapFuseable.java:287)
		at reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:218)
		at reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:351)
		at reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:348)
		at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:89)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
		at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:326)
		at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:313)
		at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:427)
		at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:281)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
		at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1422)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
		at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:931)
		at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163)
		at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:700)
		at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:635)
		at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:552)
		at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:514)
		at io.netty.util.concurrent.SingleThreadEventExecutor$6.run(SingleThreadEventExecutor.java:1050)
		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:830)
18954 [reactor-tcp-nio-5] ERROR email event insert error - Error message: Source emitted more than one item
Error Type: class java.lang.IndexOutOfBoundsException
18954 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Response: CloseComplete{}
18954 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18954 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Response: CloseComplete{}
18954 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18954 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: BindComplete{}
18954 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18954 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Response: BindComplete{}
18954 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: CommandComplete{command=INSERT, rowId=0, rows=1}
18954 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18954 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Response: CommandComplete{command=INSERT, rowId=0, rows=1}
18954 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: CloseComplete{}
18954 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18954 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Response: CloseComplete{}
18954 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18954 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Response: BindComplete{}
18955 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Response: BindComplete{}
18955 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18955 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Response: CommandComplete{command=INSERT, rowId=0, rows=1}
18955 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18955 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Response: CommandComplete{command=INSERT, rowId=0, rows=1}
18955 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Response: CloseComplete{}
18955 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18955 [parallel-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Parse{name='S_1', parameters=[2950], query='INSERT INTO email.event_store  (event_id)  VALUES ($1) '}
18955 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Response: CloseComplete{}
18955 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Response: BindComplete{}
18955 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18955 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18955 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Response: CommandComplete{command=INSERT, rowId=0, rows=1}
18955 [reactor-tcp-nio-7] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='S_1', type=STATEMENT}
18955 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Response: CloseComplete{}
18955 [reactor-tcp-nio-7] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18955 [parallel-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Parse{name='S_1', parameters=[2950], query='INSERT INTO identity.event_store  (event_id)  VALUES ($1) '}
18955 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18955 [reactor-tcp-nio-8] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='S_1', type=STATEMENT}
18955 [reactor-tcp-nio-8] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18955 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: BindComplete{}
18955 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18955 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: CommandComplete{command=INSERT, rowId=0, rows=1}
18955 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: CloseComplete{}
18955 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18956 [reactor-tcp-nio-7] DEBUG i.r.p.client.ReactorNettyClient - Response: ParseComplete{}
18956 [reactor-tcp-nio-8] DEBUG i.r.p.client.ReactorNettyClient - Response: ParseComplete{}
18957 [reactor-tcp-nio-8] DEBUG i.r.p.client.ReactorNettyClient - Response: ParameterDescription{parameters=[2950]}
18957 [reactor-tcp-nio-8] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18956 [reactor-tcp-nio-7] DEBUG i.r.p.client.ReactorNettyClient - Response: ParameterDescription{parameters=[2950]}
18957 [reactor-tcp-nio-8] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7929', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_1'}
18957 [reactor-tcp-nio-7] DEBUG i.r.p.client.ReactorNettyClient - Response: NoData{}
18957 [reactor-tcp-nio-8] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7929', type=PORTAL}
18957 [reactor-tcp-nio-7] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7930', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_1'}
18957 [parallel-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7931', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_1'}
18957 [reactor-tcp-nio-7] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7930', type=PORTAL}
18957 [reactor-tcp-nio-8] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7929', rows=0}
18957 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7931', type=PORTAL}
18957 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7931', rows=0}
18957 [reactor-tcp-nio-7] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7930', rows=0}
18957 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7931', type=PORTAL}
18957 [reactor-tcp-nio-8] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7929', type=PORTAL}
18958 [parallel-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7932', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_1'}
18958 [reactor-tcp-nio-3] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18958 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7932', type=PORTAL}
18958 [reactor-tcp-nio-7] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7930', type=PORTAL}
18958 [reactor-tcp-nio-8] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18958 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7932', rows=0}
18958 [reactor-tcp-nio-8] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18958 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7932', type=PORTAL}
18958 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18958 [reactor-tcp-nio-7] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18958 [reactor-tcp-nio-7] DEBUG i.r.p.client.ReactorNettyClient - Response: ReadyForQuery{transactionStatus=IDLE}
18959 [parallel-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Parse{name='S_1', parameters=[2950], query='INSERT INTO identity.event_store  (event_id)  VALUES ($1) '}
18959 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='S_1', type=STATEMENT}
18959 [reactor-tcp-nio-5] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18959 [parallel-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Parse{name='S_1', parameters=[2950], query='INSERT INTO email.event_store  (event_id)  VALUES ($1) '}
18960 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='S_1', type=STATEMENT}
18960 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18961 [parallel-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7933', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_0'}
18962 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7933', type=PORTAL}
18962 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7933', rows=0}
18962 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7933', type=PORTAL}
18962 [parallel-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7934', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_0'}
18962 [reactor-tcp-nio-6] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18962 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7934', type=PORTAL}
18962 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7934', rows=0}
18962 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7934', type=PORTAL}
18963 [reactor-tcp-nio-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18963 [parallel-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7935', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_0'}
18963 [parallel-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7936', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_0'}
18963 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7935', type=PORTAL}
18963 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7935', rows=0}
18963 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7936', type=PORTAL}
18963 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7935', type=PORTAL}
18963 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7936', rows=0}
18963 [reactor-tcp-nio-4] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18963 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7936', type=PORTAL}
18964 [reactor-tcp-nio-2] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}
18965 [parallel-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7937', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_1'}
18965 [parallel-1] DEBUG i.r.p.client.ReactorNettyClient - Request:  Bind{name='B_7938', parameterFormats=[FORMAT_TEXT], parameters=[CompositeByteBuf(ridx: 0, widx: 36, cap: 36, components=1)], resultFormats=[], source='S_1'}
18965 [reactor-tcp-nio-7] DEBUG i.r.p.client.ReactorNettyClient - Request:  Describe{name='B_7937', type=PORTAL}
18965 [reactor-tcp-nio-7] DEBUG i.r.p.client.ReactorNettyClient - Request:  Execute{name='B_7937', rows=0}
18965 [reactor-tcp-nio-7] DEBUG i.r.p.client.ReactorNettyClient - Request:  Close{name='B_7937', type=PORTAL}
18965 [reactor-tcp-nio-7] DEBUG i.r.p.client.ReactorNettyClient - Request:  Sync{}

@mp911de mp911de added this to the 0.8.1.RELEASE milestone Jan 21, 2020
@mp911de mp911de added the status: in-progress An issue that is currently being worked on label Jan 21, 2020
@Squiry Squiry mentioned this issue Jan 21, 2020
Merged
mp911de added a commit that referenced this issue Jan 22, 2020
@mp911de
Polishing
8c34f54 
Rename ResponseReceiver to Conversation to reflect how the driver works internally. Add Javadoc to outline conversation semantics and to explain the test assertions.

[#207][#230]
mp911de pushed a commit that referenced this issue Jan 22, 2020
@Squiry @mp911de
Fix for concurrent exchanging on one client
5af51a9 
[#207][#230]
mp911de added a commit that referenced this issue Jan 22, 2020
@mp911de
Polishing
3c91ca8 
Rename ResponseReceiver to Conversation to reflect how the driver works internally. Add Javadoc to outline conversation semantics and to explain the test assertions.

[#207][#230]
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Assignees
No one assigned
Labels
status: in-progress An issue that is currently being worked on type: bug A general bug
Projects
None yet
Milestone
0.8.1.RELEASE
Development

Successfully merging a pull request may close this issue.

Fix for concurrent exchanging on one client Squiry/r2dbc-postgresql
4 participants
@mcgin @max-grigoriev @mp911de @Squiry