Exclude warmed containers in disabled invokers. (apache#5313)

* Exclude warmed containers in disabled invokers.

* Exclude warmed containers in disabled invokers.

* Find the first warmed container.

* Remove the code added by mistake.

* Add more logs for error cases.

Author: style95

common/scala/src/main/scala/org/apache/openwhisk/core/connector/Message.scala

@@ -636,6 +636,23 @@ object ContainerMessage extends DefaultJsonProtocol {
 sealed trait ContainerCreationError
 
 object ContainerCreationError extends Enumeration {
+  import scala.language.implicitConversions
+  implicit def containerCreationErrorToString(x: ContainerCreationError): String = {
+    x match {
+      case NoAvailableInvokersError         => "no available invoker is found"
+      case NoAvailableResourceInvokersError => "no available invoker with the resources is found: "
+      case ResourceNotEnoughError           => "invoker(s) have not enough resources"
+      case WhiskError                       => "whisk error(recoverable) happens"
+      case UnknownError                     => "a unknown error happens"
+      case TimeoutError                     => "a timeout error happens"
+      case ShuttingDownError                => "shutting down error happens"
+      case NonExecutableActionError         => "no executable found for the action"
+      case DBFetchError                     => "an error happens while fetching data from DB"
+      case BlackBoxError                    => "a blackbox error happens"
+      case ZeroNamespaceLimit               => "the namespace has 0 limit configured"
+      case TooManyConcurrentRequests        => "too many concurrent requests are in flight."
+    }
+  }
 
   case object NoAvailableInvokersError extends ContainerCreationError
 

core/invoker/src/main/scala/org/apache/openwhisk/core/invoker/FPCInvokerReactive.scala

@@ -376,34 +376,20 @@ class FPCInvokerReactive(config: WhiskConfig,
   override def enable(): Route = {
     invokerHealthManager ! Enable
     pool ! Enable
-    // re-enable consumer
-    if (consumer.isEmpty)
-      consumer = Some(
-        new ContainerMessageConsumer(
-          instance,
-          pool,
-          entityStore,
-          cfg,
-          msgProvider,
-          longPollDuration = 1.second,
-          maxPeek,
-          sendAckToScheduler))
     warmUp()
     complete("Success enable invoker")
   }
 
   override def disable(): Route = {
     invokerHealthManager ! GracefulShutdown
     pool ! GracefulShutdown
-    consumer.foreach(_.close())
-    consumer = None
     warmUpWatcher.foreach(_.close())
     warmUpWatcher = None
     complete("Successfully disabled invoker")
   }
 
   override def isEnabled(): Route = {
-    complete(InvokerEnabled(consumer.nonEmpty && warmUpWatcher.nonEmpty).serialize())
+    complete(InvokerEnabled(warmUpWatcher.nonEmpty).serialize())
   }
 
   override def backfillPrewarm(): Route = {