fix: EventWriter snapshot event

* The EventWriter.Write now takes a isSnapshotEvent parameter,
  which means that sequence number gaps before the snapshot will
  be filled with filtered event payloads
* The EventEnvelope.source is "SN" for snapshot events
* No dependency to akka-persistence-r2dbc from akka-projection-grpc so
  hardcoding source "SN" (can't use EnvelopeOrigin)

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

@@ -25,10 +25,23 @@ import akka.projection.grpc.internal.proto.EventConsumerServicePowerApi
 import akka.stream.scaladsl.Source
 import io.grpc.Status
 import org.slf4j.LoggerFactory
-
 import scala.concurrent.ExecutionContext
 import scala.concurrent.Promise
 
+import akka.persistence.query.typed.EventEnvelope
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[akka] object EventPusherConsumerServiceImpl {
+  // See akka.persistence.r2dbc.internal.EnvelopeOrigin, but we don't have a dependency
+  // to akka-persistence-r2dbc here
+  def fromSnapshot(env: EventEnvelope[_]): Boolean =
+    env.source == "SN"
+
+}
+
 /**
  * INTERNAL API
  *
@@ -40,6 +53,7 @@ private[akka] final class EventPusherConsumerServiceImpl(
     preferProtobuf: ProtoAnySerialization.Prefer)(implicit system: ActorSystem[_])
     extends EventConsumerServicePowerApi {
 
+  import EventPusherConsumerServiceImpl.fromSnapshot
   import ProtobufProtocolConversions._
 
   private val logger = LoggerFactory.getLogger(classOf[EventPusherConsumerServiceImpl])
@@ -134,6 +148,7 @@ private[akka] final class EventPusherConsumerServiceImpl(
                       transformedEventEnvelope.persistenceId,
                       transformedEventEnvelope.sequenceNr,
                       transformedEventEnvelope.eventOption.getOrElse(FilteredPayload),
+                      isSnapshotEvent = fromSnapshot(transformedEventEnvelope),
                       transformedEventEnvelope.eventMetadata,
                       transformedEventEnvelope.tags,
                       _))(destination.eventProducerPushDestination.settings.journalWriteTimeout, system.scheduler)

project/Dependencies.scala

@@ -19,7 +19,7 @@ object Dependencies {
   val AkkaProjectionVersionInDocs = "1.5"
 
   object Versions {
-    val akka = sys.props.getOrElse("build.akka.version", "2.9.0-M3")
+    val akka = sys.props.getOrElse("build.akka.version", "2.9.0-M3+24-4e2bc5b5-SNAPSHOT")
     val akkaPersistenceCassandra = "1.2.0-M1"
     val akkaPersistenceJdbc = "5.3.0-M1"
     val akkaPersistenceR2dbc = "1.2.0-M7"

samples/grpc/iot-service-scala/build.sbt

@@ -28,7 +28,7 @@ run / javaOptions ++= sys.props
   .fold(Seq.empty[String])(res => Seq(s"-Dconfig.resource=$res"))
 Global / cancelable := false // ctrl-c
 
-val AkkaVersion = "2.9.0-M3"
+val AkkaVersion = "2.9.0-M3+24-4e2bc5b5-SNAPSHOT"
 val AkkaHttpVersion = "10.6.0-M2"
 val AkkaManagementVersion = "1.5.0-M1"
 val AkkaPersistenceR2dbcVersion = "1.2.0-M7"

samples/grpc/local-drone-control-java/pom.xml

@@ -17,7 +17,7 @@
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <akka.version>2.9.0-M3</akka.version>
+        <akka.version>2.9.0-M3+24-4e2bc5b5-SNAPSHOT</akka.version>
         <akka-projection.version>1.5.0-M5</akka-projection.version>
         <akka-persistence-r2dbc.version>1.2.0-M7</akka-persistence-r2dbc.version>
         <akka-management.version>1.5.0-M1</akka-management.version>

samples/grpc/local-drone-control-scala/build.sbt

@@ -31,7 +31,7 @@ run / javaOptions ++= sys.props
   .fold(Seq.empty[String])(res => Seq(s"-Dconfig.resource=$res"))
 Global / cancelable := false // ctrl-c
 
-val AkkaVersion = "2.9.0-M3"
+val AkkaVersion = "2.9.0-M3+24-4e2bc5b5-SNAPSHOT"
 val AkkaHttpVersion = "10.6.0-M2"
 val AkkaManagementVersion = "1.5.0-M1"
 val AkkaPersistenceR2dbcVersion = "1.2.0-M7"

samples/grpc/restaurant-drone-deliveries-service-java/pom.xml

@@ -17,7 +17,7 @@
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <akka.version>2.9.0-M3</akka.version>
+        <akka.version>2.9.0-M3+24-4e2bc5b5-SNAPSHOT</akka.version>
         <akka-projection.version>1.5.0-M5</akka-projection.version>
         <akka-persistence-r2dbc.version>1.2.0-M6</akka-persistence-r2dbc.version>
         <akka-management.version>1.5.0-M1</akka-management.version>

samples/grpc/restaurant-drone-deliveries-service-scala/build.sbt

@@ -28,7 +28,7 @@ run / javaOptions ++= sys.props
   .fold(Seq.empty[String])(res => Seq(s"-Dconfig.resource=$res"))
 Global / cancelable := false // ctrl-c
 
-val AkkaVersion = "2.9.0-M3"
+val AkkaVersion = "2.9.0-M3+24-4e2bc5b5-SNAPSHOT"
 val AkkaHttpVersion = "10.6.0-M2"
 val AkkaManagementVersion = "1.5.0-M1"
 val AkkaPersistenceR2dbcVersion = "1.2.0-M6"

samples/grpc/shopping-analytics-service-java/pom.xml

@@ -17,7 +17,7 @@
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <akka.version>2.9.0-M3</akka.version>
+        <akka.version>2.9.0-M3+24-4e2bc5b5-SNAPSHOT</akka.version>
         <akka-projection.version>1.5.0-M5</akka-projection.version>
         <akka-persistence-r2dbc.version>1.2.0-M7</akka-persistence-r2dbc.version>
         <akka-management.version>1.5.0-M1</akka-management.version>

samples/grpc/shopping-analytics-service-scala/build.sbt

@@ -28,7 +28,7 @@ run / javaOptions ++= sys.props
   .fold(Seq.empty[String])(res => Seq(s"-Dconfig.resource=$res"))
 Global / cancelable := false // ctrl-c
 
-val AkkaVersion = "2.9.0-M3"
+val AkkaVersion = "2.9.0-M3+24-4e2bc5b5-SNAPSHOT"
 val AkkaHttpVersion = "10.6.0-M2"
 val AkkaManagementVersion = "1.5.0-M1"
 val AkkaPersistenceR2dbcVersion = "1.2.0-M7"

samples/grpc/shopping-cart-service-java/pom.xml

@@ -17,7 +17,7 @@
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <akka.version>2.9.0-M3</akka.version>
+        <akka.version>2.9.0-M3+24-4e2bc5b5-SNAPSHOT</akka.version>
         <akka-projection.version>1.5.0-M5</akka-projection.version>
         <akka-persistence-r2dbc.version>1.2.0-M7</akka-persistence-r2dbc.version>
         <akka-management.version>1.5.0-M1</akka-management.version>

samples/grpc/shopping-cart-service-scala/build.sbt

@@ -28,7 +28,7 @@ run / javaOptions ++= sys.props
   .fold(Seq.empty[String])(res => Seq(s"-Dconfig.resource=$res"))
 Global / cancelable := false // ctrl-c
 
-val AkkaVersion = "2.9.0-M3"
+val AkkaVersion = "2.9.0-M3+24-4e2bc5b5-SNAPSHOT"
 val AkkaHttpVersion = "10.6.0-M2"
 val AkkaManagementVersion = "1.5.0-M1"
 val AkkaPersistenceR2dbcVersion = "1.2.0-M7"

samples/replicated/shopping-cart-service-java/pom.xml

@@ -17,7 +17,7 @@
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <akka.version>2.9.0-M3</akka.version>
+        <akka.version>2.9.0-M3+24-4e2bc5b5-SNAPSHOT</akka.version>
         <akka-projection.version>1.5.0-M5</akka-projection.version>
         <akka-persistence-r2dbc.version>1.2.0-M7</akka-persistence-r2dbc.version>
         <akka-management.version>1.5.0-M1</akka-management.version>

samples/replicated/shopping-cart-service-scala/build.sbt

@@ -29,7 +29,7 @@ run / javaOptions ++= sys.props
   .fold(Seq.empty[String])(res => Seq(s"-Dconfig.resource=$res"))
 Global / cancelable := false // ctrl-c
 
-val AkkaVersion = sys.props.getOrElse("akka.version", "2.9.0-M3")
+val AkkaVersion = sys.props.getOrElse("akka.version", "2.9.0-M3+24-4e2bc5b5-SNAPSHOT")
 val AkkaHttpVersion = "10.6.0-M2"
 val AkkaManagementVersion = "1.5.0-M1"
 val AkkaPersistenceR2dbcVersion = "1.2.0-M7"