This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch port/b179bb1527-filtered-payload in repository https://gitbox.apache.org/repos/asf/pekko.git
commit 47ada0f9958939dd50f9cbdb90ad6a5e53d6d892 Author: He-Pin(kerr) <[email protected]> AuthorDate: Tue Jun 30 15:17:04 2026 +0800 feat: add FilteredPayload placeholder for filtered journal events 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]
