add defensive code for dealing with concurrency tracking when cluster events cause maps to reset

Author: tysonnorris

common/scala/src/main/scala/org/apache/openwhisk/common/NestedSemaphore.scala

@@ -100,14 +100,19 @@ class NestedSemaphore[T](memoryPermits: Int) extends ForcibleSemaphore(memoryPer
     if (maxConcurrent == 1) {
       super.release(memoryPermits)
     } else {
-      val concurrentSlots = actionConcurrentSlotsMap(actionid)
-      val (memoryRelease, actionRelease) = concurrentSlots.release(1, true)
-      //concurrent slots
-      if (memoryRelease) {
-        super.release(memoryPermits)
-      }
-      if (actionRelease) {
-        actionConcurrentSlotsMap.remove(actionid)
+      //This map may be recreated (multiple times) due to cluster membership events, so don't assume the entry exists...
+      actionConcurrentSlotsMap.get(actionid) match {
+        case Some(concurrentSlots) =>
+          val (memoryRelease, actionRelease) = concurrentSlots.release(1, true)
+          //concurrent slots
+          if (memoryRelease) {
+            super.release(memoryPermits)
+          }
+          if (actionRelease) {
+            actionConcurrentSlotsMap.remove(actionid)
+          }
+        case None =>
+        //this case can occur if one or more cluster members is restarted, resetting the state of the outstanding activations
       }
     }
   }

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

@@ -213,7 +213,13 @@ class MessageFeed(description: String,
       outstandingMessages = outstandingMessages.tail
 
       if (logHandoff) logging.debug(this, s"processing $topic[$partition][$offset] ($occupancy/$handlerCapacity)")
-      handler(bytes)
+      handler(bytes).andThen {
+        {
+          case Failure(e) =>
+            logging.error(this, s"Failed to process message for topic $topic : $e  (stack trace included)")
+            e.printStackTrace()
+        }
+      }
       handlerCapacity -= 1
 
       sendOutstandingMessages()