diff --git a/common/scala/src/main/scala/org/apache/openwhisk/common/NestedSemaphore.scala b/common/scala/src/main/scala/org/apache/openwhisk/common/NestedSemaphore.scala
index 4a6942b33d9..1e7bee5512a 100644
--- a/common/scala/src/main/scala/org/apache/openwhisk/common/NestedSemaphore.scala
+++ b/common/scala/src/main/scala/org/apache/openwhisk/common/NestedSemaphore.scala
@@ -100,14 +100,17 @@ 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...
+      //(this case can occur if one or more cluster members is restarted, resetting the state of the outstanding activations)
+      actionConcurrentSlotsMap.get(actionid).foreach { concurrentSlots =>
+        val (memoryRelease, actionRelease) = concurrentSlots.release(1, true)
+        //concurrent slots
+        if (memoryRelease) {
+          super.release(memoryPermits)
+        }
+        if (actionRelease) {
+          actionConcurrentSlotsMap.remove(actionid)
+        }
       }
     }
   }
diff --git a/common/scala/src/main/scala/org/apache/openwhisk/core/connector/MessageConsumer.scala b/common/scala/src/main/scala/org/apache/openwhisk/core/connector/MessageConsumer.scala
index 8713b4afde5..f85fcc8589e 100644
--- a/common/scala/src/main/scala/org/apache/openwhisk/core/connector/MessageConsumer.scala
+++ b/common/scala/src/main/scala/org/apache/openwhisk/core/connector/MessageConsumer.scala
@@ -26,6 +26,7 @@ import scala.util.Failure
 import org.apache.kafka.clients.consumer.CommitFailedException
 import akka.actor.FSM
 import akka.pattern.pipe
+import org.apache.commons.lang3.exception.ExceptionUtils
 import org.apache.openwhisk.common.Logging
 import org.apache.openwhisk.common.TransactionId
 
@@ -213,7 +214,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) =>
+            val stacktrace = ExceptionUtils.getStackTrace(e)
+            logging.error(this, s"Failed to process message for topic $topic : $e : stacktrace: $stacktrace")
+        }
+      }
       handlerCapacity -= 1
 
       sendOutstandingMessages()