You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by fa...@apache.org on 2026/09/06 08:01:01 UTC
(pekko) branch main updated: remove the Akka 2.6.4 rolling-migration manifests from ClusterMessageSerializer (#3516)
This is an automated email from the ASF dual-hosted git repository.
pjfanning 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 0c1de3363f remove the Akka 2.6.4 rolling-migration manifests from ClusterMessageSerializer (#3516)
0c1de3363f is described below
commit 0c1de3363f1617bdef8dd7faf46008d23869ae20
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Sun Sep 6 09:00:55 2026 +0100
remove the Akka 2.6.4 rolling-migration manifests from ClusterMessageSerializer (#3516)
Motivation:
The Old* and *Pre2523 manifest vals were kept "for one version iteration
from Akka 2.6.4" to allow rolling migration to short manifests, removable
"in Akka 2.6.6 or later". No peer a Pekko node can talk to emits them:
Pekko has always emitted the short manifests, and an actual old Akka node
would emit akka.cluster.* strings, which these org.apache.pekko.cluster.*
constants (renamed in the fork) can never match. Each dead branch was a
wire-reachable deserialization path.
Modification:
Remove the vals, their fromBinary cases, and the two heartbeat
deserializers only those cases used. Replace the spec's old-manifest
round-trip tests and their now-unused helpers with a test asserting every
removed manifest is rejected as unknown.
Result:
The class-based manifests are refused with "Unknown manifest". Short
manifests are unchanged.
Tests:
- sbt "cluster/testOnly org.apache.pekko.cluster.protobuf.ClusterMessageSerializerSpec" - 8 passed
- the new rejection test fails without the production change (the manifest
routes to deserializeJoin and raises InvalidProtocolBufferException)
- sbt "cluster/mimaReportBinaryIssues" - no issues
- sbt "cluster/scalafmtCheckAll" - clean
References:
None - removes dead Akka 2.6.4 compatibility code
---
.../protobuf/ClusterMessageSerializer.scala | 40 +-------
.../protobuf/ClusterMessageSerializerSpec.scala | 105 +++++----------------
2 files changed, 26 insertions(+), 119 deletions(-)
diff --git a/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala b/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala
index 23a6f77ff0..b4e5db8f6e 100644
--- a/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala
+++ b/cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala
@@ -40,22 +40,6 @@ import com.typesafe.config.{ Config, ConfigFactory, ConfigRenderOptions }
*/
@InternalApi
private[pekko] object ClusterMessageSerializer {
- // Kept for one version iteration from Akka 2.6.4 to allow rolling migration to short manifests
- // can be removed in Akka 2.6.6 or later.
- val OldJoinManifest = s"org.apache.pekko.cluster.InternalClusterAction$$Join"
- val OldWelcomeManifest = s"org.apache.pekko.cluster.InternalClusterAction$$Welcome"
- val OldLeaveManifest = s"org.apache.pekko.cluster.ClusterUserAction$$Leave"
- val OldDownManifest = s"org.apache.pekko.cluster.ClusterUserAction$$Down"
- val OldInitJoinManifest = s"org.apache.pekko.cluster.InternalClusterAction$$InitJoin$$"
- val OldInitJoinAckManifest = s"org.apache.pekko.cluster.InternalClusterAction$$InitJoinAck"
- val OldInitJoinNackManifest = s"org.apache.pekko.cluster.InternalClusterAction$$InitJoinNack"
- val HeartBeatManifestPre2523 = s"org.apache.pekko.cluster.ClusterHeartbeatSender$$Heartbeat"
- val HeartBeatRspManifest2523 = s"org.apache.pekko.cluster.ClusterHeartbeatSender$$HeartbeatRsp"
- val OldExitingConfirmedManifest = s"org.apache.pekko.cluster.InternalClusterAction$$ExitingConfirmed"
- val OldGossipStatusManifest = "org.apache.pekko.cluster.GossipStatus"
- val OldGossipEnvelopeManifest = "org.apache.pekko.cluster.GossipEnvelope"
- val OldClusterRouterPoolManifest = "org.apache.pekko.cluster.routing.ClusterRouterPool"
-
// is handled on the deserializing side in Akka 2.6.2 and then on the serializing side in Akka 2.6.3
val JoinManifest = "J"
val WelcomeManifest = "W"
@@ -139,21 +123,7 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
case DownManifest => deserializeDown(bytes)
case ExitingConfirmedManifest => deserializeExitingConfirmed(bytes)
case ClusterRouterPoolManifest => deserializeClusterRouterPool(bytes)
- // needs to stay in Akka 2.6.5 to be able to talk to an Akka 2.5.{3,4} node during rolling upgrade
- case HeartBeatManifestPre2523 => deserializeHeartBeatAsAddress(bytes)
- case HeartBeatRspManifest2523 => deserializeHeartBeatRspAsUniqueAddress(bytes)
- case OldGossipStatusManifest => deserializeGossipStatus(bytes)
- case OldGossipEnvelopeManifest => deserializeGossipEnvelope(bytes)
- case OldInitJoinManifest => deserializeInitJoin(bytes)
- case OldInitJoinAckManifest => deserializeInitJoinAck(bytes)
- case OldInitJoinNackManifest => deserializeInitJoinNack(bytes)
- case OldJoinManifest => deserializeJoin(bytes)
- case OldWelcomeManifest => deserializeWelcome(bytes)
- case OldLeaveManifest => deserializeLeave(bytes)
- case OldDownManifest => deserializeDown(bytes)
- case OldExitingConfirmedManifest => deserializeExitingConfirmed(bytes)
- case OldClusterRouterPoolManifest => deserializeClusterRouterPool(bytes)
- case _ => throw new IllegalArgumentException(s"Unknown manifest [$manifest]")
+ case _ => throw new IllegalArgumentException(s"Unknown manifest [$manifest]")
}
def compress(msg: MessageLite): Array[Byte] = {
@@ -323,14 +293,6 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
InternalClusterAction.ExitingConfirmed(uniqueAddressFromBinary(bytes))
}
- private def deserializeHeartBeatRspAsUniqueAddress(bytes: Array[Byte]): ClusterHeartbeatSender.HeartbeatRsp = {
- ClusterHeartbeatSender.HeartbeatRsp(uniqueAddressFromBinary(bytes), -1, -1)
- }
-
- private def deserializeHeartBeatAsAddress(bytes: Array[Byte]): ClusterHeartbeatSender.Heartbeat = {
- ClusterHeartbeatSender.Heartbeat(addressFromBinary(bytes), -1, -1)
- }
-
def deserializeHeartBeat(bytes: Array[Byte]): ClusterHeartbeatSender.Heartbeat = {
val hb = cm.Heartbeat.parseFrom(bytes)
ClusterHeartbeatSender.Heartbeat(addressFromProto(hb.getFrom), hb.getSequenceNr, hb.getCreationTime)
diff --git a/cluster/src/test/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializerSpec.scala b/cluster/src/test/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializerSpec.scala
index ec993a6ead..14d0974e40 100644
--- a/cluster/src/test/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializerSpec.scala
+++ b/cluster/src/test/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializerSpec.scala
@@ -56,26 +56,6 @@ class ClusterMessageSerializerSpec extends PekkoSpec("pekko.actor.provider = clu
}
}
- private def roundtripWithManifest[T <: AnyRef](obj: T, manifest: String): T = {
- val blob = serializer.toBinary(obj)
- serializer.fromBinary(blob, manifest).asInstanceOf[T]
- }
-
- private def checkDeserializationWithManifest(obj: AnyRef, deserializationManifest: String): Unit = {
- (obj, roundtripWithManifest(obj, deserializationManifest)) match {
- case (env: GossipEnvelope, env2: GossipEnvelope) =>
- env2.from should ===(env.from)
- env2.to should ===(env.to)
- env2.gossip should ===(env.gossip)
- env.gossip.members.foreach { m1 =>
- val m2 = env.gossip.members.find(_.uniqueAddress == m1.uniqueAddress).get
- checkSameMember(m1, m2)
- }
- case (_, ref) =>
- ref should ===(obj)
- }
- }
-
private def checkSameMember(m1: Member, m2: Member): Unit = {
m1.uniqueAddress should ===(m2.uniqueAddress)
m1.status should ===(m2.status)
@@ -138,50 +118,32 @@ class ClusterMessageSerializerSpec extends PekkoSpec("pekko.actor.provider = clu
checkSerialization(InternalClusterAction.Welcome(uniqueAddress, g2))
}
- // can be removed in Akka 2.6.3 only checks deserialization with new not yet in effect manifests for Akka 2.6.2
- "be de-serializable with class manifests from Akka 2.6.4 and earlier nodes" in {
+ "reject the class-based manifests that Akka 2.6.4 and earlier nodes sent" in {
+ // these were kept "for one version iteration from Akka 2.6.4" to allow rolling
+ // migration to the short manifests; nothing a Pekko node can talk to emits them,
+ // so they must be refused as unknown rather than kept as wire-reachable paths
val address = Address("pekko", "system", "some.host.org", 4711)
- val uniqueAddress = UniqueAddress(address, 17L)
- val address2 = Address("pekko", "system", "other.host.org", 4711)
- val uniqueAddress2 = UniqueAddress(address2, 18L)
- checkDeserializationWithManifest(
- InternalClusterAction.Join(uniqueAddress, Set("foo", "bar", "dc-A"), Version.Zero),
- ClusterMessageSerializer.OldJoinManifest)
- checkDeserializationWithManifest(ClusterUserAction.Leave(address), ClusterMessageSerializer.LeaveManifest)
- checkDeserializationWithManifest(ClusterUserAction.Down(address), ClusterMessageSerializer.DownManifest)
- checkDeserializationWithManifest(
- InternalClusterAction.InitJoin(ConfigFactory.empty),
- ClusterMessageSerializer.OldInitJoinManifest)
- checkDeserializationWithManifest(
- InternalClusterAction.InitJoinAck(address, CompatibleConfig(ConfigFactory.empty)),
- ClusterMessageSerializer.OldInitJoinAckManifest)
- checkDeserializationWithManifest(
- InternalClusterAction.InitJoinNack(address),
- ClusterMessageSerializer.OldInitJoinNackManifest)
- checkDeserializationWithManifest(
- InternalClusterAction.ExitingConfirmed(uniqueAddress),
- ClusterMessageSerializer.OldExitingConfirmedManifest)
-
- val node1 = VectorClock.Node("node1")
- val node2 = VectorClock.Node("node2")
- val node3 = VectorClock.Node("node3")
- val node4 = VectorClock.Node("node4")
- val g1 = (Gossip(SortedSet(a1, b1, c1, d1)) :+ node1 :+ node2).seen(a1.uniqueAddress).seen(b1.uniqueAddress)
- val g2 = (g1 :+ node3 :+ node4).seen(a1.uniqueAddress).seen(c1.uniqueAddress)
- val reachability3 = Reachability.empty
- .unreachable(a1.uniqueAddress, e1.uniqueAddress)
- .unreachable(b1.uniqueAddress, e1.uniqueAddress)
- checkDeserializationWithManifest(
- GossipEnvelope(a1.uniqueAddress, uniqueAddress2, g1),
- ClusterMessageSerializer.OldGossipEnvelopeManifest)
-
- checkDeserializationWithManifest(
- GossipStatus(a1.uniqueAddress, g1.version, g1.seenDigest),
- ClusterMessageSerializer.OldGossipStatusManifest)
-
- checkDeserializationWithManifest(
- InternalClusterAction.Welcome(uniqueAddress, g2),
- ClusterMessageSerializer.OldWelcomeManifest)
+ val bytes = serializer.toBinary(ClusterUserAction.Leave(address))
+ Seq(
+ "org.apache.pekko.cluster.InternalClusterAction$Join",
+ "org.apache.pekko.cluster.InternalClusterAction$Welcome",
+ "org.apache.pekko.cluster.ClusterUserAction$Leave",
+ "org.apache.pekko.cluster.ClusterUserAction$Down",
+ "org.apache.pekko.cluster.InternalClusterAction$InitJoin$",
+ "org.apache.pekko.cluster.InternalClusterAction$InitJoinAck",
+ "org.apache.pekko.cluster.InternalClusterAction$InitJoinNack",
+ "org.apache.pekko.cluster.ClusterHeartbeatSender$Heartbeat",
+ "org.apache.pekko.cluster.ClusterHeartbeatSender$HeartbeatRsp",
+ "org.apache.pekko.cluster.InternalClusterAction$ExitingConfirmed",
+ "org.apache.pekko.cluster.GossipStatus",
+ "org.apache.pekko.cluster.GossipEnvelope",
+ "org.apache.pekko.cluster.routing.ClusterRouterPool").foreach { manifest =>
+ withClue(s"manifest [$manifest]: ") {
+ intercept[IllegalArgumentException] {
+ serializer.fromBinary(bytes, manifest)
+ }.getMessage should include("Unknown manifest")
+ }
+ }
}
"reject gossip that refers to a lookup table entry it did not send" in {
@@ -300,23 +262,6 @@ class ClusterMessageSerializerSpec extends PekkoSpec("pekko.actor.provider = clu
}
}
- // support for deserializing a new format with a string based manifest was added in Akka 2.5.23 but the next step
- // was never done, meaning that 2.6.4 still emits the old format
- "Rolling upgrades for heart beat message changes in Akka 2.5.23" must {
-
- "deserialize heart beats represented by just an address Address to support versions prior to Akka 2.6.5" in {
- val serialized = serializer.addressToProto(a1.address).build().toByteArray
- val deserialized = serializer.fromBinary(serialized, ClusterMessageSerializer.HeartBeatManifestPre2523)
- deserialized should ===(ClusterHeartbeatSender.Heartbeat(a1.address, -1, -1))
- }
-
- "deserialize heart beat responses as UniqueAddress to support versions prior to 2.5.23" in {
- val serialized = serializer.uniqueAddressToProto(a1.uniqueAddress).build().toByteArray
- val deserialized = serializer.fromBinary(serialized, ClusterMessageSerializer.HeartBeatRspManifest2523)
- deserialized should ===(ClusterHeartbeatSender.HeartbeatRsp(a1.uniqueAddress, -1, -1))
- }
- }
-
"Cluster router pool" must {
"be serializable with no role" in {
checkSerialization(
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org