Take revision into consideration when choose warm container

core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/container/ContainerManager.scala

@@ -251,7 +251,7 @@ class ContainerManager(jobManagerFactory: ActorRefFactory => ActorRef,
   // Filter out messages which can use warmed container
   private def filterWarmedCreations(msgs: List[ContainerCreationMessage]) = {
     msgs.filter { msg =>
-      val warmedPrefix = containerPrefix(ContainerKeys.warmedPrefix, msg.invocationNamespace, msg.action)
+      val warmedPrefix = containerPrefix(ContainerKeys.warmedPrefix, msg.invocationNamespace, msg.action, Some(msg.revision))
       val chosenInvoker = warmedContainers
         .filter(!inProgressWarmedContainers.values.toSeq.contains(_))
         .find { container =>

tests/src/test/scala/org/apache/openwhisk/core/scheduler/container/test/ContainerManagerTests.scala

@@ -278,7 +278,7 @@ class ContainerManagerTests
     )
     expectGetInvokers(mockEtcd, invokers)
     expectGetInvokers(mockEtcd, invokers)
-    expectGetInvokers(mockEtcd, invokers) // this test case will run `getPrefix` twice
+    expectGetInvokers(mockEtcd, invokers) // this test case will run `getPrefix` twice, and another one for warmup
 
     val mockJobManager = TestProbe()
     val mockWatcher = TestProbe()
@@ -377,6 +377,65 @@ class ContainerManagerTests
     receiver.expectMsg(s"invoker0-$msg1")
   }
 
+  it should "not try warmed containers if revision is unmatched" in {
+    val mockEtcd = mock[EtcdClient]
+
+    // for test, only invoker2 is healthy, so that no-warmed creations can be only sent to invoker2
+    val invokers: List[InvokerHealth] = List(
+      InvokerHealth(InvokerInstanceId(0, userMemory = testMemory, tags = Seq.empty[String]), Unhealthy),
+      InvokerHealth(InvokerInstanceId(1, userMemory = testMemory, tags = Seq.empty[String]), Unhealthy),
+      InvokerHealth(InvokerInstanceId(2, userMemory = testMemory, tags = Seq.empty[String]), Healthy),
+    )
+    expectGetInvokers(mockEtcd, invokers)
+    expectGetInvokers(mockEtcd, invokers) // one for warmup
+
+    val mockJobManager = TestProbe()
+    val mockWatcher = TestProbe()
+    val receiver = TestProbe()
+
+    val manager =
+      system.actorOf(ContainerManager
+        .props(factory(mockJobManager), mockMessaging(Some(receiver.ref)), testsid, mockEtcd, config, mockWatcher.ref))
+
+    // there are 1 warmed container for `test-namespace/test-action` but with a different revision
+    manager ! WatchEndpointInserted(
+      ContainerKeys.warmedPrefix,
+      ContainerKeys.warmedContainers(
+        testInvocationNamespace,
+        testfqn,
+        DocRevision("2-testRev"),
+        InvokerInstanceId(0, userMemory = 0.bytes),
+        ContainerId("fake")),
+      "",
+      true)
+
+    val msg =
+      ContainerCreationMessage(
+        TransactionId.testing,
+        testInvocationNamespace,
+        testfqn,
+        testRevision,
+        actionMetadata,
+        testsid,
+        schedulerHost,
+        rpcPort)
+
+    // it should not reuse the warmed container
+    manager ! ContainerCreation(List(msg), 128.MB, testInvocationNamespace)
+
+    // ignore warmUp message
+    receiver.ignoreMsg {
+      case s: String => s.contains("warmUp")
+    }
+
+    // it should be scheduled to the sole health invoker: invoker2
+    receiver.expectMsg(s"invoker2-$msg")
+
+    mockJobManager.expectMsgPF() {
+      case RegisterCreationJob(`msg`) => true
+    }
+  }
+
   it should "rescheduling container creation" in {
     val mockEtcd = mock[EtcdClient]
     (mockEtcd