Go to the NamespaceThrottled state rather than Flushing state. (#5303)

* Currently MemoryQueue will go to Flushing state when receive a EnableNamespaceThrottling(dropMsg=true) message, but the Flushing state doesn't have a case to disable namespace throttling at all.

* Remove unused import.

core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/queue/MemoryQueue.scala

@@ -17,15 +17,13 @@
 
 package org.apache.openwhisk.core.scheduler.queue
 
-import java.time.{Duration, Instant}
-import java.util.concurrent.atomic.AtomicInteger
 import akka.actor.Status.{Failure => FailureMessage}
 import akka.actor.{ActorRef, ActorSystem, Cancellable, FSM, Props, Stash}
 import akka.util.Timeout
 import org.apache.openwhisk.common._
 import org.apache.openwhisk.core.ConfigKeys
 import org.apache.openwhisk.core.ack.ActiveAck
-import org.apache.openwhisk.core.connector.ContainerCreationError.{TooManyConcurrentRequests, ZeroNamespaceLimit}
+import org.apache.openwhisk.core.connector.ContainerCreationError.ZeroNamespaceLimit
 import org.apache.openwhisk.core.connector._
 import org.apache.openwhisk.core.containerpool.Interval
 import org.apache.openwhisk.core.database.{NoDocumentException, UserContext}
@@ -44,10 +42,12 @@ import org.apache.openwhisk.core.scheduler.message.{
 import org.apache.openwhisk.core.scheduler.{SchedulerEndpoints, SchedulingConfig}
 import org.apache.openwhisk.core.service._
 import org.apache.openwhisk.http.Messages.{namespaceLimitUnderZero, tooManyConcurrentRequests}
-import pureconfig.generic.auto._
 import pureconfig.loadConfigOrThrow
 import spray.json._
+import pureconfig.generic.auto._
 
+import java.time.{Duration, Instant}
+import java.util.concurrent.atomic.AtomicInteger
 import scala.annotation.tailrec
 import scala.collection.immutable.Queue
 import scala.collection.mutable
@@ -224,18 +224,9 @@ class MemoryQueue(private val etcdClient: EtcdClient,
  logging.info(this, s"[$invocationNamespace:$action:$stateName] Enable namespace throttling.")
  enableNamespaceThrottling()
 
- // if no container could be created, it is same with Flushing state.
- if (dropMsg) {
+ if (dropMsg)
  completeAllActivations(tooManyConcurrentRequests, isWhiskError = false)
- goto(Flushing) using FlushingData(
- data.schedulerActor,
- data.droppingActor,
- TooManyConcurrentRequests,
- tooManyConcurrentRequests)
- } else {
- // if there are already some containers running, activations can still be processed so goto the NamespaceThrottled state.
- goto(NamespaceThrottled) using ThrottledData(data.schedulerActor, data.droppingActor)
- }
+ goto(NamespaceThrottled) using ThrottledData(data.schedulerActor, data.droppingActor)
 
  case Event(StateTimeout, data: RunningData) =>
  if (queue.isEmpty && (containers.size + creationIds.size) <= 0) {

tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/MemoryQueueFlowTests.scala

@@ -280,7 +280,7 @@ class MemoryQueueFlowTests
  probe.expectTerminated(fsm, 10.seconds)
  }
 
- it should "go to the Flushing state dropping messages when it can't create an initial container" in {
+ it should "go to the NamespaceThrottled state dropping messages when it can't create an initial container" in {
  val mockEtcdClient = mock[EtcdClient]
  val parent = TestProbe()
  val watcher = TestProbe()
@@ -340,7 +340,7 @@ class MemoryQueueFlowTests
  fsm ! message
 
  dataMgmtService.expectMsg(RegisterData(namespaceThrottlingKey, true.toString, failoverEnabled = false))
- probe.expectMsg(Transition(fsm, Running, Flushing))
+ probe.expectMsg(Transition(fsm, Running, NamespaceThrottled))
 
  awaitAssert({
  ackedMessageCount shouldBe 1
@@ -352,13 +352,17 @@ class MemoryQueueFlowTests
  fsm.underlyingActor.queue.size shouldBe 0
  }, 5.seconds)
 
- parent.expectMsg(flushGrace * 2 + 5.seconds, queueRemovedMsg)
- probe.expectMsg(Transition(fsm, Flushing, Removed))
+ fsm ! GracefulShutdown
+
+ parent.expectMsg(queueRemovedMsg)
+ probe.expectMsg(Transition(fsm, NamespaceThrottled, Removing))
 
  fsm ! QueueRemovedCompleted
 
  expectDataCleanUp(watcher, dataMgmtService)
 
+ probe.expectMsg(Transition(fsm, Removing, Removed))
+
  probe.expectTerminated(fsm, 10.seconds)
  }
 

tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/MemoryQueueTests.scala

@@ -1357,7 +1357,7 @@ class MemoryQueueTests
  parent.expectMsg(10 seconds, Transition(fsm, Uninitialized, Running))
 
  fsm ! EnableNamespaceThrottling(dropMsg = true)
- parent.expectMsg(10 seconds, Transition(fsm, Running, Flushing))
+ parent.expectMsg(10 seconds, Transition(fsm, Running, NamespaceThrottled))
  dataManagementService.expectMsg(RegisterData(namespaceThrottlingKey, true.toString, false))
 
  fsm.stop()