This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new b9bd7d0f5a feat: add FilteredPayload placeholder for filtered journal
events (#3275)
b9bd7d0f5a is described below
commit b9bd7d0f5a9f6845573b69307e869835936c43b1
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Wed Jul 1 17:56:17 2026 +0800
feat: add FilteredPayload placeholder for filtered journal events (#3275)
Motivation:
Projections and events-by-slice queries need to keep per-persistence-id
sequence numbers gap free, which means filtered events must still be stored
in the journal. There was no shared placeholder payload for this in Pekko
Persistence, so query/projection plugins had nothing to align on.
Modification:
- Add the public `FilteredPayload` case object, serialized to a 0-byte
representation by the new internal `FilteredPayloadSerializer` (manifest
"F", serializer id 34 - matching Pekko's convention of keeping upstream
serializer ids, verified free in Pekko).
- Register the serializer, binding and identifier in persistence
reference.conf.
- Make typed `EventEnvelope.event`/`getEvent` raise a clearer error when the
payload is absent because the event was filtered (vs. simply not loaded).
Result:
Pekko Persistence has a standard placeholder for filtered journal events,
flagged as `filtered` in the typed `EventEnvelope`, enabling query and
projection plugins to represent gap-free sequence numbers.
Tests:
- sbt "persistence/testOnly
org.apache.pekko.persistence.serialization.FilteredPayloadSerializerSpec" (1/1
passed)
- sbt "persistence-query/testOnly
org.apache.pekko.persistence.query.typed.EventEnvelopeSpec" (2/2 passed;
directional test for the filtered vs not-loaded error, not present upstream)
- sbt "persistence/mimaReportBinaryIssues"
"persistence-query/mimaReportBinaryIssues" (clean)
References:
Upstream: akka/akka-core@b179bb15279962d657b9192f374a98123fabd7ee
(akka#31985), which is now Apache licensed.
---
.../persistence/query/typed/EventEnvelope.scala | 18 ++++++--
.../query/typed/EventEnvelopeSpec.scala | 53 ++++++++++++++++++++++
persistence/src/main/resources/reference.conf | 3 ++
.../apache/pekko/persistence/FilteredPayload.scala | 26 +++++++++++
.../serialization/FilteredPayloadSerializer.scala | 45 ++++++++++++++++++
.../FilteredPayloadSerializerSpec.scala | 35 ++++++++++++++
6 files changed, 176 insertions(+), 4 deletions(-)
diff --git
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/EventEnvelope.scala
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/EventEnvelope.scala
index 6cbb7e5335..5f8018365a 100644
---
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/EventEnvelope.scala
+++
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/EventEnvelope.scala
@@ -187,8 +187,13 @@ final class EventEnvelope[Event](
eventOption match {
case Some(evt) => evt
case None =>
- throw new IllegalStateException(
- "Event was not loaded. Use eventOption and load the event on demand
with LoadEventQuery.")
+ if (filtered) {
+ throw new IllegalStateException(
+ "Event was filtered so payload is not present. Use eventOption to
handle more gracefully.")
+ } else {
+ throw new IllegalStateException(
+ "Event was not loaded. Use eventOption and load the event on
demand with LoadEventQuery.")
+ }
}
/**
@@ -198,8 +203,13 @@ final class EventEnvelope[Event](
eventOption match {
case Some(evt) => evt
case None =>
- throw new IllegalStateException(
- "Event was not loaded. Use getOptionalEvent and load the event on
demand with LoadEventQuery.")
+ if (filtered) {
+ throw new IllegalStateException(
+ "Event was filtered so payload is not present. Use
getOptionalEvent to handle more gracefully.")
+ } else {
+ throw new IllegalStateException(
+ "Event was not loaded. Use getOptionalEvent and load the event on
demand with LoadEventQuery.")
+ }
}
/**
diff --git
a/persistence-query/src/test/scala/org/apache/pekko/persistence/query/typed/EventEnvelopeSpec.scala
b/persistence-query/src/test/scala/org/apache/pekko/persistence/query/typed/EventEnvelopeSpec.scala
new file mode 100644
index 0000000000..ff482463d6
--- /dev/null
+++
b/persistence-query/src/test/scala/org/apache/pekko/persistence/query/typed/EventEnvelopeSpec.scala
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence.query.typed
+
+import org.apache.pekko
+import pekko.persistence.query.NoOffset
+import pekko.testkit.PekkoSpec
+
+class EventEnvelopeSpec extends PekkoSpec {
+
+ private def envelope(filtered: Boolean): EventEnvelope[String] =
+ new EventEnvelope[String](
+ NoOffset,
+ persistenceId = "pid-1",
+ sequenceNr = 1L,
+ eventOption = None,
+ timestamp = 0L,
+ eventMetadata = None,
+ entityType = "entity",
+ slice = 0,
+ filtered = filtered,
+ source = "")
+
+ "EventEnvelope" should {
+
+ "throw a filtered-specific error from event/getEvent when the payload was
filtered out" in {
+ val env = envelope(filtered = true)
+ intercept[IllegalStateException](env.event).getMessage should
include("filtered")
+ intercept[IllegalStateException](env.getEvent()).getMessage should
include("filtered")
+ }
+
+ "throw a not-loaded error from event/getEvent when the payload was simply
not loaded" in {
+ val env = envelope(filtered = false)
+ intercept[IllegalStateException](env.event).getMessage should
include("not loaded")
+ intercept[IllegalStateException](env.getEvent()).getMessage should
include("not loaded")
+ }
+ }
+}
diff --git a/persistence/src/main/resources/reference.conf
b/persistence/src/main/resources/reference.conf
index 6634dd6233..ff34a035bb 100644
--- a/persistence/src/main/resources/reference.conf
+++ b/persistence/src/main/resources/reference.conf
@@ -233,14 +233,17 @@ pekko.actor {
serializers {
pekko-persistence-message =
"org.apache.pekko.persistence.serialization.MessageSerializer"
pekko-persistence-snapshot =
"org.apache.pekko.persistence.serialization.SnapshotSerializer"
+ pekko-persistence-filtered =
"org.apache.pekko.persistence.serialization.FilteredPayloadSerializer"
}
serialization-bindings {
"org.apache.pekko.persistence.serialization.Message" =
pekko-persistence-message
"org.apache.pekko.persistence.serialization.Snapshot" =
pekko-persistence-snapshot
+ "org.apache.pekko.persistence.FilteredPayload$" =
pekko-persistence-filtered
}
serialization-identifiers {
"org.apache.pekko.persistence.serialization.MessageSerializer" = 7
"org.apache.pekko.persistence.serialization.SnapshotSerializer" = 8
+ "org.apache.pekko.persistence.serialization.FilteredPayloadSerializer"
= 34
}
}
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/FilteredPayload.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/FilteredPayload.scala
new file mode 100644
index 0000000000..d624f1b0b0
--- /dev/null
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/FilteredPayload.scala
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence
+
+/**
+ * In some use cases with projections and events by slice filtered events
needs to be stored in the journal
+ * to keep the sequence numbers for a given persistence id gap free. This
placeholder payload is used for those
+ * cases and serialized down to a 0-byte representation when stored in the
database.
+ *
+ * This payload is not in general expected to show up for users but in some
scenarios/queries it may.
+ *
+ * In the typed queries `EventEnvelope` this should be flagged as `filtered`
and turned into a non-present payload
+ * by the query plugin implementations.
+ */
+case object FilteredPayload
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/serialization/FilteredPayloadSerializer.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/serialization/FilteredPayloadSerializer.scala
new file mode 100644
index 0000000000..c95dbbe39c
--- /dev/null
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/serialization/FilteredPayloadSerializer.scala
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.serialization
+
+import org.apache.pekko
+import pekko.actor.ExtendedActorSystem
+import pekko.annotation.InternalApi
+import pekko.persistence.FilteredPayload
+import pekko.serialization.BaseSerializer
+import pekko.serialization.SerializerWithStringManifest
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+final class FilteredPayloadSerializer(val system: ExtendedActorSystem)
+ extends SerializerWithStringManifest
+ with BaseSerializer {
+
+ private val FilteredPayloadManifest = "F"
+
+ override def manifest(o: AnyRef): String = o match {
+ case FilteredPayload => FilteredPayloadManifest
+ case _ => throw new IllegalArgumentException(s"Can't
serialize object of type ${o.getClass}")
+ }
+
+ override def toBinary(o: AnyRef): Array[Byte] = o match {
+ case FilteredPayload => Array.empty
+ case _ => throw new IllegalArgumentException(s"Can't
serialize object of type ${o.getClass}")
+ }
+
+ override def fromBinary(bytes: Array[Byte], manifest: String): AnyRef =
FilteredPayload
+
+}
diff --git
a/persistence/src/test/scala/org/apache/pekko/persistence/serialization/FilteredPayloadSerializerSpec.scala
b/persistence/src/test/scala/org/apache/pekko/persistence/serialization/FilteredPayloadSerializerSpec.scala
new file mode 100644
index 0000000000..1858a5486c
--- /dev/null
+++
b/persistence/src/test/scala/org/apache/pekko/persistence/serialization/FilteredPayloadSerializerSpec.scala
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2020-2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.serialization
+
+import org.apache.pekko
+import pekko.persistence.FilteredPayload
+import pekko.serialization.SerializationExtension
+import pekko.serialization.SerializerWithStringManifest
+import pekko.testkit.PekkoSpec
+
+class FilteredPayloadSerializerSpec extends PekkoSpec {
+
+ "FilteredPayload serializer" should {
+ "serialize FilteredPayload to zero-byte array" in {
+ val serialization = SerializationExtension(system)
+ val serializer =
serialization.findSerializerFor(FilteredPayload).asInstanceOf[SerializerWithStringManifest]
+ val manifest = serializer.manifest(FilteredPayload)
+ val serialized = serializer.toBinary(FilteredPayload)
+ serialized should have(size(0))
+ serializer.fromBinary(serialized, manifest) should be(FilteredPayload)
+ }
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
