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 <[email protected]>
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: [email protected]
For additional commands, e-mail: [email protected]
