Detect duplicate serializer identifiers, #29100

* Detect duplicate serializer identifiers, #29100
* and expand the documentation for the serializer identifier
* and mention BaseSerializer to read the identifier from config
* same class same id is ok

akka-actor-tests/src/test/scala/akka/serialization/SerializeSpec.scala

@@ -270,6 +270,29 @@ class SerializeSpec extends AkkaSpec(SerializationTests.serializeConf) {
         ser.serialize(new Other).get
       }
     }
+
+    "detect duplicate serializer ids" in {
+      (intercept[IllegalArgumentException] {
+        val sys = ActorSystem(
+          "SerializeSpec",
+          ConfigFactory.parseString(s"""
+          akka {
+            actor {
+              serializers {
+                test = "akka.serialization.NoopSerializer"
+                test-same = "akka.serialization.NoopSerializerSameId"
+              }
+      
+              serialization-bindings {
+                "akka.serialization.SerializationTests$$Person" = test
+                "akka.serialization.SerializationTests$$Address" = test-same
+              }
+            }
+          }
+          """))
+        shutdown(sys)
+      }.getMessage should include).regex("Serializer identifier \\[9999\\].*is not unique")
+    }
   }
 }
 
@@ -578,6 +601,8 @@ protected[akka] class NoopSerializer2 extends Serializer {
   def fromBinary(bytes: Array[Byte], clazz: Option[Class[_]]): AnyRef = null
 }
 
+protected[akka] class NoopSerializerSameId extends NoopSerializer
+
 @SerialVersionUID(1)
 protected[akka] final case class FakeThrowable(msg: String) extends Throwable(msg) with Serializable {
   override def fillInStackTrace = null

akka-actor/src/main/scala/akka/serialization/Serialization.scala

@@ -503,8 +503,21 @@ class Serialization(val system: ExtendedActorSystem) extends Extension {
   /**
    * Maps from a Serializer Identity (Int) to a Serializer instance (optimization)
    */
-  val serializerByIdentity: Map[Int, Serializer] =
-    Map(NullSerializer.identifier -> NullSerializer) ++ serializers.map { case (_, v) => (v.identifier, v) }
+  val serializerByIdentity: Map[Int, Serializer] = {
+    val zero: Map[Int, Serializer] = Map(NullSerializer.identifier -> NullSerializer)
+    serializers.foldLeft(zero) {
+      case (acc, (_, ser)) =>
+        val id = ser.identifier
+        acc.get(id) match {
+          case Some(existing) if existing != ser =>
+            throw new IllegalArgumentException(
+              s"Serializer identifier [$id] of [${ser.getClass.getName}] " +
+              s"is not unique. It is also used by [${acc(id).getClass.getName}].")
+          case _ =>
+            acc.updated(id, ser)
+        }
+    }
+  }
 
   /**
    * Serializers with id 0 - 1023 are stored in an array for quick allocation free access

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

@@ -82,6 +82,7 @@ Scala
 Java
 :  @@snip [SerializationDocTest.java](/akka-docs/src/test/java/jdocs/serialization/SerializationDocTest.java) { #programmatic }
 
+
 The manifest is a type hint so that the same serializer can be used for different classes.
 
 Note that when deserializing from bytes the manifest and the identifier of the serializer are needed.
@@ -119,6 +120,21 @@ Scala
 Java
 :  @@snip [SerializationDocTest.java](/akka-docs/src/test/java/jdocs/serialization/SerializationDocTest.java) { #my-own-serializer }
 
+The `identifier` must be unique. The identifier is used when selecting which serializer to use for deserialization.
+If you have accidentally configured several serializers with the same identifier that will be detected and prevent
+the `ActorSystem` from being started. It can be a hardcoded value because it must remain the same value to support
+rolling updates. 
+
+@@@ div { .group-scala }
+
+If you prefer to define the identifier in cofiguration that is supported by the `BaseSerializer` trait, which
+implements the `def identifier` by reading it from configuration based on the serializer's class name:
+
+Scala
+:  @@snip [SerializationDocSpec.scala](/akka-docs/src/test/scala/docs/serialization/SerializationDocSpec.scala) { #serialization-identifiers-config }
+
+@@@
+
 The manifest is a type hint so that the same serializer can be used for different
 classes. The manifest parameter in @scala[`fromBinary`]@java[`fromBinaryJava`] is the class of the object that
 was serialized. In `fromBinary` you can match on the class and deserialize the

akka-docs/src/test/scala/docs/serialization/SerializationDocSpec.scala

@@ -109,6 +109,21 @@ package docs.serialization {
    */
   trait JsonSerializable
 
+  object SerializerIdConfig {
+    val config =
+      """
+        #//#serialization-identifiers-config
+        akka {
+          actor {
+            serialization-identifiers {
+              "docs.serialization.MyOwnSerializer" = 1234567
+            }
+          }
+        }
+        #//#serialization-identifiers-config
+        """
+  }
+
   class SerializationDocSpec extends AkkaSpec {
     "demonstrate configuration of serialize messages" in {
       val config = ConfigFactory.parseString("""