Concurrent modification in R2dbcOffsetStore

[patriknw]

It seems to happen when failing the source provider stream, not for failures from the handler.

Found this from akka-projection-testing with a grouped handler.

It's probably not a critical issue since it is detected and restarted and thereby healing by itself.

[info] [2023-01-31 13:48:04,322] [WARN] [akka.stream.scaladsl.RestartWithBackoffSource] [RestartWithBackoffSource(akka://test)] [] [test-akka.actor.default-dispatcher-25] [] - Restarting stream due to failure [8]: akka.projection.testing.FailingEventsBySlicesProvider$FailingEventsBySlicesSourceProvider$$anon$1: Persistence id configurable|4b38cf7f-d2f2-484c-af4c-218189ae87a9 sequence nr 82 offset TimestampOffset(2023-01-31T12:47:32.963478Z,2023-01-31T12:48:02.984758Z,Map(configurable|4b38cf7f-d2f2-484c-af4c-218189ae87a9 -> 82)) Restart the stream!  MDC: {akkaAddress=akka://[email protected]:2551, akkaUid=27943348980114434, sourceThread=test-akka.actor.default-dispatcher-36, akkaSource=RestartWithBackoffSource(akka://test), sourceActorSystem=test, akkaTimestamp=12:48:04.322UTC}
[info] [2023-01-31 13:48:05,012] [WARN] [akka.stream.scaladsl.RestartWithBackoffSource] [RestartWithBackoffSource(akka://test)] [] [test-akka.actor.default-dispatcher-36] [] - Restarting stream due to failure [7]: akka.projection.testing.FailingEventsBySlicesProvider$FailingEventsBySlicesSourceProvider$$anon$1: Persistence id configurable|a07e996e-670f-4773-ad82-1a6646628e68 sequence nr 23 offset TimestampOffset(2023-01-31T12:47:41.935589Z,2023-01-31T12:48:03.665258Z,Map(configurable|a07e996e-670f-4773-ad82-1a6646628e68 -> 23)) Restart the stream!  MDC: {akkaAddress=akka://[email protected]:2551, akkaUid=27943348980114434, sourceThread=test-akka.actor.default-dispatcher-35, akkaSource=RestartWithBackoffSource(akka://test), sourceActorSystem=test, akkaTimestamp=12:48:05.012UTC}
[info] [2023-01-31 13:48:05,240] [WARN] [akka.stream.scaladsl.RestartWithBackoffSource] [RestartWithBackoffSource(akka://test)] [] [test-akka.actor.default-dispatcher-32] [] - Restarting stream due to failure [5]: akka.projection.testing.FailingEventsBySlicesProvider$FailingEventsBySlicesSourceProvider$$anon$1: Persistence id configurable|97be523c-2888-44b7-8b3d-912989619246 sequence nr 72 offset TimestampOffset(2023-01-31T12:47:36.159747Z,2023-01-31T12:48:03.898956Z,Map(configurable|97be523c-2888-44b7-8b3d-912989619246 -> 72)) Restart the stream!  MDC: {akkaAddress=akka://[email protected]:2551, akkaUid=27943348980114434, sourceThread=test-akka.actor.default-dispatcher-3, akkaSource=RestartWithBackoffSource(akka://test), sourceActorSystem=test, akkaTimestamp=12:48:05.240UTC}
[info] [2023-01-31 13:48:05,782] [WARN] [akka.stream.scaladsl.RestartWithBackoffSource] [RestartWithBackoffSource(akka://test)] [] [test-akka.actor.default-dispatcher-35] [] - Restarting stream due to failure [1]: akka.projection.testing.FailingEventsBySlicesProvider$FailingEventsBySlicesSourceProvider$$anon$1: Persistence id configurable|19d4d4a8-f3ed-4fd2-9315-abcf503b6b3f sequence nr 79 offset TimestampOffset(2023-01-31T12:47:36.307464Z,2023-01-31T12:48:02.804370Z,Map(configurable|19d4d4a8-f3ed-4fd2-9315-abcf503b6b3f -> 79)) Restart the stream!  MDC: {akkaAddress=akka://[email protected]:2551, akkaUid=27943348980114434, sourceThread=test-akka.actor.default-dispatcher-24, akkaSource=RestartWithBackoffSource(akka://test), sourceActorSystem=test, akkaTimestamp=12:48:05.782UTC}
[info] [2023-01-31 13:48:05,986] [WARN] [akka.stream.scaladsl.RestartWithBackoffSource] [RestartWithBackoffSource(akka://test)] [] [test-akka.actor.default-dispatcher-33] [] - Restarting stream due to failure [10]: akka.projection.testing.FailingEventsBySlicesProvider$FailingEventsBySlicesSourceProvider$$anon$1: Persistence id configurable|17ec9de3-d3b4-4bfe-9a7e-54bd981086f0 sequence nr 38 offset TimestampOffset(2023-01-31T12:47:28.651005Z,2023-01-31T12:48:04.636777Z,Map(configurable|17ec9de3-d3b4-4bfe-9a7e-54bd981086f0 -> 38)) Restart the stream!  MDC: {akkaAddress=akka://[email protected]:2551, akkaUid=27943348980114434, sourceThread=test-akka.actor.default-dispatcher-32, akkaSource=RestartWithBackoffSource(akka://test), sourceActorSystem=test, akkaTimestamp=12:48:05.986UTC}
[info] [2023-01-31 13:48:06,040] [ERROR] [akka.projection.r2dbc.internal.R2dbcProjectionImpl$R2dbcInternalProjectionState] [R2dbcProjectionImpl$R2dbcInternalProjectionState(akka://test)] [] [test-akka.actor.default-dispatcher-25] [] - [test-projection-id-0-768-895] Failed to process envelopes with offsets from [TimestampOffset(2023-01-31T12:47:35.965764Z,2023-01-31T12:48:02.804370Z,Map(configurable|07b05b45-e446-43dd-b7c8-5c1e9fbb1dee -> 74))] to [TimestampOffset(2023-01-31T12:47:36.073117Z,2023-01-31T12:48:02.804370Z,Map(configurable|b1446c13-462f-4e48-a25e-0c1f0eec1eee -> 74))]. Projection will stop as defined by recovery strategy.  MDC: {akkaAddress=akka://[email protected]:2551, akkaUid=27943348980114434, sourceThread=test-akka.actor.default-dispatcher-25, akkaSource=R2dbcProjectionImpl$R2dbcInternalProjectionState(akka://test), sourceActorSystem=test, akkaTimestamp=12:48:06.040UTC}
[info] java.lang.IllegalStateException: Unexpected concurrent modification of state from saveOffset.
[info] at akka.projection.r2dbc.internal.R2dbcOffsetStore.$anonfun$saveTimestampOffsetInTx$3(R2dbcOffsetStore.scala:480)
[info] at akka.projection.r2dbc.internal.R2dbcOffsetStore.$anonfun$saveTimestampOffsetInTx$3$adapted(R2dbcOffsetStore.scala:476)