[!463] - Feature/580 upgrade akka in order to be compatible with java 17

# New features and improvements

Change the Akka version to be compatible with Java 17.
List of dependency updates:
- akka = "2.6.21"
- akkaHttp = "10.2.10"
- typesafeConfig = "1.4.2"
- akkaKryo = "1.1.5"

New dependency: akka-cluster-metrics

Removed dependency: akka-contrib

Change akkaKryo 

from: "**com.github.romix.akka**" %% "akka-kryo-serialization" 

to: "**io.altoo**" %% "akka-kryo-serialization"

Akka Remoting has a new behavior called Artery, but we keep the **classic** behavior in this feature.

# Breaking changes

- Scala 2.11 is no longer supported
- Prefix serializer "com.romix.akka.serialization.*" in configs not working anymore 
- Akka Remoting has a new behavior called Artery, so the configuration akka.remote.* is not working anymore, and also artery is enabled by default

# Migration

Applications using this WASP version should pay attention to configuration changes.

- New prefix serializer: io.altoo.akka.serialization.*
- akka.remote configuration changes needed: 
  - akka.remote.artery.enabled = false
  - other akka.remote.* configs must be changed to akka.remote.classic.*

For other clarification, here are the Akka migration guides: 

- https://doc.akka.io/libraries/akka-core/2.5/project/migration-guide-2.4.x-2.5.x.html
- https://doc.akka.io/libraries/akka-core/2.6/project/migration-guide-2.5.x-2.6.x.html


# Bug fixes

none

# How this feature was tested

Unit test

# Related issue

Closes #580

Author: Matteo Seclì

consumers-spark/src/main/resources/reference.conf

@@ -14,7 +14,7 @@ wasp {
     actor {
 
       serializers {
-        kryo = "com.romix.akka.serialization.kryo.KryoSerializer"
+        kryo = "io.altoo.akka.serialization.kryo.KryoSerializer"
       }
 
       kryo {
@@ -66,7 +66,8 @@ wasp {
 
     }
     remote {
-      netty.tcp {
+      artery.enabled = false
+      classic.netty.tcp {
         port = 0
         hostname = "localhost"
       }

consumers-spark/src/main/scala/it/agilelab/bigdata/wasp/consumers/spark/streaming/actor/master/SparkConsumersStreamingMasterGuardian.scala

@@ -146,7 +146,7 @@ class SparkConsumersStreamingMasterGuardian(
       unstashAll()
       log.debug("Unstashing")
       log.info(s"Setting ${Timers.unschedulableCheck} to recover unschedulable pipegraphs")
-      setTimer(Timers.unschedulableCheck, RecoverUnschedulable, unschedulableCheckInterval, repeat = true)
+      startTimerAtFixedRate(Timers.unschedulableCheck, RecoverUnschedulable, unschedulableCheckInterval)
   }
 
   private def handleUnschedulable: StateFunction = {
@@ -351,7 +351,7 @@ class SparkConsumersStreamingMasterGuardian(
 
     case Event(_: ChildProtocol.WorkNotCancelled, _: Schedule)
     =>
-      setTimer(
+      startTimerAtFixedRate(
         Timers.workNotCancelledRetryTimer,
         RetryEnvelope(ChildProtocol.WorkNotCancelled, sender()),
         retryInterval

consumers-spark/src/main/scala/it/agilelab/bigdata/wasp/consumers/spark/streaming/actor/pipegraph/PipegraphGuardian.scala

@@ -305,7 +305,7 @@ class PipegraphGuardian(private val master: ActorRef,
     case (Activating, Activating) => nextStateData match {
       case ActivatingData.ShouldRetry() =>
         log.info("[Activating->Activating] Scheduling Retry")
-        setTimer("retry", MyProtocol.PerformRetry, retryDuration)
+        startTimerAtFixedRate("retry", MyProtocol.PerformRetry, retryDuration)
       case ActivatingData.ToBeActivated(etl) =>
         log.info("[Activating->Activating] Activating [{}]", etl.name)
         self ! MyProtocol.ActivateETL(etl)
@@ -386,7 +386,7 @@ class PipegraphGuardian(private val master: ActorRef,
     case (Materializing, Materializing) => nextStateData match {
       case MaterializingData.ShouldRetry() =>
         log.info("[Materializing->Materializing] Scheduling Retry")
-        setTimer("retry", MyProtocol.PerformRetry, retryDuration)
+        startTimerAtFixedRate("retry", MyProtocol.PerformRetry, retryDuration)
       case MaterializingData.ToBeMaterialized(WorkerToEtlAssociation(worker, data)) =>
         log.info("[Materializing->Materializing] materialize [{}->{}]", worker, data.name)
         self ! MyProtocol.MaterializeETL(worker, data)
@@ -425,13 +425,13 @@ class PipegraphGuardian(private val master: ActorRef,
     case (Monitoring, Monitored) => nextStateData match {
       case MonitoredData.ShouldRetry() =>
         log.info("[Monitoring->Monitored] Scheduling Retry")
-        setTimer("retry", MyProtocol.PerformRetry, retryDuration)
+        startTimerAtFixedRate("retry", MyProtocol.PerformRetry, retryDuration)
       case MonitoredData.ShouldStopAll() =>
         log.info("[Monitoring->Monitored] StopAll")
         self ! MyProtocol.CancelWork
       case MonitoredData.ShouldMonitorAgain() =>
         log.info("[Monitoring->Monitored] Scheduling Monitoring")
-        setTimer("monitoring", MyProtocol.MonitorPipegraph, monitoringInterval)
+        startTimerAtFixedRate("monitoring", MyProtocol.MonitorPipegraph, monitoringInterval)
 
       case MonitoredData.NothingToMonitor() =>
         log.info("[Monitoring->Monitored] Nothing to monitor")
@@ -453,8 +453,9 @@ class PipegraphGuardian(private val master: ActorRef,
       case ActivatingData.ToBeActivated(etl) =>
         log.info("[Monitored->Activating] Scheduling activation")
         self ! MyProtocol.ActivateETL(etl)
-
-
+      case ActivatingData.AllActive() =>
+        log.info("[Monitored->Activating] AllActive")
+        self ! MyProtocol.ActivationFinished
     }
 
     case (Monitored, Monitoring) => nextStateData match {

consumers-spark/src/main/scala/it/agilelab/bigdata/wasp/consumers/spark/streaming/actor/telemetry/TelemetryActor.scala

@@ -168,7 +168,7 @@ class TelemetryActor private() extends Actor with CompatibilityTelemetryActor{
 
     implicit val ec: ExecutionContextExecutor = context.system.dispatcher
 
-    val _ = context.system.scheduler.schedule(
+    val _ = context.system.scheduler.scheduleAtFixedRate(
       FiniteDuration(5, TimeUnit.SECONDS),
       FiniteDuration(5, TimeUnit.SECONDS),
       mediator,

consumers-spark/src/main/scala/it/agilelab/bigdata/wasp/consumers/spark/streaming/actor/watchdog/SparkContextWatchDog.scala

@@ -14,8 +14,8 @@ class SparkContextWatchDog private(sc: SparkContext, failureAction: () => Unit)
   import scala.concurrent.duration._
 
   implicit val ec = context.system.dispatcher
-  context.system.scheduler.schedule(0.seconds, 1.second, self, MonitorSparkContext)
-  context.system.scheduler.schedule(0.seconds, 1.second, self, MonitorHdfTokens)
+  context.system.scheduler.scheduleAtFixedRate(0.seconds, 1.second, self, MonitorSparkContext)
+  context.system.scheduler.scheduleAtFixedRate(0.seconds, 1.second, self, MonitorHdfTokens)
 
 
   def waitForSparkContextToBeAvailable : Receive  = {

consumers-spark/src/test/resources/application.conf

@@ -8,11 +8,14 @@ akka {
 
 
   remote {
-    log-remote-lifecycle-events = off
-    enabled-transports = ["akka.remote.netty.tcp"]
-    netty.tcp {
-      port = 2892
-      hostname = "localhost"
+    artery.enabled = false
+    classic {
+      log-remote-lifecycle-events = off
+      enabled-transports = ["akka.remote.classic.netty.tcp"]
+      netty.tcp {
+        port = 2892
+        hostname = "localhost"
+      }
     }
   }
 
@@ -32,7 +35,7 @@ akka {
   }
 }
 
-system-0 {
+default_multimaster {
   akka {
     loglevel = "DEBUG"
     loggers = ["akka.event.slf4j.Slf4jLogger"]
@@ -43,11 +46,13 @@ system-0 {
 
 
     remote {
-      log-remote-lifecycle-events = off
-      enabled-transports = ["akka.remote.netty.tcp"]
-      netty.tcp {
-        port = 2893
-        hostname = "localhost"
+      artery.enabled = false
+      classic {
+        log-remote-lifecycle-events = off
+        enabled-transports = ["akka.remote.classic.netty.tcp"]
+        netty.tcp {
+          hostname = "localhost"
+        }
       }
     }
 
@@ -63,116 +68,21 @@ system-0 {
       publish-stats-interval = 10s
       metrics.gossip-interval = 10s
       metrics.collect-interval = 10s
-      roles = ["consumers-spark-streaming"]
     }
   }
 }
 
-
-system-1 {
-  akka {
-    loglevel = "DEBUG"
-    loggers = ["akka.event.slf4j.Slf4jLogger"]
-    logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
-    logger-startup-timeout = 60s
-    log-dead-letters = off
-    log-dead-letters-during-shutdown = off
-
-
-    remote {
-      log-remote-lifecycle-events = off
-      enabled-transports = ["akka.remote.netty.tcp"]
-      netty.tcp {
-        port = 2894
-        hostname = "localhost"
-      }
-    }
-
-    actor {
-      provider = "akka.cluster.ClusterActorRefProvider"
-      debug.fsm = true
-    }
-
-    cluster {
-      log-info = on
-      seed-nodes = ["akka.tcp://WASP@localhost:2893"]
-      gossip-interval = 5s
-      publish-stats-interval = 10s
-      metrics.gossip-interval = 10s
-      metrics.collect-interval = 10s
-      roles = ["consumers-spark-streaming"]
-    }
-  }
+system-0.akka {
+  remote.classic.netty.tcp.port = 2893
+  cluster.roles=["consumers-spark-streaming"]
 }
+system-1.akka {
+  remote.classic.netty.tcp.port = 2894
+  cluster.roles=["consumers-spark-streaming"]
 
-
-system-2 {
-  akka {
-    loglevel = "DEBUG"
-    loggers = ["akka.event.slf4j.Slf4jLogger"]
-    logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
-    logger-startup-timeout = 60s
-    log-dead-letters = off
-    log-dead-letters-during-shutdown = off
-
-
-    remote {
-      log-remote-lifecycle-events = off
-      enabled-transports = ["akka.remote.netty.tcp"]
-      netty.tcp {
-        port = 2895
-        hostname = "localhost"
-      }
-    }
-
-    actor {
-      provider = "akka.cluster.ClusterActorRefProvider"
-      debug.fsm = true
-    }
-
-    cluster {
-      log-info = on
-      seed-nodes = ["akka.tcp://WASP@localhost:2893"]
-      gossip-interval = 5s
-      publish-stats-interval = 10s
-      metrics.gossip-interval = 10s
-      metrics.collect-interval = 10s
-      roles = ["consumers-spark-streaming"]
-    }
-  }
 }
-
-coordinator {
-  akka {
-    loglevel = "DEBUG"
-    loggers = ["akka.event.slf4j.Slf4jLogger"]
-    logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
-    logger-startup-timeout = 60s
-    log-dead-letters = off
-    log-dead-letters-during-shutdown = off
-
-
-    remote {
-      log-remote-lifecycle-events = off
-      enabled-transports = ["akka.remote.netty.tcp"]
-      netty.tcp {
-        port = 2896
-        hostname = "localhost"
-      }
-    }
-
-    actor {
-      provider = "akka.cluster.ClusterActorRefProvider"
-      debug.fsm = true
-    }
-
-    cluster {
-      log-info = on
-      seed-nodes = ["akka.tcp://WASP@localhost:2893"]
-      gossip-interval = 5s
-      publish-stats-interval = 10s
-      metrics.gossip-interval = 10s
-      metrics.collect-interval = 10s
-    }
-  }
+system-2.akka {
+  remote.classic.netty.tcp.port = 2895
+  cluster.roles=["consumers-spark-streaming"]
 }
+coordinator.akka.remote.classic.netty.tcp.port = 2896

consumers-spark/src/test/resources/log4j.xml

@@ -2,11 +2,11 @@
 <Configuration status="WARN">
     <Appenders>
         <Console name="Console" target="SYSTEM_OUT">
-            <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
+            <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level  %c{1}.%M - %msg%n"/>
         </Console>
     </Appenders>
     <Loggers>
-        <Root level="ERROR">
+        <Root level="WARN">
             <AppenderRef ref="Console"/>
         </Root>
     </Loggers>