Manifest-less Jackson serializers, #28113 (#28299)

* This adds the ability to make Jackson serializers not output the class
  name in the manifest in cases where type information is stored in a more
  concise format in the JSON itself.

* documentation about rolling updates

* Ensure programatic bindings are search in Jackson serializer

akka-docs/src/main/paradox/serialization-jackson.md

@@ -391,6 +391,35 @@ different settings for remote messages and persisted events.
 
 @@snip [config](/akka-serialization-jackson/src/test/scala/doc/akka/serialization/jackson/SerializationDocSpec.scala) { #several-config }
 
+### Manifest-less serialization
+
+When using the Jackson serializer for persistence, given that the fully qualified class name is
+stored in the manifest, this can result in a lot of wasted disk and IO used, especially when the
+events are small. To address this, a `type-in-manifest` flag can be turned off, which will result
+in the class name not appearing in the manifest.
+
+When deserializing, the Jackson serializer will use the type defined in `deserialization-type`, if
+present, otherwise it will look for exactly one serialization binding class, and use that. For
+this to be useful, generally that single type must be a 
+@ref:[Polymorphic type](#polymorphic-types), with all type information necessary to deserialize to
+the various sub types contained in the JSON message.
+
+When switching serializers, for example, if doing a rolling update as described
+@ref:[here](additional/rolling-updates.md#from-java-serialization-to-jackson), there will be
+periods of time when you may have no serialization bindings declared for the type. In such
+circumstances, you must use the `deserialization-type` configuration attribute to specify which
+type should be used to deserialize messages.
+
+Since this configuration can only be applied to a single root type, you will usually only want to
+apply it to a per binding configuration, not to the regular `jackson-json` or `jackson-cbor`
+configurations.
+
+@@snip [config](/akka-serialization-jackson/src/test/scala/doc/akka/serialization/jackson/SerializationDocSpec.scala) { #manifestless }
+
+Note that Akka remoting already implements manifest compression, and so this optimization will have
+no significant impact for messages sent over remoting. It's only useful for messages serialized for
+other purposes, such as persistence or distributed data.
+
 ## Additional features
 
 Additional Jackson serialization features can be enabled/disabled in configuration. The default values from

akka-serialization-jackson/src/main/resources/reference.conf

@@ -144,6 +144,29 @@ akka.serialization.jackson {
     compress-larger-than = 0 KiB
   }
 
+  # Whether the type should be written to the manifest.
+  # If this is off, then either deserialization-type must be defined, or there must be exactly
+  # one serialization binding declared for this serializer, and the type in that binding will be
+  # used as the deserialization type. This feature will only work if that type either is a
+  # concrete class, or if it is a supertype that uses Jackson polymorphism (ie, the
+  # @JsonTypeInfo annotation) to store type information in the JSON itself. The intention behind
+  # disabling this is to remove extraneous type information (ie, fully qualified class names) when
+  # serialized objects are persisted in Akka persistence or replicated using Akka distributed
+  # data. Note that Akka remoting already has manifest compression optimizations that address this,
+  # so for types that just get sent over remoting, this offers no optimization.
+  type-in-manifest = on
+
+  # The type to use for deserialization.
+  # This is only used if type-in-manifest is disabled. If set, this type will be used to
+  # deserialize all messages. This is useful if the binding configuration you want to use when
+  # disabling type in manifest cannot be expressed as a single type. Examples of when you might
+  # use this include when changing serializers, so you don't want this serializer used for
+  # serialization and you haven't declared any bindings for it, but you still want to be able to
+  # deserialize messages that were serialized with this serializer, as well as situations where
+  # you only want some sub types of a given Jackson polymorphic type to be serialized using this
+  # serializer.
+  deserialization-type = ""
+
   # Specific settings for jackson-json binding can be defined in this section to
   # override the settings in 'akka.serialization.jackson'
   jackson-json {}

akka-serialization-jackson/src/main/scala/akka/serialization/jackson/JacksonSerializer.scala

@@ -14,7 +14,6 @@ import scala.annotation.tailrec
 import scala.util.Failure
 import scala.util.Success
 import scala.util.control.NonFatal
-
 import akka.actor.ExtendedActorSystem
 import akka.annotation.InternalApi
 import akka.event.LogMarker
@@ -165,10 +164,46 @@ import com.fasterxml.jackson.dataformat.cbor.CBORFactory
     import akka.util.ccompat.JavaConverters._
     conf.getStringList("whitelist-class-prefix").asScala.toVector
   }
+  private val typeInManifest: Boolean = conf.getBoolean("type-in-manifest")
+  // Calculated eagerly so as to fail fast
+  private val configuredDeserializationType: Option[Class[_ <: AnyRef]] = conf.getString("deserialization-type") match {
+    case "" => None
+    case className =>
+      system.dynamicAccess.getClassFor[AnyRef](className) match {
+        case Success(c) => Some(c)
+        case Failure(_) =>
+          throw new IllegalArgumentException(
+            s"Cannot find deserialization-type [$className] for Jackson serializer [$bindingName]")
+      }
+  }
 
   // This must lazy otherwise it will deadlock the ActorSystem creation
   private lazy val serialization = SerializationExtension(system)
 
+  // This must be lazy since it depends on serialization above
+  private lazy val deserializationType: Option[Class[_ <: AnyRef]] = if (typeInManifest) {
+    None
+  } else {
+    configuredDeserializationType.orElse {
+      val bindings = serialization.bindings.filter(_._2.identifier == identifier)
+      bindings match {
+        case Nil =>
+          throw new IllegalArgumentException(
+            s"Jackson serializer [$bindingName] with type-in-manifest disabled must either declare" +
+            " a deserialization-type or have exactly one binding configured, but none were configured")
+
+        case Seq((clazz, _)) =>
+          Some(clazz.asSubclass(classOf[AnyRef]))
+
+        case multiple =>
+          throw new IllegalArgumentException(
+            s"Jackson serializer [$bindingName] with type-in-manifest disabled must either declare" +
+            " a deserialization-type or have exactly one binding configured, but multiple bindings" +
+            s" were configured [${multiple.mkString(", ")}]")
+      }
+    }
+  }
+
   // doesn't have to be volatile, doesn't matter if check is run more than once
   private var serializationBindingsCheckedOk = false
 
@@ -176,6 +211,13 @@ import com.fasterxml.jackson.dataformat.cbor.CBORFactory
 
   override def manifest(obj: AnyRef): String = {
     checkAllowedSerializationBindings()
+    deserializationType match {
+      case Some(clazz) =>
+        migrations.get(clazz.getName) match {
+          case Some(transformer) => "#" + transformer.currentVersion
+          case None              => ""
+        }
+      case None =>
         val className = obj.getClass.getName
         checkAllowedClassName(className)
         checkAllowedClass(obj.getClass)
@@ -184,6 +226,7 @@ import com.fasterxml.jackson.dataformat.cbor.CBORFactory
           case None              => className
         }
     }
+  }
 
   override def toBinary(obj: AnyRef): Array[Byte] = {
     checkAllowedSerializationBindings()
@@ -220,9 +263,9 @@ import com.fasterxml.jackson.dataformat.cbor.CBORFactory
     val startTime = if (isDebugEnabled) System.nanoTime else 0L
 
     val (fromVersion, manifestClassName) = parseManifest(manifest)
-    checkAllowedClassName(manifestClassName)
+    if (typeInManifest) checkAllowedClassName(manifestClassName)
 
-    val migration = migrations.get(manifestClassName)
+    val migration = migrations.get(deserializationType.fold(manifestClassName)(_.getName))
 
     val className = migration match {
       case Some(transformer) if fromVersion < transformer.currentVersion =>
@@ -234,7 +277,7 @@ import com.fasterxml.jackson.dataformat.cbor.CBORFactory
       case _ => manifestClassName
     }
 
-    if (className ne manifestClassName)
+    if (typeInManifest && (className ne manifestClassName))
       checkAllowedClassName(className)
 
     if (isCaseObject(className)) {
@@ -250,13 +293,15 @@ import com.fasterxml.jackson.dataformat.cbor.CBORFactory
       logFromBinaryDuration(bytes, bytes, startTime, clazz)
       result
     } else {
-      val clazz = system.dynamicAccess.getClassFor[AnyRef](className) match {
+      val clazz = deserializationType.getOrElse {
+        system.dynamicAccess.getClassFor[AnyRef](className) match {
           case Success(c) => c
           case Failure(_) =>
             throw new NotSerializableException(
               s"Cannot find manifest class [$className] for serializer [${getClass.getName}].")
         }
-      checkAllowedClass(clazz)
+      }
+      if (typeInManifest) checkAllowedClass(clazz)
 
       val decompressedBytes = decompress(bytes)
 

akka-serialization-jackson/src/test/scala/akka/serialization/jackson/JacksonSerializerSpec.scala

@@ -478,6 +478,70 @@ class JacksonJsonSerializerSpec extends JacksonSerializerSpec("jackson-json") {
       JacksonSerializer.isGZipped(bytes) should ===(false)
     }
   }
+
+  "JacksonJsonSerializer without type in manifest" should {
+    import ScalaTestMessages._
+
+    "deserialize messages using the serialization bindings" in withSystem(
+      """
+        akka.actor {
+          serializers.animal = "akka.serialization.jackson.JacksonJsonSerializer"
+          serialization-identifiers.animal = 9091
+          serialization-bindings {
+            "akka.serialization.jackson.ScalaTestMessages$Animal" = animal
+          }
+        }
+        akka.serialization.jackson.animal.type-in-manifest = off
+        """) { sys =>
+      val msg = Elephant("Dumbo", 1)
+      val serializer = serializerFor(msg, sys)
+      serializer.manifest(msg) should ===("")
+      val bytes = serializer.toBinary(msg)
+      val deserialized = serializer.fromBinary(bytes, "")
+      deserialized should ===(msg)
+    }
+
+    "deserialize messages using the configured deserialization type" in withSystem(
+      """
+        akka.actor {
+          serializers.animal = "akka.serialization.jackson.JacksonJsonSerializer"
+          serialization-identifiers.animal = 9091
+          serialization-bindings {
+            "akka.serialization.jackson.ScalaTestMessages$Elephant" = animal
+            "akka.serialization.jackson.ScalaTestMessages$Lion" = animal
+          }
+        }
+        akka.serialization.jackson.animal {
+          type-in-manifest = off
+          deserialization-type = "akka.serialization.jackson.ScalaTestMessages$Animal"
+        }
+      """) { sys =>
+      val msg = Elephant("Dumbo", 1)
+      val serializer = serializerFor(msg, sys)
+      serializer.manifest(msg) should ===("")
+      val bytes = serializer.toBinary(msg)
+      val deserialized = serializer.fromBinary(bytes, "")
+      deserialized should ===(msg)
+    }
+
+    "fail if multiple serialization bindings are declared with no deserialization type" in {
+      an[IllegalArgumentException] should be thrownBy {
+        withSystem("""
+        akka.actor {
+          serializers.animal = "akka.serialization.jackson.JacksonJsonSerializer"
+          serialization-identifiers.animal = 9091
+          serialization-bindings {
+            "akka.serialization.jackson.ScalaTestMessages$Elephant" = animal
+            "akka.serialization.jackson.ScalaTestMessages$Lion" = animal
+          }
+        }
+        akka.serialization.jackson.animal {
+          type-in-manifest = off
+        }
+      """)(sys => checkSerialization(Elephant("Dumbo", 1), sys))
+      }
+    }
+  }
 }
 
 abstract class JacksonSerializerSpec(serializerName: String)

akka-serialization-jackson/src/test/scala/doc/akka/serialization/jackson/SerializationDocSpec.scala

@@ -101,6 +101,31 @@ object SerializationDocSpec {
     #//#several-config
   """
 
+  val configManifestless = """
+    #//#manifestless
+    akka.actor {
+      serializers {
+        jackson-json-event = "akka.serialization.jackson.JacksonJsonSerializer"
+      }
+      serialization-identifiers {
+        jackson-json-event = 9001
+      }
+      serialization-bindings {
+        "com.myservice.MyEvent" = jackson-json-event
+      }
+    }
+    akka.serialization.jackson {
+      jackson-json-event {
+        type-in-manifest = off
+        # Since there is exactly one serialization binding declared for this
+        # serializer above, this is optional, but if there were none or many,
+        # this would be mandatory.
+        deserialization-type = "com.myservice.MyEvent"
+      }
+    }
+    #//#manifestless
+  """
+
   //#polymorphism
   final case class Zoo(primaryAttraction: Animal) extends MySerializable