perf: Scalable slice queries for many consumers #31957

patriknw · 2023-06-02T09:09:32Z

No description provided.

patriknw · 2023-06-02T09:24:27Z

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala

+
+  final class SlowConsumerException(message: String) extends RuntimeException(message) with NoStackTrace
+
+  final case class FirehoseKey(pluginId: String, entityType: String, sliceRange: Range)


I intend to create a follow up ticket about colocating shards. So when several projections are started with ShardedDaemonProcess they would try to allocate slice ranges to the same nodes, so that firehose for a slice range is not started in more places than necessary.

that was #31967

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala

johanandren

Looking good so far (also reviewed several rounds in the original PR).

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala

...y/src/main/scala/akka/persistence/query/typed/javadsl/EventsBySliceFirehoseReadJournal.scala

octonato

I did a first pass last Friday. I have a bunch of questions but need to re-review and refine them.

I'm leaving a nitpick formatting suggestion though. 😄

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala

octonato

Looking really good.

Leaving some comments already.
I want to do a second pass to review more in detail the slow consumer detection.

akka-persistence-query/src/main/resources/reference.conf

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala

akka-persistence-query/src/main/resources/reference.conf

octonato · 2023-06-07T09:50:38Z

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala

+              else if (diffFastest < 0) s"ahead of fastest [$diffFastest] ms" // not possible
+              else "same as fastest"
+            val diffSlowest = slowestConsumer.offsetTimestamp.toEpochMilli - tracking.offsetTimestamp.toEpochMilli
+            val diffSlowestStr =
+              if (diffSlowest > 0) s"behind slowest [$diffSlowest] ms" // not possible


Agree it's not possible, but what if we have a bug and it shows up in logs. Wouldn't that be confusing?

Shall we log something else or fail the stream?

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala

octonato

I'm finding it a really cool feature.

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala

johanandren

More detailed review of the slow-consumer-detection.

.../src/main/scala/akka/persistence/query/typed/scaladsl/EventsBySliceFirehoseReadJournal.scala

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala

johanandren · 2023-06-08T12:07:54Z

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala

+            if (existing.history.size <= settings.broadcastBufferSize)
+              existing.history :+ offset
+            else
+              existing.history.tail :+ offset // drop one, add one


Expensive ops with vector tail + append for each element, is there a better data structure we could use, some form of queue?

I don't think tail is that bad, still listed as eC. Scala immutable Queue is probably faster for this, but then we also use size, which is not good for Queue since it's delegating to the underlying Lists.

I think I'll leave it as Vector for now.

johanandren · 2023-06-08T12:14:41Z

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala

+        val slowestConsumer = trackingValues.minBy(_.offsetTimestamp)
+        val fastestConsumer = trackingValues.maxBy(_.offsetTimestamp)
+
+        val behind = elementsBehind(fastestConsumer.history, slowestConsumer.history)


Hmmm, the slowest consumer will hold back the fastest so that it should always have the timestamp of the slower in its history, because we keep as long offset history as the size of the buffer?

Exactly. Not great to have to keep the history for each consumer but I couldn't think of another way. It's just small objects, and probably held in memory by other things.

There is a case when the fastest is a new consumer and then it wouldn't have the full history yet. Then it keeps going.

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala

* it's also possible to have more than one events-by-slice-firehose, each can have it's own underlying plugin, and creating its own Firehose instance

Co-authored-by: Francisco Lopez-Sancho <franciscolopezsancho@gmail.com> Co-authored-by: Renato Cavalcanti <renato@cavalcanti.be>

Co-authored-by: Renato Cavalcanti <renato@cavalcanti.be>

Co-authored-by: Johan Andrén <johan@markatta.com>

Co-authored-by: Renato Cavalcanti <renato@cavalcanti.be>

johanandren

LGTM, great work @patriknw

patriknw mentioned this pull request Jun 2, 2023

perf: Scalable slice queries for many consumers akka/akka-projection#907

Closed

patriknw commented Jun 2, 2023

View reviewed changes

octonato reviewed Jun 2, 2023

View reviewed changes

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala Outdated Show resolved Hide resolved

patriknw commented Jun 5, 2023

View reviewed changes

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala Outdated Show resolved Hide resolved

johanandren reviewed Jun 5, 2023

View reviewed changes

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala Show resolved Hide resolved

...y/src/main/scala/akka/persistence/query/typed/javadsl/EventsBySliceFirehoseReadJournal.scala Outdated Show resolved Hide resolved

octonato reviewed Jun 5, 2023

View reviewed changes

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala Outdated Show resolved Hide resolved

octonato reviewed Jun 7, 2023

View reviewed changes

franciscolopezsancho reviewed Jun 7, 2023

View reviewed changes

akka-persistence-query/src/main/resources/reference.conf Outdated Show resolved Hide resolved

franciscolopezsancho reviewed Jun 7, 2023

View reviewed changes

akka-persistence-query/src/main/resources/reference.conf Outdated Show resolved Hide resolved

octonato reviewed Jun 7, 2023

View reviewed changes

octonato approved these changes Jun 7, 2023

View reviewed changes

...tence-query/src/main/scala/akka/persistence/query/typed/internal/EventsBySliceFirehose.scala Outdated Show resolved Hide resolved

patriknw mentioned this pull request Jun 7, 2023

feat: Collocating shards with consistent hashing #31967

Merged

2 tasks

johanandren reviewed Jun 8, 2023

View reviewed changes

patriknw and others added 17 commits June 12, 2023 16:00

perf: Scalable slice queries for many consumers

639ca5b

configuration

6e61341

* it's also possible to have more than one events-by-slice-firehose, each can have it's own underlying plugin, and creating its own Firehose instance

fixme

176a9b8

fmt

f3a79bc

deduplication when emitting from both

39698c0

verbose logging

175670a

refined logic for detecting slow consumers

5fc9265

updateConsumerTracking replace

a08507e

updateFirehoseOffset no isAfter

d8f8058

Apply suggestions from code review

1a15060

Co-authored-by: Francisco Lopez-Sancho <franciscolopezsancho@gmail.com> Co-authored-by: Renato Cavalcanti <renato@cavalcanti.be>

format

083112b

Co-authored-by: Renato Cavalcanti <renato@cavalcanti.be>

final

bb18b3b

Co-authored-by: Johan Andrén <johan@markatta.com>

fmt

78e4cd4

delegate eventsBySliceQuery

f9233e5

api docs and impl notes

4e0bde2

docs

ac788fb

rename to align with others

338edcb

patriknw force-pushed the wip-firehose-queries-patriknw branch from 96acc5c to 338edcb Compare June 12, 2023 14:56

patriknw and others added 6 commits June 12, 2023 17:21

fmt

85e5a1e

Apply suggestions from code review

00ce59e

Co-authored-by: Renato Cavalcanti <renato@cavalcanti.be>

compute in updateConsumerTracking

edbd176

Option.exists

6809861

computeIfPresent

3bb9a92

slowConsumerLagThreshold

e932372

johanandren approved these changes Jun 14, 2023

View reviewed changes

patriknw merged commit 362ab93 into main Jun 14, 2023

patriknw deleted the wip-firehose-queries-patriknw branch June 14, 2023 14:48

patriknw added this to the 2.8.3 milestone Jun 14, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

perf: Scalable slice queries for many consumers #31957

perf: Scalable slice queries for many consumers #31957

patriknw commented Jun 2, 2023

patriknw Jun 2, 2023

patriknw Jun 13, 2023

johanandren left a comment

octonato left a comment

octonato left a comment

octonato Jun 7, 2023

octonato left a comment

johanandren left a comment

johanandren Jun 8, 2023

patriknw Jun 13, 2023

johanandren Jun 8, 2023

patriknw Jun 8, 2023

johanandren left a comment


		final class SlowConsumerException(message: String) extends RuntimeException(message) with NoStackTrace

		final case class FirehoseKey(pluginId: String, entityType: String, sliceRange: Range)

perf: Scalable slice queries for many consumers #31957

perf: Scalable slice queries for many consumers #31957

Conversation

patriknw commented Jun 2, 2023

patriknw Jun 2, 2023

Choose a reason for hiding this comment

patriknw Jun 13, 2023

Choose a reason for hiding this comment

johanandren left a comment

Choose a reason for hiding this comment

octonato left a comment

Choose a reason for hiding this comment

octonato left a comment

Choose a reason for hiding this comment

octonato Jun 7, 2023

Choose a reason for hiding this comment

octonato left a comment

Choose a reason for hiding this comment

johanandren left a comment

Choose a reason for hiding this comment

johanandren Jun 8, 2023

Choose a reason for hiding this comment

patriknw Jun 13, 2023

Choose a reason for hiding this comment

johanandren Jun 8, 2023

Choose a reason for hiding this comment

patriknw Jun 8, 2023

Choose a reason for hiding this comment

johanandren left a comment

Choose a reason for hiding this comment