Skip to content

Send a queue removed message to the queue manager #5391

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

Merged
merged 3 commits into from
Apr 4, 2023
Merged

Send a queue removed message to the queue manager #5391

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ced1043
Send a queue removed message to the queue manager when cleaning up th…
style95 Mar 30, 2023
31f0996
Add one more missing part
style95 Mar 31, 2023
3e99442
Fix test cases
style95 Mar 31, 2023
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
4 changes: 4 additions & 0 deletions core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/queue/MemoryQueue.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -694,13 +694,17 @@ class MemoryQueue(private val etcdClient: EtcdClient,
cleanUpWatcher()
cleanUpData()

context.parent ! queueRemovedMsg

goto(Removed) using NoData()
}

private def cleanUpActorsAndGotoRemoved(data: FlushingData) = {
cleanUpActors(data)
cleanUpData()

context.parent ! queueRemovedMsg

goto(Removed) using NoData()
}

Expand Down
13 changes: 4 additions & 9 deletions tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/MemoryQueueTests.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -654,13 +654,14 @@ class MemoryQueueTests
fsm ! StateTimeout
expectMsg(Transition(fsm, Idle, Removed))
queueRef.queue.length shouldBe 0
parent.expectMsg(queueRemovedMsg)

fsm ! message

// queue is timed out again in the Removed state.
parent.expectMsg(message)

fsm ! StateTimeout
parent.expectMsg(queueRemovedMsg)

expectNoMessage()

Expand Down Expand Up @@ -1103,10 +1104,7 @@ class MemoryQueueTests
fsm ! message
probe.expectMsg(ActivationResponse.developerError("nonExecutbleAction error"))

parent.expectMsgAnyOf(
2 * queueConfig.flushGrace + 5.seconds,
QueueRemoved(testInvocationNamespace, fqn.toDocId.asDocInfo(action.rev), Some(leaderKey)),
Transition(fsm, Flushing, Removed))
parent.expectMsgAllOf(2 * queueConfig.flushGrace + 5.seconds, queueRemovedMsg, Transition(fsm, Flushing, Removed))

fsm ! StateTimeout
parent.expectMsg(queueRemovedMsg)
Expand Down Expand Up @@ -1409,10 +1407,7 @@ class MemoryQueueTests

val duration = FiniteDuration(queueConfig.maxBlackboxRetentionMs, MILLISECONDS) + queueConfig.flushGrace
probe.expectMsg(duration, ActivationResponse.whiskError("no available invokers"))
parent.expectMsgAnyOf(
duration,
QueueRemoved(testInvocationNamespace, fqn.toDocId.asDocInfo(action.rev), Some(leaderKey)),
Transition(fsm, Flushing, Removed))
parent.expectMsgAllOf(duration, queueRemovedMsg, Transition(fsm, Flushing, Removed))
fsm ! QueueRemovedCompleted
parent.expectTerminated(fsm)

Expand Down