update kryo and use compatibility serializer for future message format changes

Author: Brendan Doyle

core/scheduler/build.gradle

@@ -69,7 +69,7 @@ dependencies {
     }
     implementation "org.scala-lang:scala-library:${gradle.scala.version}"
     implementation project(':common:scala')
-    implementation "io.altoo:akka-kryo-serialization_${gradle.scala.depVersion}:1.0.0"
+    implementation "io.altoo:akka-kryo-serialization_${gradle.scala.depVersion}:2.5.0"
     implementation "com.lightbend.akka.management:akka-management-cluster-bootstrap_${gradle.scala.depVersion}:${gradle.akka_management.version}"
     implementation "com.lightbend.akka.discovery:akka-discovery-kubernetes-api_${gradle.scala.depVersion}:${gradle.akka_management.version}"
 }

core/scheduler/src/main/resources/application.conf

@@ -53,6 +53,8 @@ akka {
   }
 }
 
+akka-kryo-serialization.kryo-initializer = "org.apache.openwhisk.core.scheduler.CompatibleKryoInitializer"
+
 whisk {
   cluster {
     use-cluster-bootstrap: false

core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/Scheduler.scala

@@ -18,14 +18,16 @@
 package org.apache.openwhisk.core.scheduler
 
 import akka.Done
-import akka.actor.{ActorRef, ActorRefFactory, ActorSelection, ActorSystem, CoordinatedShutdown, Props}
+import akka.actor.{ActorRef, ActorRefFactory, ActorSelection, ActorSystem, CoordinatedShutdown, ExtendedActorSystem, Props}
 import akka.http.scaladsl.Http
 import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
 import akka.management.scaladsl.AkkaManagement
 import akka.management.cluster.bootstrap.ClusterBootstrap
 import akka.pattern.ask
 import akka.util.Timeout
 import com.typesafe.config.ConfigValueFactory
+import io.altoo.akka.serialization.kryo.DefaultKryoInitializer
+import io.altoo.akka.serialization.kryo.serializer.scala.ScalaKryo
 import kamon.Kamon
 import org.apache.openwhisk.common.Https.HttpsConfig
 import org.apache.openwhisk.common._
@@ -425,3 +427,9 @@ case class SchedulingConfig(staleThreshold: FiniteDuration,
                             dropInterval: FiniteDuration,
                             allowOverProvisionBeforeThrottle: Boolean,
                             namespaceOverProvisionBeforeThrottleRatio: Double)
+
+class CompatibleKryoInitializer extends DefaultKryoInitializer {
+  override def preInit(kryo: ScalaKryo, system: ExtendedActorSystem): Unit = {
+    kryo.setDefaultSerializer(classOf[com.esotericsoftware.kryo.serializers.CompatibleFieldSerializer[_]])
+  }
+}