Fix build error (#5123)

Author: ningyougang

core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/v2/ActivationClientProxy.scala

@@ -21,7 +21,6 @@ import akka.actor.Status.{Failure => FailureMessage}
 import akka.actor.{ActorRef, ActorSystem, FSM, Props, Stash}
 import akka.grpc.internal.ClientClosedException
 import akka.pattern.pipe
-import akka.stream.ActorMaterializer
 import io.grpc.StatusRuntimeException
 import org.apache.openwhisk.common.{Logging, TransactionId}
 import org.apache.openwhisk.core.connector.ActivationMessage
@@ -73,7 +72,6 @@ class ActivationClientProxy(
   containerId: ContainerId,
   activationClientFactory: (String, FullyQualifiedEntityName, String, Int, Boolean) => Future[ActivationServiceClient])(
   implicit actorSystem: ActorSystem,
-  mat: ActorMaterializer,
   logging: Logging)
     extends FSM[ActivationClientProxyState, ActivationClientProxyData]
     with Stash {
@@ -406,13 +404,12 @@ object ActivationClientProxy {
             schedulerHost: String,
             rpcPort: Int,
             containerId: ContainerId,
-            activationClientFactory: (String,
-                                      FullyQualifiedEntityName,
-                                      String,
-                                      Int,
-                                      Boolean) => Future[ActivationServiceClient])(implicit actorSystem: ActorSystem,
-                                                                                   mat: ActorMaterializer,
-                                                                                   logging: Logging) = {
+            activationClientFactory: (
+              String,
+              FullyQualifiedEntityName,
+              String,
+              Int,
+              Boolean) => Future[ActivationServiceClient])(implicit actorSystem: ActorSystem, logging: Logging) = {
     Props(
       new ActivationClientProxy(
         invocationNamespace,

core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/v2/FunctionPullingContainerProxy.scala

@@ -25,7 +25,6 @@ import akka.actor.{ActorRef, ActorRefFactory, ActorSystem, FSM, Props, Stash}
 import akka.event.Logging.InfoLevel
 import akka.io.{IO, Tcp}
 import akka.pattern.pipe
-import akka.stream.ActorMaterializer
 import org.apache.openwhisk.common.tracing.WhiskTracerProvider
 import org.apache.openwhisk.common.{LoggingMarkers, TransactionId, _}
 import org.apache.openwhisk.core.ConfigKeys
@@ -194,7 +193,7 @@ class FunctionPullingContainerProxy(
   poolConfig: ContainerPoolConfig,
   timeoutConfig: ContainerProxyTimeoutConfig,
   healtCheckConfig: ContainerProxyHealthCheckConfig,
-  testTcp: Option[ActorRef])(implicit actorSystem: ActorSystem, mat: ActorMaterializer, logging: Logging)
+  testTcp: Option[ActorRef])(implicit actorSystem: ActorSystem, logging: Logging)
     extends FSM[ProxyState, Data]
     with Stash {
   startWith(Uninitialized, NonexistentData())
@@ -373,7 +372,7 @@ class FunctionPullingContainerProxy(
     case Event(initializedData: InitializedData, _) =>
       context.parent ! Initialized(initializedData)
       initializedData.clientProxy ! RequestActivation()
-      setTimer(UnusedTimeoutName, StateTimeout, unusedTimeout)
+      startSingleTimer(UnusedTimeoutName, StateTimeout, unusedTimeout)
       stay() using initializedData
 
     // 2. read executable action data from db
@@ -506,7 +505,7 @@ class FunctionPullingContainerProxy(
     // 1. request activation message to client
     case Event(activationResult: RunActivationCompleted, data: WarmData) =>
       // create timeout
-      setTimer(UnusedTimeoutName, StateTimeout, unusedTimeout)
+      startSingleTimer(UnusedTimeoutName, StateTimeout, unusedTimeout)
       data.clientProxy ! RequestActivation(activationResult.duration)
       stay() using data
 
@@ -613,7 +612,7 @@ class FunctionPullingContainerProxy(
         logging.info(
           this,
           s"Remain running activations ${runningActivations.keySet().toString()} when received ClientClosed")
-        setTimer(RunningActivationTimeoutName, ClientClosed, runningActivationTimeout)
+        startSingleTimer(RunningActivationTimeoutName, ClientClosed, runningActivationTimeout)
         stay
       }
 
@@ -674,7 +673,7 @@ class FunctionPullingContainerProxy(
           // since akka port will no be used, we can put any value except 0 here
           data.clientProxy ! RequestActivation(
             newScheduler = Some(SchedulerEndpoints(job.schedulerHost, job.rpcPort, 10)))
-          setTimer(UnusedTimeoutName, StateTimeout, unusedTimeout)
+          startSingleTimer(UnusedTimeoutName, StateTimeout, unusedTimeout)
           timedOut = false
         }
         .recover {
@@ -715,7 +714,7 @@ class FunctionPullingContainerProxy(
       logging.info(
         this,
         s"This is the remaining container for ${data.action}. The container will stop after $warmedContainerKeepingTimeout.")
-      setTimer(KeepingTimeoutName, Remove, warmedContainerKeepingTimeout)
+      startSingleTimer(KeepingTimeoutName, Remove, warmedContainerKeepingTimeout)
       stay
 
     case Event(Remove | GracefulShutdown, data: WarmData) =>
@@ -777,7 +776,7 @@ class FunctionPullingContainerProxy(
         }
       }
       unstashAll()
-    case _ -> Paused   => setTimer(IdleTimeoutName, StateTimeout, idleTimeout)
+    case _ -> Paused   => startSingleTimer(IdleTimeoutName, StateTimeout, idleTimeout)
     case _ -> Removing => unstashAll()
   }
 
@@ -1245,7 +1244,7 @@ object FunctionPullingContainerProxy {
             timeoutConfig: ContainerProxyTimeoutConfig,
             healthCheckConfig: ContainerProxyHealthCheckConfig =
               loadConfigOrThrow[ContainerProxyHealthCheckConfig](ConfigKeys.containerProxyHealth),
-            tcp: Option[ActorRef] = None)(implicit actorSystem: ActorSystem, mat: ActorMaterializer, logging: Logging) =
+            tcp: Option[ActorRef] = None)(implicit actorSystem: ActorSystem, logging: Logging) =
     Props(
       new FunctionPullingContainerProxy(
         factory,

tests/src/test/scala/org/apache/openwhisk/core/containerpool/v2/test/ActivationClientProxyTests.scala

@@ -21,7 +21,6 @@ import akka.Done
 import akka.actor.FSM.{CurrentState, SubscribeTransitionCallBack, Transition}
 import akka.actor.{ActorRef, ActorSystem}
 import akka.grpc.internal.ClientClosedException
-import akka.stream.ActorMaterializer
 import akka.testkit.{ImplicitSender, TestKit, TestProbe}
 import common.StreamLogging
 import io.grpc.StatusRuntimeException
@@ -59,7 +58,6 @@ class ActivationClientProxyTests
 
   override def afterAll: Unit = TestKit.shutdownActorSystem(system)
 
-  implicit val mat = ActorMaterializer()
   implicit val ec = system.dispatcher
 
   val timeout = 20.seconds

tests/src/test/scala/org/apache/openwhisk/core/containerpool/v2/test/FunctionPullingContainerProxyTests.scala

@@ -26,7 +26,6 @@ import akka.actor.FSM.{CurrentState, StateTimeout, SubscribeTransitionCallBack,
 import akka.actor.{Actor, ActorRef, ActorRefFactory, ActorSystem, Props}
 import akka.http.scaladsl.model
 import akka.io.Tcp.Connect
-import akka.stream.ActorMaterializer
 import akka.stream.scaladsl.{Sink, Source}
 import akka.testkit.{ImplicitSender, TestKit, TestProbe}
 import akka.util.ByteString
@@ -86,7 +85,6 @@ class FunctionPullingContainerProxyTests
     super.afterAll()
   }
 
-  implicit val mat = ActorMaterializer()
   implicit val ece: ExecutionContextExecutor = system.dispatcher
 
   val timeout = 20.seconds