You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2017/12/05 14:32:23 UTC
flink git commit: [FLINK-8204] [tests] Harden
JobManagerLeaderSessionIDITCase
Repository: flink
Updated Branches:
refs/heads/master f93dc8aa0 -> dd48a40ab
[FLINK-8204] [tests] Harden JobManagerLeaderSessionIDITCase
Replace wait on lock with OneShotLatch to resolve race condition between task
start up and unblocking of invokable.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/dd48a40a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/dd48a40a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/dd48a40a
Branch: refs/heads/master
Commit: dd48a40ab7b3a721c7c78a4b0d6efad5b50f947e
Parents: f93dc8a
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue Dec 5 15:27:04 2017 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Tue Dec 5 15:31:41 2017 +0100
----------------------------------------------------------------------
.../JobManagerLeaderSessionIDITCase.scala | 32 +++++++++-----------
1 file changed, 14 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/dd48a40a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerLeaderSessionIDITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerLeaderSessionIDITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerLeaderSessionIDITCase.scala
index b57c0da..85747d5 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerLeaderSessionIDITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerLeaderSessionIDITCase.scala
@@ -19,18 +19,19 @@
package org.apache.flink.api.scala.runtime.jobmanager
import java.util.UUID
+import java.util.concurrent.CountDownLatch
import akka.actor.ActorSystem
import akka.testkit.{ImplicitSender, TestKit}
+import org.apache.flink.core.testutils.OneShotLatch
import org.apache.flink.runtime.akka.{AkkaUtils, ListeningBehaviour}
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable
import org.apache.flink.runtime.jobgraph.{JobGraph, JobVertex}
import org.apache.flink.runtime.messages.JobManagerMessages._
-import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages.{AllVerticesRunning, WaitForAllVerticesToBeRunning}
import org.apache.flink.runtime.testingUtils.{ScalaTestingUtils, TestingUtils}
import org.junit.runner.RunWith
-import org.scalatest.{BeforeAndAfterAll, FunSuiteLike, Matchers}
import org.scalatest.junit.JUnitRunner
+import org.scalatest.{BeforeAndAfterAll, FunSuiteLike, Matchers}
@RunWith(classOf[JUnitRunner])
class JobManagerLeaderSessionIDITCase(_system: ActorSystem)
@@ -41,9 +42,7 @@ class JobManagerLeaderSessionIDITCase(_system: ActorSystem)
with BeforeAndAfterAll
with ScalaTestingUtils {
- val numTaskManagers = 2
- val taskManagerNumSlots = 2
- val numSlots = numTaskManagers * taskManagerNumSlots
+ import BlockingUntilSignalNoOpInvokable._
val cluster = TestingUtils.startTestingCluster(
taskManagerNumSlots,
@@ -73,13 +72,11 @@ class JobManagerLeaderSessionIDITCase(_system: ActorSystem)
expectMsg(JobSubmitSuccess(jobGraph.getJobID))
- jmGateway.tell(WaitForAllVerticesToBeRunning(jobGraph.getJobID), self)
-
- expectMsg(AllVerticesRunning(jobGraph.getJobID))
+ BlockingUntilSignalNoOpInvokable.countDownLatch.await()
jm ! LeaderSessionMessage(oldSessionID, CancelJob(jobGraph.getJobID))
- BlockingUntilSignalNoOpInvokable.triggerExecution()
+ BlockingUntilSignalNoOpInvokable.oneShotLatch.trigger()
expectMsgType[JobResultSuccess]
}
@@ -89,18 +86,17 @@ class JobManagerLeaderSessionIDITCase(_system: ActorSystem)
class BlockingUntilSignalNoOpInvokable extends AbstractInvokable {
override def invoke(): Unit = {
- BlockingUntilSignalNoOpInvokable.lock.synchronized{
- BlockingUntilSignalNoOpInvokable.lock.wait()
- }
+ BlockingUntilSignalNoOpInvokable.countDownLatch.countDown()
+ BlockingUntilSignalNoOpInvokable.oneShotLatch.await()
}
}
object BlockingUntilSignalNoOpInvokable {
- val lock = new Object
+ val numTaskManagers = 2
+ val taskManagerNumSlots = 2
+ val numSlots = numTaskManagers * taskManagerNumSlots
- def triggerExecution(): Unit = {
- lock.synchronized{
- lock.notifyAll()
- }
- }
+ val countDownLatch = new CountDownLatch(numSlots)
+
+ val oneShotLatch = new OneShotLatch()
}