Producer initiated pushing of events over gRPC #932

johanandren · 2023-06-26T14:42:05Z

Superseedes #920

complete end user APIs Java and Scala
handling duplicates/redelivery in the event writer
more thorough test coverage for just the event writer
something like the producer interceptors for pluggable auth based on request metadata
some form of consumer-decided filtering, already on the producer
transformation of more than the persistence id on the consumer side, with access to origin id
initial docs

akka-projection-grpc/src/main/protobuf/akka/projection/grpc/event_consumer.proto

...-projection-grpc/src/main/scala/akka/projection/grpc/internal/EventConsumerServiceImpl.scala

akka-projection-grpc/src/main/scala/akka/persistence/EventWriter.scala

akka-projection-grpc/src/main/scala/akka/projection/grpc/internal/FilteredPayload.scala

akka-projection-grpc-tests/src/it/scala/akka/projection/grpc/producer/ProducerPushSpec.scala

patriknw · 2023-06-28T14:10:32Z

akka-projection-grpc/src/main/protobuf/akka/projection/grpc/event_consumer.proto

+  oneof message {
+    // always sent first
+    ConsumerEventInit init = 1;
+    Event event = 2;


in event_producer.proto we have Event and FilteredEvent. Should we use FilteredEvent here too?

That would make the protocol more clear yeah, I'll do that.

Changed in 80d67c0

patriknw · 2023-06-28T14:24:16Z

...-projection-grpc/src/main/scala/akka/projection/grpc/internal/EventConsumerServiceImpl.scala

+      }
+      .via(ActorFlow.askWithStatus(8)(eventWriter) {
+        (in: proto.Event, replyTo: ActorRef[StatusReply[EventWriter.WriteAck]]) =>
+          // FIXME would we want request metadata/producer ip/entire envelope in to persistence id transformer?


yes, one case could be that based on the origin of the event you want to adjust the persistenceId to include that origin (for example making it more unique)

another thing that could be good to amend before writing the event would be the tags, e.g. adding a tag for the origin, which can then be used in filtering later

we could have a transformation function of the Entire EventEnvelope, but we shouldn't allow changing the seqNr

btw, would then probably be nice with some "copy methods" on the EventEnvelope, such as withTags, withPersistenceId.

I guess we could accept a transform but always check that sequence number was not changed. Or more a more limited transform API instead of adding API to let users actually transform the entire envelope, Transform.addTags().

Added a try at a transform API but didn't include the origin as a parameter yet: fdb350e

And now the transform can be created per origin/event stream based on origin id and request metadata fda1176

akka-projection-grpc/src/main/scala/akka/projection/grpc/internal/FilteredPayload.scala

patriknw · 2023-06-30T09:18:37Z

...-projection-grpc/src/main/scala/akka/projection/grpc/internal/EventConsumerServiceImpl.scala

+      // FIXME config for parallelism, and perPartition (aligned with event writer batch config)
+      .mapAsyncPartitioned(1000, 20)(_.persistenceId) { (in, _) =>
+        // FIXME would we want request metadata/producer ip/entire envelope in to persistence id transformer?
+        val envelope = ProtobufProtocolConversions.eventToEnvelope[Any](in, protoAnySerialization)


Would be good if we can skip the deserialization-serialization roundtrip here, and that may impact what kind of transformer we want to support. If the transformer is not changing the payload we can avoid the serialization.

patriknw

I just looked briefly at this again, let me know when it's time for a deeper look.

patriknw · 2023-07-03T13:46:32Z

akka-projection-grpc/src/main/scala/akka/persistence/EventWriter.scala

+
+          }
+        })
+      .onFailure[Exception](SupervisorStrategy.restart)


should have backoff restart, because if it starts failing we don't want to overload the db?

Honestly not quite sure how this would crash, other than some bug in the batching logic, errors persisting events are fed back to the sender and shouldn't crash the writer itself.

Right, we are not failing the writer actor when we there are db errors. The CircuitBreaker in AsyncWriteJournal will protect the db from overload when it's already in trouble. Ok, we can leave it as plain restart.

akka-projection-grpc/src/main/scala/akka/persistence/EventWriter.scala

akka-projection-grpc/src/main/scala/akka/projection/grpc/internal/ConsumerSerializer.scala

johanandren · 2023-07-03T14:42:21Z

@patriknw the largest task that is left is to come up with proper user API, and once that is in place, break out the sample I started on to be able to test it properly, and some docs.

I think you can review the internal bits and pieces a bit more if you have time.

Possible small-to-medium things left, not sure if we need all right now or can iterate:

something like the producer interceptors for pluggable auth based on request metadata
still feels like some form of consumer-decided filtering would be good - filtering out events already on the producer, could maybe be static and sent as immediate response to the consumer opening up the stream
transformation of more than the persistence id on the consumer side
avoiding deserialization unless needed

johanandren · 2023-07-04T13:08:13Z

@patriknw WDYT is this producer API simple enough while being flexible enough that you can run it in a non-clustered actor system to just push events for one or a few entities, but ofc also in a cluster with SDP to push a larger number of entities.

Or do we need something higher level like we did with the Replication which takes care of starting sharding, naming projections etc for the user?

akka-projection-grpc/src/main/scala/akka/projection/grpc/consumer/scaladsl/EventConsumer.scala

...-projection-grpc/src/main/scala/akka/projection/grpc/internal/EventConsumerServiceImpl.scala

akka-projection-grpc/src/main/scala/akka/projection/grpc/internal/EventPusher.scala

...jection-grpc/src/main/scala/akka/projection/grpc/producer/scaladsl/ActiveEventProducer.scala

patriknw · 2023-07-04T13:39:24Z

akka-projection-grpc/src/main/scala/akka/persistence/EventWriter.scala

+                    }
+                }
+
+              case JournalProtocol.WriteMessageFailure(message, error, `actorInstanceId`) =>


How should we handle duplicates, which will result in write failures? We will have duplicates since it's at-least-once.

One idea is to optimistically write, but if it fails we can check if the pid/seqNr exists with EventTimestampQuery or LoadEventQuery, or even eventsByPersistenceId since we are interested in more than one in the batch.

Not great to add a dependency to readJournalPluginId too, so maybe we can do this with the journal protocol?
asyncReadHighestSequenceNr is not exposed as journal protocol message, so if we don't want to change Akka we could use ReplayMessages to find the highest sequence number.

Hmm, and the exception is from the plugin, so there is no way to determine that it is a duplicate-write from the failure response. Replay seems like a good trick, because we could do fromSequenceNr = toSequenceNr just for that seqNr.

Exception is from plugin, but we can't see the difference of duplicate error to any other error, at this level.
Some plugins may not fail on duplicates, such as Cassandra, but that is another story.

and I think you know which other seqNrs are pending from the same batch, and it's very likely that they will also fail. Then it's probably easier to use WriteMessagesFailure to trigger these checks. We know that all or none will fail because they are in AtomicWrite.

…ith it

patriknw

EventWriter...

akka-projection-grpc/src/main/scala/akka/persistence/typed/EventWriter.scala

akka-projection-grpc-tests/src/it/scala/akka/persistence/typed/EventWriterSpec.scala

patriknw

javadsl looking good

...grpc/src/main/scala/akka/projection/grpc/consumer/javadsl/EventProducerPushDestination.scala

patriknw · 2023-07-07T14:11:38Z

avoiding deserialization unless needed

that can wait until later, probably a rather big task, if we think about all cases where it would be nice

…re updates

johanandren · 2023-07-10T13:40:24Z

This needs an Akka patch release with the event writer, but then I think we can merge and iterate on top.

patriknw

LGTM

patriknw · 2023-08-08T11:51:27Z

...projection-grpc-tests/src/it/scala/akka/projection/grpc/producer/EventProducerPushSpec.scala

+              .registerTagMapper[String](_ => Set("added-tag"))
+              .registerPayloadMapper[String, String](env => env.eventOption.map(_.toUpperCase))
+          }
+          .withConsumerFilters(Vector(ConsumerFilter.ExcludeEntityIds(Set(consumerFilterExcludedPid.id))))


patriknw reviewed Jun 26, 2023

View reviewed changes

johanandren force-pushed the wip-producer-event-push branch 3 times, most recently from 45e023f to 52d8b0c Compare June 27, 2023 14:20

johanandren commented Jun 28, 2023

View reviewed changes

akka-projection-grpc/src/main/scala/akka/projection/grpc/internal/FilteredPayload.scala Outdated Show resolved Hide resolved

patriknw reviewed Jun 28, 2023

View reviewed changes

patriknw mentioned this pull request Jun 29, 2023

H2 default schema in config that can be included #935

Closed

johanandren mentioned this pull request Jun 29, 2023

feat: Placeholder event representation for filtered journal events akka/akka#31985

Merged

patriknw reviewed Jun 30, 2023

View reviewed changes

patriknw reviewed Jul 3, 2023

View reviewed changes

johanandren force-pushed the wip-producer-event-push branch 2 times, most recently from b426abe to 4f5e427 Compare July 4, 2023 11:51

patriknw reviewed Jul 4, 2023

View reviewed changes

johanandren added 14 commits July 4, 2023 20:15

WIP first little step, push based replication/projection

7036d2a

First PoC test passing

23d7e52

formatting, headers, all Scala compilers happy

f41ae1c

Move test, speed it up

f500abd

Persistence id transformation

68d2397

Producer filtering of events, first stab

3313215

Stream of event over gRPC instead of request-response

bf71e4f

Microbatching of writes in event writer

5de13c2

Iteration on the EventWriter and how the consumer service interacts w…

210e8fa

…ith it

Consumer decides if it accepts a given stream id or not

53b5314

Some arbitrary changes

2807d96

Seed of a sample, making it easy to run multiple producer JVMs

4515866

Make it compile on Scala 2.12

e40368a

Keep alive event from producer

9bfba36

johanandren added 7 commits July 6, 2023 15:40

document config keys

d7302af

Smaller review feedback things

d7f1eda

event writer test coverage, further simplification and config

9799279

headers and formatting

fd8894d

All I want is to make Scala 2.12 happy

7e9f54c

test writing to many pids

5cdb01b

runaway Done

138e83d

patriknw reviewed Jul 6, 2023

View reviewed changes

johanandren added 3 commits July 7, 2023 15:02

Java APIs and API doc improvements

fd97245

headersnformatting

2a83c26

correct copyright headers

5571754

patriknw reviewed Jul 7, 2023

View reviewed changes

...grpc/src/main/scala/akka/projection/grpc/consumer/javadsl/EventProducerPushDestination.scala Outdated Show resolved Hide resolved

johanandren added 6 commits July 7, 2023 17:05

Akka snapshot with event writer

ac11dff

some more test deps to align akka versions

b27f8c3

Api doc clarification

87f604a

And more test dependencies

6f0d0b7

First iteration of docs for producer push

5871ea9

Small diagram improvement, also, include the drawio document for futu…

43340a5

…re updates

johanandren marked this pull request as ready for review July 10, 2023 13:40

patriknw approved these changes Aug 8, 2023

View reviewed changes

johanandren modified the milestones: 1.5.0-M1, 1.5.0-M2 Aug 15, 2023

johanandren added 4 commits August 15, 2023 16:58

bump: Akka 2.8.4

cbb4083

EventProducerSettings constructor now private, mima exlude

4cfe1c5

Doc samples never used vals are ok

e8243b8

Vector for Scala 2.12

f260502

johanandren merged commit 3b8a054 into main Aug 15, 2023

johanandren deleted the wip-producer-event-push branch August 15, 2023 16:17

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Producer initiated pushing of events over gRPC #932

Producer initiated pushing of events over gRPC #932

johanandren commented Jun 26, 2023 •

edited

Loading

patriknw Jun 28, 2023

johanandren Jun 28, 2023

johanandren Jul 3, 2023

patriknw Jun 28, 2023

patriknw Jun 28, 2023

patriknw Jun 28, 2023

johanandren Jun 28, 2023

johanandren Jul 4, 2023

johanandren Jul 5, 2023

patriknw Jun 30, 2023

patriknw left a comment

patriknw Jul 3, 2023

johanandren Jul 3, 2023

patriknw Jul 4, 2023

johanandren commented Jul 3, 2023 •

edited

Loading

johanandren commented Jul 4, 2023

patriknw Jul 4, 2023

johanandren Jul 4, 2023

patriknw Jul 4, 2023

patriknw left a comment

patriknw left a comment

patriknw commented Jul 7, 2023

johanandren commented Jul 10, 2023

patriknw left a comment

patriknw Aug 8, 2023

Producer initiated pushing of events over gRPC #932

Producer initiated pushing of events over gRPC #932

Conversation

johanandren commented Jun 26, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

patriknw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

johanandren commented Jul 3, 2023 • edited Loading

johanandren commented Jul 4, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

patriknw left a comment

Choose a reason for hiding this comment

patriknw left a comment

Choose a reason for hiding this comment

patriknw commented Jul 7, 2023

johanandren commented Jul 10, 2023

patriknw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

johanandren commented Jun 26, 2023 •

edited

Loading

johanandren commented Jul 3, 2023 •

edited

Loading