support reading akka-persistence snapshots (apache#837)

* support reading akka-persistence snapshots

* add test

* Update SnapshotSerializerSpec.scala

* Update SnapshotSerializerSpec.scala

persistence/src/main/scala/org/apache/pekko/persistence/serialization/SnapshotSerializer.scala

@@ -112,7 +112,15 @@ class SnapshotSerializer(val system: ExtendedActorSystem) extends BaseSerializer
 
     val (serializerId, manifest) = headerFromBinary(headerBytes)
 
-    serialization.deserialize(snapshotBytes, serializerId, manifest).get
+    // suggested in https://github.com/scullxbones/pekko-persistence-mongo/pull/14#issuecomment-1847223850
+    serialization
+      .deserialize(snapshotBytes, serializerId, manifest)
+      .recover {
+        case _: NotSerializableException if manifest.startsWith("akka") =>
+          serialization
+            .deserialize(snapshotBytes, serializerId, manifest.replaceFirst("akka", "org.apache.pekko"))
+      }
+      .get
   }
 
   private def writeInt(out: OutputStream, i: Int): Unit = {

persistence/src/test/scala/org/apache/pekko/persistence/serialization/SnapshotSerializerSpec.scala

@@ -0,0 +1,48 @@
+/*
+ * 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.serialization
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.persistence.fsm.PersistentFSM.PersistentFSMSnapshot
+import pekko.serialization.SerializationExtension
+import pekko.testkit.PekkoSpec
+
+import java.util.Base64
+import scala.util.Success
+
+class SnapshotSerializerSpec extends PekkoSpec {
+
+  "Snapshot serializer" should {
+    "deserialize akka snapshots" in {
+      val system = ActorSystem()
+      val serialization = SerializationExtension(system)
+      // https://github.com/apache/incubator-pekko/pull/837#issuecomment-1847320309
+      val data =
+        "PAAAAAcAAABha2thLnBlcnNpc3RlbmNlLmZzbS5QZXJzaXN0ZW50RlNNJFBlcnNpc3RlbnRGU01TbmFwc2hvdAoPdGVzdC1pZGVudGlmaWVyEg0IFBIJdGVzdC1kYXRh"
+      val bytes = Base64.getDecoder.decode(data)
+      val result = serialization.deserialize(bytes, classOf[Snapshot]).get
+      val deserialized = result.data
+      deserialized shouldBe a[Success[_]]
+      val innerResult = deserialized.asInstanceOf[Success[_]].get
+      innerResult shouldBe a[PersistentFSMSnapshot[_]]
+      val persistentFSMSnapshot = innerResult.asInstanceOf[PersistentFSMSnapshot[_]]
+      persistentFSMSnapshot shouldEqual PersistentFSMSnapshot[String]("test-identifier", "test-data", None)
+    }
+  }
+}