flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ches...@apache.org
Subject flink git commit: [FLINK-7212][tests] re-enable JobManagerLeaderSessionIDITCase
Date Tue, 18 Jul 2017 14:11:56 GMT
Repository: flink
Updated Branches:
  refs/heads/master 67fb2f3ab -> 673a883ed


[FLINK-7212][tests] re-enable JobManagerLeaderSessionIDITCase

This test was previously named JobManagerLeaderSessionIDITSuite and has not
been executed for a while by maven because of it having the wrong naming
scheme. After a renaming, it runs again but needed a minor change to be
successful again which is included in this commit.

This closes #4354.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/673a883e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/673a883e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/673a883e

Branch: refs/heads/master
Commit: 673a883ed89098127c7b7a67216d4399885a7fce
Parents: 67fb2f3
Author: Nico Kruber <nico@data-artisans.com>
Authored: Mon Jul 17 14:46:17 2017 +0200
Committer: zentol <chesnay@apache.org>
Committed: Tue Jul 18 16:09:33 2017 +0200

----------------------------------------------------------------------
 .../JobManagerLeaderSessionIDITCase.scala       | 106 ++++++++++++++++++
 .../JobManagerLeaderSessionIDITSuite.scala      | 110 -------------------
 2 files changed, 106 insertions(+), 110 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/673a883e/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
new file mode 100644
index 0000000..b57c0da
--- /dev/null
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerLeaderSessionIDITCase.scala
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.scala.runtime.jobmanager
+
+import java.util.UUID
+
+import akka.actor.ActorSystem
+import akka.testkit.{ImplicitSender, TestKit}
+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
+
+@RunWith(classOf[JUnitRunner])
+class JobManagerLeaderSessionIDITCase(_system: ActorSystem)
+  extends TestKit(_system)
+  with ImplicitSender
+  with FunSuiteLike
+  with Matchers
+  with BeforeAndAfterAll
+  with ScalaTestingUtils {
+
+  val numTaskManagers = 2
+  val taskManagerNumSlots = 2
+  val numSlots = numTaskManagers * taskManagerNumSlots
+
+  val cluster = TestingUtils.startTestingCluster(
+    taskManagerNumSlots,
+    numTaskManagers,
+    TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT);
+
+  def this() = this(ActorSystem("TestingActorSystem", AkkaUtils.getDefaultAkkaConfig))
+
+  override def afterAll(): Unit = {
+    cluster.stop()
+    TestKit.shutdownActorSystem(system)
+  }
+
+  test("A JobManager should not process CancelJob messages with the wrong leader session
ID") {
+    val sender = new JobVertex("BlockingSender");
+    sender.setParallelism(numSlots)
+    sender.setInvokableClass(classOf[BlockingUntilSignalNoOpInvokable])
+    val jobGraph = new JobGraph("TestJob", sender)
+
+    val oldSessionID = UUID.randomUUID()
+
+    val jmGateway = cluster.getLeaderGateway(TestingUtils.TESTING_DURATION)
+    val jm = jmGateway.actor()
+
+    within(TestingUtils.TESTING_DURATION) {
+      jmGateway.tell(SubmitJob(jobGraph, ListeningBehaviour.EXECUTION_RESULT), self)
+
+      expectMsg(JobSubmitSuccess(jobGraph.getJobID))
+
+      jmGateway.tell(WaitForAllVerticesToBeRunning(jobGraph.getJobID), self)
+
+      expectMsg(AllVerticesRunning(jobGraph.getJobID))
+
+      jm ! LeaderSessionMessage(oldSessionID, CancelJob(jobGraph.getJobID))
+
+      BlockingUntilSignalNoOpInvokable.triggerExecution()
+
+      expectMsgType[JobResultSuccess]
+    }
+  }
+}
+
+class BlockingUntilSignalNoOpInvokable extends AbstractInvokable {
+
+  override def invoke(): Unit = {
+    BlockingUntilSignalNoOpInvokable.lock.synchronized{
+      BlockingUntilSignalNoOpInvokable.lock.wait()
+    }
+  }
+}
+
+object BlockingUntilSignalNoOpInvokable {
+  val lock = new Object
+
+  def triggerExecution(): Unit = {
+    lock.synchronized{
+      lock.notifyAll()
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/673a883e/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerLeaderSessionIDITSuite.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerLeaderSessionIDITSuite.scala
b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerLeaderSessionIDITSuite.scala
deleted file mode 100644
index 78bc0ee..0000000
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerLeaderSessionIDITSuite.scala
+++ /dev/null
@@ -1,110 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.scala.runtime.jobmanager
-
-import java.util.UUID
-
-import akka.actor.ActorSystem
-import akka.actor.Status.Success
-import akka.testkit.{ImplicitSender, TestKit}
-import org.apache.flink.api.common.ExecutionConfig
-import org.apache.flink.runtime.akka.{ListeningBehaviour, AkkaUtils}
-import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable
-import org.apache.flink.runtime.jobgraph.{JobGraph, JobVertex}
-import org.apache.flink.runtime.messages.JobManagerMessages.{LeaderSessionMessage, CancelJob,
-JobResultSuccess, SubmitJob}
-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.{FunSuiteLike, Matchers, BeforeAndAfterAll}
-import org.scalatest.junit.JUnitRunner
-
-@RunWith(classOf[JUnitRunner])
-class JobManagerLeaderSessionIDITSuite(_system: ActorSystem)
-  extends TestKit(_system)
-  with ImplicitSender
-  with FunSuiteLike
-  with Matchers
-  with BeforeAndAfterAll
-  with ScalaTestingUtils {
-
-  val numTaskManagers = 2
-  val taskManagerNumSlots = 2
-  val numSlots = numTaskManagers * taskManagerNumSlots
-
-  val cluster = TestingUtils.startTestingCluster(
-    taskManagerNumSlots,
-    numTaskManagers,
-    TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT);
-
-  def this() = this(ActorSystem("TestingActorSystem", AkkaUtils.getDefaultAkkaConfig))
-
-  override def afterAll(): Unit = {
-    cluster.stop()
-    TestKit.shutdownActorSystem(system)
-  }
-
-  test("A JobManager should not process CancelJob messages with the wrong leader session
ID") {
-    val sender = new JobVertex("BlockingSender");
-    sender.setParallelism(numSlots)
-    sender.setInvokableClass(classOf[BlockingUntilSignalNoOpInvokable])
-    val jobGraph = new JobGraph("TestJob", sender)
-
-    val oldSessionID = UUID.randomUUID()
-
-    val jmGateway = cluster.getLeaderGateway(TestingUtils.TESTING_DURATION)
-    val jm = jmGateway.actor()
-
-    within(TestingUtils.TESTING_DURATION) {
-      jmGateway.tell(SubmitJob(jobGraph, ListeningBehaviour.EXECUTION_RESULT), self)
-
-      expectMsg(Success(jobGraph.getJobID))
-
-      jmGateway.tell(WaitForAllVerticesToBeRunning(jobGraph.getJobID), self)
-
-      expectMsg(AllVerticesRunning(jobGraph.getJobID))
-
-      jm ! LeaderSessionMessage(oldSessionID, CancelJob(jobGraph.getJobID))
-
-      BlockingUntilSignalNoOpInvokable.triggerExecution()
-
-      expectMsgType[JobResultSuccess]
-    }
-  }
-}
-
-class BlockingUntilSignalNoOpInvokable extends AbstractInvokable {
-
-  override def invoke(): Unit = {
-    BlockingUntilSignalNoOpInvokable.lock.synchronized{
-      BlockingUntilSignalNoOpInvokable.lock.wait()
-    }
-  }
-}
-
-object BlockingUntilSignalNoOpInvokable {
-  val lock = new Object
-
-  def triggerExecution(): Unit = {
-    lock.synchronized{
-      lock.notifyAll()
-    }
-  }
-}


Mime
View raw message