Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

add defensive code for dealing with loadbalancer concurrency tracking #4951

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
Open

add defensive code for dealing with loadbalancer concurrency tracking #4951

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0addfbe
add defensive code for dealing with concurrency tracking when cluster…
tysonnorris Aug 25, 2020
91e87e7
embed stacktrace in log message
tysonnorris Sep 3, 2020
6235e31
Update common/scala/src/main/scala/org/apache/openwhisk/common/Nested…
tysonnorris Sep 8, 2020
61ac430
cleanup
tysonnorris Sep 8, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 11 additions & 8 deletions common/scala/src/main/scala/org/apache/openwhisk/common/NestedSemaphore.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
}
}
}
Expand Down
9 changes: 8 additions & 1 deletion common/scala/src/main/scala/org/apache/openwhisk/core/connector/MessageConsumer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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 {
{
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are the inner braces needed?

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()
Expand Down