Artifical back pressure on JDBC projections

[rsmidt]

Hey team,

I'm currently running into an issue while rolling out a new sharded projection (events by tag; JDBC with Postgres). It's basically processing envelopes so fast that the DB CPU becomes saturated, resulting in other parts of the application throttling.

I've tried the following things so far:

1. Artificially throttling each projector instance

I thought that I could somehow slow down the projector requesting new events from the journal by sleeping somewhere. I can't really do it in the handler because this would block the JDBC transaction. Therefore, I tried to do it right before session creation, in the session creator callback:

return JdbcProjection.groupedWithin(
    ProjectionId.of(PROJECTION_NAME, tag),
    sourceProvider,
    {
        // The session creation is called on each run of the projector to create the session (transaction).
        // By running sleep here, we can delay the run of the projector without blocking the DB connection/transaction.
        maybeSleep()
        HibernateSession(...)
    },
    { },
    system
)

Unfortunately, this only works partially because it blocks the full thread (via LockSupport.parkNanos()). So other, unrelated projections are also throttled accidentally (I noticed ever-increasing projection lag). I tried to fix that by scheduling all projector instances on a dedicated pool, but from the config, this does not seem possible as there is only one dispatcher for all projections (reference).

2. Reducing the number of concurrent projector instances

We're having > 100 tags, so I thought if I manually restricted the number of instances to, let's say, 20 and had these 20 do more than just one tag, it could keep the impact low. However, I'm not seeing any API to do that.

Am I maybe missing something crucial here?

[pjfanning]

Pekko-Projections is not an area that I claim expertise in. I know that it is not an area that the active Pekko volunteers generally look at a great deal and it doesn't appear to be heavily used.

Hopefully, someone with more Projections experience may be able to help. Here are some devil's advocate suggestions that you might consider.

Have a checked your DB instance to see if its performance issues can be addressed? Maybe some DB tuning could help.
Have you considered testing other persistence alternatives like Cassandra or DynamoDB?
We support R2DBC as well as JDBC - if you are stuck with Postgres then R2DBC may be worth checking out.

[rsmidt]

@pjfanning, thanks for the suggestions. Yes, we would love to simply migrate to R2DBC, but it's out of scope for now. For greenfield projects, we will definitely consider R2DBC.

[ptrdom]

For eventsByTag I think we do not provide any utilities to control parallelism in a manner that would help you, but with eventsBySlices we do, though that is only implemented in R2DBC plugin. If it is possible for your project I would suggest migrating to R2DBC plugin, as eventsBySlices is a newer and improved solution over eventsByTag, and I feel that we are overall more likely to provide better support and future improvements to R2DBC plugin.

[ptrdom]

@pjfanning Maybe we should consider adding a message to JDBC plugin that directs/advises users towards considering R2DBC? I have a feeling that developers get naturally guided towards a poorer/outdated solution because of its familiar name.

[pjfanning]

I guess we could start by adding statements like this to the docs. We could later follow up with logging stuff like this.

[rsmidt]

@ptrdom, thanks for the explanations. When we implemented our current service, we didn't notice the R2DBC variant or it wasn't available back then. That's why we stuck with JDBC. As mentioned in another comment, we would love to migrate away from JDBC, but can't really afford it currently.

[mdedetrich]

Might be a bit of a radical idea, but as a long term solution it seems like r2dbc in conjunction with pekko streams would properly solve this issue (r2dbc implements the reactive streams protocol which mean converting them to pekko-streams is trivial)?

There is already pekko-projection-r2dbc albeit its not implemented via pure streams (instead it materializes the transactions prematurely into Future). There is an opening for Pekko 2.0.0, we could add an r2dbc module to pekko connectors and pekko-projection-r2dbc could then use that to implement a pure backpressured/streamed implementation for pekko-projections

[mdedetrich]

But that is the whole point of r2dbc/backpressure, as long as the entire flow is a stream then r2dbc will automatically slow itself down so it won't overwhelm the database. Right now even if you use postgres-r2dbc projection adaptor it won't do this because we just use the future API, which means it will send out transactions/statements as fast as the connection can handle.

Of course I can't say for certain if it will help this scenario as I am not aware of the details, but the whole point of bi-directional backpressure (which is what pekko-streams and r2dbc implement) is to avoid manually throttling as both the consumers and products for each stream in the pipeline signal how fast they can produce and/or how fast they can consume

[ptrdom]

I understand the described problem as multiple separate streams that share a single resource - database, and one of those streams works so well that it does not signal backpressure and completely saturates the database, making so that the other streams cannot work properly. How would backpressure work in this case? Would all the other streams start signaling to this stream that works so well that they are not receiving any elements and that this well working stream should slow down? As far as I know that is not the backpressure we are talking about here, and it would not help, or would it?

[mdedetrich]

How would backpressure work in this case? Would all the other streams start signaling to this stream that works so well that they are not receiving any elements and that this well working stream should slow down?

Yes, if you use the pekko-streams API this is exactly what fan-in/fan-out API's do.

My point is that current implementation is not even a stream at all, look at the API at https://github.com/apache/pekko-persistence-r2dbc/blob/beee0d1bfd0cd46969e8ab940cf8244e7f07b5b3/projection/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcOffsetStore.scala, as you can see its all Future based which means that we essentially lose all of the dynamic backpressuring goodness that pekko/reactive streams/r2dbc gives us

[ptrdom]

Yes, if you use the pekko-streams API this is exactly what fan-in/fan-out API's do.

Fan-in/fan-out is still about connected streams, here I do not think the streams are connected.

What you linked in R2DBC is offset store, not journal queries, but journal queries also operate on Future too. They still convert to a SourceProvider down the line and then the backpressure starts to work. The end-to-end streaming is likely not an option at a query level, because queries do flip flop between live data and backtracking between batches, so while I guess you could do live query as a stream, backtracking would still need to be done in batches at certain intervals.

[mdedetrich]

Fan-in/fan-out is still about connected streams, here I do not think the streams are connected.

Yes thats the entire issue I am pointing out

What you linked in R2DBC is offset store, not journal queries, but journal queries also operate on Future too.

True, not that familiar with projection but I remember the few times I worked on it I noticed that projection-r2dbc was entirely Future based

They still convert to a SourceProvider down the line and then the backpressure starts to work. The end-to-end streaming is likely not an option at a query level, because queries do flip flop between live data and backtracking between batches, so while I guess you could do live query as a stream, backtracking would still need to be done in batches at certain intervals.

As I said I don't really know the details as I haven't worked with projection that much specifically, but if we are talking about pekko/reactive streams they are only as good as the weakest link, or to be put differently the minute you have some part of the stream thats disconnected (i.e. not a stream anymore) then you can lose a lot of the benefits, and jdbc is definitely not a streaming safe API.

In any case, I am not sure what is meant be backtracking here (ill look into it) but on the surface I am not entirely convinced that it can't be represented as a stream that can be automatically backpressured. r2dbc projection not being a pure stream is kind of an issue design wise anyways

[ptrdom]

@rsmidt Maybe sourceProvider could be wrapped to add throttling to the source?

JdbcProjection.atLeastOnce(
    ProjectionId.of(PROJECTION_NAME, tag),
    new SourceProvider[Offset, Envelope] {
        val wrapped = sourceProvider

        override def source(offset: () => Future[Option[Long]]): Future[Source[Envelope, NotUsed]] =
          wrapped.source(offset).map(_.throttle(1, 1.second))

        override def extractOffset(envelope: Envelope): Long = wrapped.extractOffset(envelope)

        override def extractCreationTime(envelope: Envelope): Long = wrapped.extractOffset(envelope)
      },
    ...
)

[rsmidt]

Thanks @ptrdom! This is looking promising. I'll give it a try next week 👍🏻

[pjfanning]

I don't know if the dispatcher set up has much bearing on the use case here but we have some unit tests for pekko persistence where the dispatcher setup matters. Also, the stashing config in Pekko persistence could be tuned.

[rsmidt]

For others having similar issues: After working on this for a while, I can confidently say that the solution worked out just fine.

However, after rolling it out, I've realized a second issue: given that we have 105 tags, filling up the buffer (with a default size of 500 events) for each tagged projector instance results in ~50k rows being fetched more or less simultaneously. This also blocks the DB pool for too long, resulting in throttling in other parts of the application.

Coincidentally, this was merged two weeks ago: apache/pekko-persistence-jdbc@6f420d2. This configuration property is precisely what we need. We'll keep playing around with it and the max-buffer-size to see what our system can tolerate.

Again, @pjfanning thanks a lot!