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 5c21c24988 feat: add SerializedEvent for writing pre-serialized events
(#3650)
5c21c24988 is described below
commit 5c21c24988dcb2d91c6216171129c8987e585244
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Oct 23 04:33:58 2026 +0100
feat: add SerializedEvent for writing pre-serialized events (#3650)
Motivation:
Some journal implementations can write an event payload together with its
serialization
information without serializing it again. Callers that already hold the
serialized bytes of
an event need a way to pass them through to the journal without a
deserialize/serialize round
trip. Such a wrapper may also be sent across the cluster, so it needs a
serializer.
Modification:
- New public `org.apache.pekko.persistence.SerializedEvent(bytes,
serializerId, serializerManifest)`.
- The internal `FilteredPayloadSerializer` (id 34) now also serializes
`SerializedEvent`
with manifest `"S"`, using the existing `PersistentPayload` protobuf
message.
- `SerializedEvent` is bound to the `pekko-persistence-filtered` serializer
in `reference.conf`.
- Unknown manifests now fail with `NotSerializableException` instead of
decoding to `FilteredPayload`.
- `FilteredPayloadSerializerSpec` extended with `SerializedEvent` round
trip tests.
Result:
`SerializedEvent` can be handed to journals that support it and can be sent
over remoting.
Tests:
Not run locally - left to CI
References:
Ported from (Apache-2.0 since 2026-10-23):
- https://github.com/akka/akka-core/pull/32068
Notes for reviewers:
- `FilteredPayloadSerializer` keeps its class name, serializer id 34 and
`"F"` manifest, so
the `FilteredPayload` wire format is unchanged and no MiMa excludes are
needed.
- Rolling upgrade: nodes on older versions decode any manifest of
serializer 34 as
`FilteredPayload`, so a `SerializedEvent` sent to an old node would be
silently read as
`FilteredPayload`. Avoid sending `SerializedEvent` across the cluster
until all nodes run
the new version.
- No journal in this repository writes `SerializedEvent` yet; support is up
to journal plugins.
- `SerializedEvent` has no structural `equals`/`hashCode` (it wraps an
`Array[Byte]`).
- No new config settings; only a new serialization binding.
---
.../paradox/migration/migration-guide-1.x-2.x.md | 3 ++
persistence/src/main/resources/reference.conf | 1 +
.../apache/pekko/persistence/SerializedEvent.scala | 24 +++++++++++
.../serialization/FilteredPayloadSerializer.scala | 42 ++++++++++++++++---
.../FilteredPayloadSerializerSpec.scala | 49 +++++++++++++++++++++-
5 files changed, 113 insertions(+), 6 deletions(-)
diff --git a/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
b/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
index 0b4d1f7a78..c5b6cd97b6 100644
--- a/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
+++ b/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
@@ -225,3 +225,6 @@ number of processes) that Pekko 1.x nodes cannot
deserialize. ([PR2755](https://
(for example with events-by-slice queries and projections), has a new
serializer with identifier `34`.
Pekko 1.x cannot deserialize it, so do not run Pekko 1.x against a journal
once such events have been
written. ([PR3275](https://github.com/apache/pekko/pull/3275))
+The same serializer also handles the new `SerializedEvent`, used to write
events that are already
+serialized; Pekko 1.x cannot deserialize that either, so do not send
`SerializedEvent` messages to
+Pekko 1.x nodes during a rolling update.
([PR3650](https://github.com/apache/pekko/pull/3650))
diff --git a/persistence/src/main/resources/reference.conf
b/persistence/src/main/resources/reference.conf
index 6e13d82a8c..6213e01e2f 100644
--- a/persistence/src/main/resources/reference.conf
+++ b/persistence/src/main/resources/reference.conf
@@ -244,6 +244,7 @@ pekko.actor {
"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
+ "org.apache.pekko.persistence.SerializedEvent" =
pekko-persistence-filtered
}
serialization-identifiers {
"org.apache.pekko.persistence.serialization.MessageSerializer" = 7
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/SerializedEvent.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/SerializedEvent.scala
new file mode 100644
index 0000000000..7a126194ff
--- /dev/null
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/SerializedEvent.scala
@@ -0,0 +1,24 @@
+/*
+ * 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
+
+/**
+ * Some journal implementations may support events of this type by writing the
event payload and
+ * serialization information without having to serialize it.
+ *
+ * @param bytes the already serialized event payload
+ * @param serializerId the identifier of the serializer that was used to
serialize the event
+ * @param serializerManifest the manifest of the serializer that was used,
empty string if none
+ */
+final class SerializedEvent(val bytes: Array[Byte], val serializerId: Int, val
serializerManifest: String)
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
index c95dbbe39c..6cd5c4a6a9 100644
---
a/persistence/src/main/scala/org/apache/pekko/persistence/serialization/FilteredPayloadSerializer.scala
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/serialization/FilteredPayloadSerializer.scala
@@ -13,15 +13,23 @@
package org.apache.pekko.persistence.serialization
+import java.io.NotSerializableException
+
import org.apache.pekko
import pekko.actor.ExtendedActorSystem
import pekko.annotation.InternalApi
import pekko.persistence.FilteredPayload
+import pekko.persistence.SerializedEvent
+import pekko.persistence.serialization.{ MessageFormats => mf }
+import pekko.protobufv3.internal.ByteString
+import pekko.protobufv3.internal.UnsafeByteOperations
import pekko.serialization.BaseSerializer
import pekko.serialization.SerializerWithStringManifest
/**
* INTERNAL API
+ *
+ * Serializer for persistence payload wrappers: `FilteredPayload` and
`SerializedEvent`.
*/
@InternalApi
final class FilteredPayloadSerializer(val system: ExtendedActorSystem)
@@ -29,17 +37,41 @@ final class FilteredPayloadSerializer(val system:
ExtendedActorSystem)
with BaseSerializer {
private val FilteredPayloadManifest = "F"
+ private val SerializedEventManifest = "S"
override def manifest(o: AnyRef): String = o match {
- case FilteredPayload => FilteredPayloadManifest
- case _ => throw new IllegalArgumentException(s"Can't
serialize object of type ${o.getClass}")
+ case FilteredPayload => FilteredPayloadManifest
+ case _: SerializedEvent => SerializedEventManifest
+ 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}")
+ case FilteredPayload => Array.empty
+ case se: SerializedEvent => payloadBuilder(se).build.toByteArray
+ case _ => throw new IllegalArgumentException(s"Can't
serialize object of type ${o.getClass}")
+ }
+
+ private def payloadBuilder(se: SerializedEvent):
mf.PersistentPayload.Builder = {
+ val builder = mf.PersistentPayload.newBuilder()
+ builder.setPayload(UnsafeByteOperations.unsafeWrap(se.bytes))
+ builder.setSerializerId(se.serializerId)
+ if (se.serializerManifest.nonEmpty)
builder.setPayloadManifest(ByteString.copyFromUtf8(se.serializerManifest))
+ builder
}
- override def fromBinary(bytes: Array[Byte], manifest: String): AnyRef =
FilteredPayload
+ override def fromBinary(bytes: Array[Byte], manifest: String): AnyRef =
manifest match {
+ case FilteredPayloadManifest => FilteredPayload
+ case SerializedEventManifest =>
serializedEvent(mf.PersistentPayload.parseFrom(bytes))
+ case _ => throw new
NotSerializableException(s"Unknown manifest [$manifest]")
+ }
+
+ private def serializedEvent(persistentPayload: mf.PersistentPayload):
SerializedEvent = {
+ val manifest =
+ if (persistentPayload.hasPayloadManifest)
+ persistentPayload.getPayloadManifest.toStringUtf8
+ else ""
+
+ new SerializedEvent(persistentPayload.getPayload.toByteArray,
persistentPayload.getSerializerId, manifest)
+ }
}
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
index 1858a5486c..0c464ed130 100644
---
a/persistence/src/test/scala/org/apache/pekko/persistence/serialization/FilteredPayloadSerializerSpec.scala
+++
b/persistence/src/test/scala/org/apache/pekko/persistence/serialization/FilteredPayloadSerializerSpec.scala
@@ -15,21 +15,68 @@ package org.apache.pekko.persistence.serialization
import org.apache.pekko
import pekko.persistence.FilteredPayload
+import pekko.persistence.SerializedEvent
import pekko.serialization.SerializationExtension
import pekko.serialization.SerializerWithStringManifest
+import pekko.serialization.Serializers
import pekko.testkit.PekkoSpec
class FilteredPayloadSerializerSpec extends PekkoSpec {
+ private val serialization = SerializationExtension(system)
+
"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)
}
+
+ "keep the existing serializer identifier and manifest for FilteredPayload"
in {
+ val serializer = serialization.findSerializerFor(FilteredPayload)
+ serializer.identifier shouldBe 34
+ Serializers.manifestFor(serializer, FilteredPayload) shouldBe "F"
+ }
+
+ "serialize SerializedEvent" in {
+ val event = "event1"
+ val eventSerializer = serialization.findSerializerFor(event)
+ val serializedEvent = new SerializedEvent(
+ bytes = eventSerializer.toBinary(event),
+ serializerId = eventSerializer.identifier,
+ serializerManifest = Serializers.manifestFor(eventSerializer, event))
+
+ val serializedEventSerializer =
serialization.findSerializerFor(serializedEvent)
+ serializedEventSerializer.identifier shouldBe 34
+ val serializedEventBytes = serialization.serialize(serializedEvent).get
+
+ val deserialized = serialization
+ .deserialize(
+ serializedEventBytes,
+ serializedEventSerializer.identifier,
+ Serializers.manifestFor(serializedEventSerializer, serializedEvent))
+ .get
+ .asInstanceOf[SerializedEvent]
+
+ deserialized.serializerId shouldBe eventSerializer.identifier
+ deserialized.serializerManifest shouldBe
serializedEvent.serializerManifest
+ val deserializedEvent =
+ serialization.deserialize(deserialized.bytes,
deserialized.serializerId, deserialized.serializerManifest).get
+ deserializedEvent shouldBe event
+ }
+
+ "serialize SerializedEvent with a non-empty serializer manifest" in {
+ val serializedEvent = new SerializedEvent(Array[Byte](1, 2, 3), 4711,
"my-manifest")
+ val serializer =
serialization.findSerializerFor(serializedEvent).asInstanceOf[SerializerWithStringManifest]
+ val deserialized = serializer
+ .fromBinary(serializer.toBinary(serializedEvent),
serializer.manifest(serializedEvent))
+ .asInstanceOf[SerializedEvent]
+ deserialized.bytes.toSeq shouldBe Seq[Byte](1, 2, 3)
+ deserialized.serializerId shouldBe 4711
+ deserialized.serializerManifest shouldBe "my-manifest"
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
