[Question posted by a user on YugabyteDB Community Slack]
I am currently using YugabyteDB with the reactive Postgres driver (io.r2dbc:r2dbc-postgresql), and I am facing some intermittent issues like this one, as you can see in the stack trace below.
I was told that the Postgres driver may not deal correctly with the Yugabyte load balancing, which maybe is leading me to this problem, and then maybe the actual Yugabyte driver would deal properly with such scenarios.
However, I am using a reactive code, which means I need an R2DBC driver, and I did not find any official R2DBC Yugabyte driver.
Do you think a more appropriate driver would really solve such problem? If so, is there any other R2DBC driver that would be more suitable for my purpose here? If not, do you have any suggestions to solve the problem below?
23:37:07.239 [reactor-tcp-epoll-1] WARN i.r.p.client.ReactorNettyClient - Error: SEVERITY_LOCALIZED=ERROR, SEVERITY_NON_LOCALIZED=ERROR, CODE=40001, MESSAGE=Query error: Restart read required at: { read: { physical: 1639179428477618 } local_limit: { physical: 1639179428477618 } global_limit: <min> in_txn_limit: <max> serial_no: 0 }, FILE=pg_yb_utils.c, LINE=333, ROUTINE=HandleYBStatusAtErrorLevel
23:37:07.247 [reactor-kafka-sender-1609501721] ERROR reactor.core.publisher.Operators - Operator called default onErrorDropped
reactor.core.Exceptions$ErrorCallbackNotImplemented: org.jooq.exception.DataAccessException: SQL [update "core"."videos" set "status" = $1 where "core"."videos"."media_key" = $2 returning "core"."videos"."user_id", "core"."videos"."media_key", "core"."videos"."post_id", "core"."videos"."status"]; Query error: Restart read required at: { read: { physical: 1639179428477618 } local_limit: { physical: 1639179428477618 } global_limit: <min> in_txn_limit: <max> serial_no: 0 }
Caused by: org.jooq.exception.DataAccessException: SQL [update "core"."videos" set "status" = $1 where "core"."videos"."media_key" = $2 returning "core"."videos"."user_id", "core"."videos"."media_key", "core"."videos"."post_id", "core"."videos"."status"]; Query error: Restart read required at: { read: { physical: 1639179428477618 } local_limit: { physical: 1639179428477618 } global_limit: <min> in_txn_limit: <max> serial_no: 0 }
at org.jooq.impl.Tools.translate(Tools.java:2978)
at org.jooq.impl.Tools.translate(Tools.java:2962)
at org.jooq.impl.R2DBC$Forwarding.onError(R2DBC.java:236)
at reactor.core.publisher.StrictSubscriber.onError(StrictSubscriber.java:106)
at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onError(ScopePassingSpanSubscriber.java:96)
at reactor.core.publisher.FluxHandle$HandleSubscriber.onError(FluxHandle.java:203)
at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onError(MonoFlatMapMany.java:255)
at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onError(ScopePassingSpanSubscriber.java:96)
at reactor.core.publisher.FluxHandleFuseable$HandleFuseableSubscriber.onNext(FluxHandleFuseable.java:191)
at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber.onNext(FluxFilterFuseable.java:337)
at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:107)
at reactor.core.publisher.FluxPeekFuseable$PeekConditionalSubscriber.onNext(FluxPeekFuseable.java:854)
at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:89)
at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:200)
at io.r2dbc.postgresql.util.FluxDiscardOnCancel$FluxDiscardOnCancelSubscriber.onNext(FluxDiscardOnCancel.java:86)
at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:89)
at reactor.core.publisher.FluxCreate$BufferAsyncSink.drain(FluxCreate.java:793)
at reactor.core.publisher.FluxCreate$BufferAsyncSink.next(FluxCreate.java:718)
at reactor.core.publisher.FluxCreate$SerializedFluxSink.next(FluxCreate.java:154)
at io.r2dbc.postgresql.client.ReactorNettyClient$Conversation.emit(ReactorNettyClient.java:735)
at io.r2dbc.postgresql.client.ReactorNettyClient$BackendMessageSubscriber.emit(ReactorNettyClient.java:986)
at io.r2dbc.postgresql.client.ReactorNettyClient$BackendMessageSubscriber.onNext(ReactorNettyClient.java:860)
at io.r2dbc.postgresql.client.ReactorNettyClient$BackendMessageSubscriber.onNext(ReactorNettyClient.java:767)
at reactor.core.publisher.FluxHandle$HandleSubscriber.onNext(FluxHandle.java:119)
at reactor.core.publisher.FluxPeekFuseable$PeekConditionalSubscriber.onNext(FluxPeekFuseable.java:854)
at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:89)
at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:89)
at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
at reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:279)
at reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:388)
at reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:404)
at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:93)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:795)
at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:480)
at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:378)
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
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:831)
Caused by: io.r2dbc.postgresql.ExceptionFactory$PostgresqlRollbackException: Query error: Restart read required at: { read: { physical: 1639179428477618 } local_limit: { physical: 1639179428477618 } global_limit: <min> in_txn_limit: <max> serial_no: 0 }
at io.r2dbc.postgresql.ExceptionFactory.createException(ExceptionFactory.java:72)
at io.r2dbc.postgresql.ExceptionFactory.handleErrorResponse(ExceptionFactory.java:111)
at reactor.core.publisher.FluxHandleFuseable$HandleFuseableSubscriber.onNext(FluxHandleFuseable.java:169)
... 43 common frames omitted
Thank you very much in advance!
The exception stack trace is related to Restart read errors. Currently, YugabyteDB supports only optimistic locking with SNAPSHOT isolation level which means whenever there is conflict on concurrent access, the driver will throw restart read errors like below:
Caused by: io.r2dbc.postgresql.ExceptionFactory$PostgresqlRollbackException: Query error: Restart read required at: { read: { physical: 1639179428477618 } local_limit: { physical: 1639179428477618 } global_limit: <min> in_txn_limit: <max> serial_no: 0 }
You will need to handle the Rollback exception and retry the operation. It's on the roadmap for YugabyteDB to support REPEATABLE READS with pessimistic locking to avoid transaction retries, until the feature is available you’ll need to retry the transaction on the client-side.