From 7782a304ad105ec95cf62cb799e365e5fb385a69 Mon Sep 17 00:00:00 2001
From: Syed Hashmi <shashmi@cloudera.com>
Date: Tue, 3 Jun 2014 12:04:47 -0700
Subject: [PATCH] [SPARK-1942] Stop clearing spark.driver.port in unit tests
stop resetting spark.driver.port in unit tests (scala, java and python).
Author: Syed Hashmi <shashmi@cloudera.com>
Author: CodingCat <zhunansjtu@gmail.com>
Closes #943 from syedhashmi/master and squashes the following commits:
885f210 [Syed Hashmi] Removing unnecessary file (created by mergetool)
b8bd4b5 [Syed Hashmi] Merge remote-tracking branch 'upstream/master'
b895e59 [Syed Hashmi] Revert "[SPARK-1784] Add a new partitioner"
57b6587 [Syed Hashmi] Revert "[SPARK-1784] Add a balanced partitioner"
1574769 [Syed Hashmi] [SPARK-1942] Stop clearing spark.driver.port in unit tests
4354836 [Syed Hashmi] Revert "SPARK-1686: keep schedule() calling in the main thread"
fd36542 [Syed Hashmi] [SPARK-1784] Add a balanced partitioner
6668015 [CodingCat] SPARK-1686: keep schedule() calling in the main thread
4ca94cc [Syed Hashmi] [SPARK-1784] Add a new partitioner
---
bagel/src/test/scala/org/apache/spark/bagel/BagelSuite.scala | 2 --
core/src/test/java/org/apache/spark/JavaAPISuite.java | 3 ---
core/src/test/scala/org/apache/spark/AkkaUtilsSuite.scala | 4 ----
.../test/scala/org/apache/spark/MapOutputTrackerSuite.scala | 3 ---
.../scala/org/apache/spark/storage/BlockManagerSuite.scala | 2 --
.../src/test/java/org/apache/spark/Java8APISuite.java | 2 --
.../scala/org/apache/spark/graphx/LocalSparkContext.scala | 2 --
.../mllib/classification/JavaLogisticRegressionSuite.java | 1 -
.../spark/mllib/classification/JavaNaiveBayesSuite.java | 1 -
.../org/apache/spark/mllib/classification/JavaSVMSuite.java | 1 -
.../org/apache/spark/mllib/clustering/JavaKMeansSuite.java | 1 -
.../org/apache/spark/mllib/recommendation/JavaALSSuite.java | 1 -
.../org/apache/spark/mllib/regression/JavaLassoSuite.java | 1 -
.../spark/mllib/regression/JavaLinearRegressionSuite.java | 1 -
.../spark/mllib/regression/JavaRidgeRegressionSuite.java | 1 -
.../scala/org/apache/spark/mllib/util/LocalSparkContext.scala | 1 -
python/pyspark/tests.py | 4 ----
repl/src/test/scala/org/apache/spark/repl/ReplSuite.scala | 2 --
.../src/main/scala/org/apache/spark/sql/hive/TestHive.scala | 1 -
.../org/apache/spark/streaming/util/MasterFailureTest.scala | 1 -
.../org/apache/spark/streaming/LocalJavaStreamingContext.java | 4 ----
.../scala/org/apache/spark/streaming/CheckpointSuite.scala | 1 -
.../test/scala/org/apache/spark/streaming/TestSuiteBase.scala | 2 --
23 files changed, 42 deletions(-)
diff --git a/bagel/src/test/scala/org/apache/spark/bagel/BagelSuite.scala b/bagel/src/test/scala/org/apache/spark/bagel/BagelSuite.scala
index 8e0f82ddb8..110bd0a9a0 100644
--- a/bagel/src/test/scala/org/apache/spark/bagel/BagelSuite.scala
+++ b/bagel/src/test/scala/org/apache/spark/bagel/BagelSuite.scala
@@ -38,8 +38,6 @@ class BagelSuite extends FunSuite with Assertions with BeforeAndAfter with Timeo
sc.stop()
sc = null
}
- // To avoid Akka rebinding to the same port, since it doesn't unbind immediately on shutdown
- System.clearProperty("spark.driver.port")
}
test("halting by voting") {
diff --git a/core/src/test/java/org/apache/spark/JavaAPISuite.java b/core/src/test/java/org/apache/spark/JavaAPISuite.java
index 3dd79243ab..7193223add 100644
--- a/core/src/test/java/org/apache/spark/JavaAPISuite.java
+++ b/core/src/test/java/org/apache/spark/JavaAPISuite.java
@@ -68,9 +68,6 @@ public class JavaAPISuite implements Serializable {
public void tearDown() {
sc.stop();
sc = null;
- // To avoid Akka rebinding to the same port, since it doesn't unbind immediately on shutdown
- System.clearProperty("spark.driver.port");
- Utils.deleteRecursively(tempDir);
}
static class ReverseIntComparator implements Comparator<Integer>, Serializable {
diff --git a/core/src/test/scala/org/apache/spark/AkkaUtilsSuite.scala b/core/src/test/scala/org/apache/spark/AkkaUtilsSuite.scala
index c645e4cbe8..4ab870e751 100644
--- a/core/src/test/scala/org/apache/spark/AkkaUtilsSuite.scala
+++ b/core/src/test/scala/org/apache/spark/AkkaUtilsSuite.scala
@@ -39,7 +39,6 @@ class AkkaUtilsSuite extends FunSuite with LocalSparkContext {
val hostname = "localhost"
val (actorSystem, boundPort) = AkkaUtils.createActorSystem("spark", hostname, 0,
conf = conf, securityManager = securityManager)
- System.setProperty("spark.driver.port", boundPort.toString) // Will be cleared by LocalSparkContext
System.setProperty("spark.hostPort", hostname + ":" + boundPort)
assert(securityManager.isAuthenticationEnabled() === true)
@@ -77,7 +76,6 @@ class AkkaUtilsSuite extends FunSuite with LocalSparkContext {
val hostname = "localhost"
val (actorSystem, boundPort) = AkkaUtils.createActorSystem("spark", hostname, 0,
conf = conf, securityManager = securityManager)
- System.setProperty("spark.driver.port", boundPort.toString) // Will be cleared by LocalSparkContext
System.setProperty("spark.hostPort", hostname + ":" + boundPort)
assert(securityManager.isAuthenticationEnabled() === false)
@@ -129,7 +127,6 @@ class AkkaUtilsSuite extends FunSuite with LocalSparkContext {
val hostname = "localhost"
val (actorSystem, boundPort) = AkkaUtils.createActorSystem("spark", hostname, 0,
conf = conf, securityManager = securityManager)
- System.setProperty("spark.driver.port", boundPort.toString) // Will be cleared by LocalSparkContext
System.setProperty("spark.hostPort", hostname + ":" + boundPort)
assert(securityManager.isAuthenticationEnabled() === true)
@@ -182,7 +179,6 @@ class AkkaUtilsSuite extends FunSuite with LocalSparkContext {
val hostname = "localhost"
val (actorSystem, boundPort) = AkkaUtils.createActorSystem("spark", hostname, 0,
conf = conf, securityManager = securityManager)
- System.setProperty("spark.driver.port", boundPort.toString) // Will be cleared by LocalSparkContext
System.setProperty("spark.hostPort", hostname + ":" + boundPort)
assert(securityManager.isAuthenticationEnabled() === true)
diff --git a/core/src/test/scala/org/apache/spark/MapOutputTrackerSuite.scala b/core/src/test/scala/org/apache/spark/MapOutputTrackerSuite.scala
index 6b2571cd92..95ba273f16 100644
--- a/core/src/test/scala/org/apache/spark/MapOutputTrackerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/MapOutputTrackerSuite.scala
@@ -124,9 +124,6 @@ class MapOutputTrackerSuite extends FunSuite with LocalSparkContext {
val (actorSystem, boundPort) = AkkaUtils.createActorSystem("spark", hostname, 0, conf = conf,
securityManager = new SecurityManager(conf))
- // Will be cleared by LocalSparkContext
- System.setProperty("spark.driver.port", boundPort.toString)
-
val masterTracker = new MapOutputTrackerMaster(conf)
masterTracker.trackerActor = actorSystem.actorOf(
Props(new MapOutputTrackerMasterActor(masterTracker, conf)), "MapOutputTracker")
diff --git a/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala b/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
index 00deecc1c3..81bd8257bc 100644
--- a/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
@@ -78,8 +78,6 @@ class BlockManagerSuite extends FunSuite with BeforeAndAfter with PrivateMethodT
}
after {
- System.clearProperty("spark.driver.port")
-
if (store != null) {
store.stop()
store = null
diff --git a/extras/java8-tests/src/test/java/org/apache/spark/Java8APISuite.java b/extras/java8-tests/src/test/java/org/apache/spark/Java8APISuite.java
index 84d3b6f243..c366c10b15 100644
--- a/extras/java8-tests/src/test/java/org/apache/spark/Java8APISuite.java
+++ b/extras/java8-tests/src/test/java/org/apache/spark/Java8APISuite.java
@@ -58,8 +58,6 @@ public class Java8APISuite implements Serializable {
public void tearDown() {
sc.stop();
sc = null;
- // To avoid Akka rebinding to the same port, since it doesn't unbind immediately on shutdown
- System.clearProperty("spark.driver.port");
}
@Test
diff --git a/graphx/src/test/scala/org/apache/spark/graphx/LocalSparkContext.scala b/graphx/src/test/scala/org/apache/spark/graphx/LocalSparkContext.scala
index 51f02f94e0..47594a800a 100644
--- a/graphx/src/test/scala/org/apache/spark/graphx/LocalSparkContext.scala
+++ b/graphx/src/test/scala/org/apache/spark/graphx/LocalSparkContext.scala
@@ -38,8 +38,6 @@ trait LocalSparkContext {
f(sc)
} finally {
sc.stop()
- // To avoid Akka rebinding to the same port, since it doesn't unbind immediately on shutdown
- System.clearProperty("spark.driver.port")
}
}
}
diff --git a/mllib/src/test/java/org/apache/spark/mllib/classification/JavaLogisticRegressionSuite.java b/mllib/src/test/java/org/apache/spark/mllib/classification/JavaLogisticRegressionSuite.java
index d75d3a6b26..faa675b59c 100644
--- a/mllib/src/test/java/org/apache/spark/mllib/classification/JavaLogisticRegressionSuite.java
+++ b/mllib/src/test/java/org/apache/spark/mllib/classification/JavaLogisticRegressionSuite.java
@@ -42,7 +42,6 @@ public class JavaLogisticRegressionSuite implements Serializable {
public void tearDown() {
sc.stop();
sc = null;
- System.clearProperty("spark.driver.port");
}
int validatePrediction(List<LabeledPoint> validationData, LogisticRegressionModel model) {
diff --git a/mllib/src/test/java/org/apache/spark/mllib/classification/JavaNaiveBayesSuite.java b/mllib/src/test/java/org/apache/spark/mllib/classification/JavaNaiveBayesSuite.java
index 743a43a139..1c90522a07 100644
--- a/mllib/src/test/java/org/apache/spark/mllib/classification/JavaNaiveBayesSuite.java
+++ b/mllib/src/test/java/org/apache/spark/mllib/classification/JavaNaiveBayesSuite.java
@@ -44,7 +44,6 @@ public class JavaNaiveBayesSuite implements Serializable {
public void tearDown() {
sc.stop();
sc = null;
- System.clearProperty("spark.driver.port");
}
private static final List<LabeledPoint> POINTS = Arrays.asList(
diff --git a/mllib/src/test/java/org/apache/spark/mllib/classification/JavaSVMSuite.java b/mllib/src/test/java/org/apache/spark/mllib/classification/JavaSVMSuite.java
index 667f76a1bd..31b9f3e8d4 100644
--- a/mllib/src/test/java/org/apache/spark/mllib/classification/JavaSVMSuite.java
+++ b/mllib/src/test/java/org/apache/spark/mllib/classification/JavaSVMSuite.java
@@ -41,7 +41,6 @@ public class JavaSVMSuite implements Serializable {
public void tearDown() {
sc.stop();
sc = null;
- System.clearProperty("spark.driver.port");
}
int validatePrediction(List<LabeledPoint> validationData, SVMModel model) {
diff --git a/mllib/src/test/java/org/apache/spark/mllib/clustering/JavaKMeansSuite.java b/mllib/src/test/java/org/apache/spark/mllib/clustering/JavaKMeansSuite.java
index 0c916ca378..31676e6402 100644
--- a/mllib/src/test/java/org/apache/spark/mllib/clustering/JavaKMeansSuite.java
+++ b/mllib/src/test/java/org/apache/spark/mllib/clustering/JavaKMeansSuite.java
@@ -44,7 +44,6 @@ public class JavaKMeansSuite implements Serializable {
public void tearDown() {
sc.stop();
sc = null;
- System.clearProperty("spark.driver.port");
}
@Test
diff --git a/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java b/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java
index b150334deb..bf2365f820 100644
--- a/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java
+++ b/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java
@@ -42,7 +42,6 @@ public class JavaALSSuite implements Serializable {
public void tearDown() {
sc.stop();
sc = null;
- System.clearProperty("spark.driver.port");
}
static void validatePrediction(MatrixFactorizationModel model, int users, int products, int features,
diff --git a/mllib/src/test/java/org/apache/spark/mllib/regression/JavaLassoSuite.java b/mllib/src/test/java/org/apache/spark/mllib/regression/JavaLassoSuite.java
index f725924a2d..8950b48888 100644
--- a/mllib/src/test/java/org/apache/spark/mllib/regression/JavaLassoSuite.java
+++ b/mllib/src/test/java/org/apache/spark/mllib/regression/JavaLassoSuite.java
@@ -41,7 +41,6 @@ public class JavaLassoSuite implements Serializable {
public void tearDown() {
sc.stop();
sc = null;
- System.clearProperty("spark.driver.port");
}
int validatePrediction(List<LabeledPoint> validationData, LassoModel model) {
diff --git a/mllib/src/test/java/org/apache/spark/mllib/regression/JavaLinearRegressionSuite.java b/mllib/src/test/java/org/apache/spark/mllib/regression/JavaLinearRegressionSuite.java
index 6dc6877691..24c4c20d9a 100644
--- a/mllib/src/test/java/org/apache/spark/mllib/regression/JavaLinearRegressionSuite.java
+++ b/mllib/src/test/java/org/apache/spark/mllib/regression/JavaLinearRegressionSuite.java
@@ -43,7 +43,6 @@ public class JavaLinearRegressionSuite implements Serializable {
public void tearDown() {
sc.stop();
sc = null;
- System.clearProperty("spark.driver.port");
}
int validatePrediction(List<LabeledPoint> validationData, LinearRegressionModel model) {
diff --git a/mllib/src/test/java/org/apache/spark/mllib/regression/JavaRidgeRegressionSuite.java b/mllib/src/test/java/org/apache/spark/mllib/regression/JavaRidgeRegressionSuite.java
index 03714ae7e4..7266eec235 100644
--- a/mllib/src/test/java/org/apache/spark/mllib/regression/JavaRidgeRegressionSuite.java
+++ b/mllib/src/test/java/org/apache/spark/mllib/regression/JavaRidgeRegressionSuite.java
@@ -43,7 +43,6 @@ public class JavaRidgeRegressionSuite implements Serializable {
public void tearDown() {
sc.stop();
sc = null;
- System.clearProperty("spark.driver.port");
}
double predictionError(List<LabeledPoint> validationData, RidgeRegressionModel model) {
diff --git a/mllib/src/test/scala/org/apache/spark/mllib/util/LocalSparkContext.scala b/mllib/src/test/scala/org/apache/spark/mllib/util/LocalSparkContext.scala
index 212fbe9288..0d4868f3d9 100644
--- a/mllib/src/test/scala/org/apache/spark/mllib/util/LocalSparkContext.scala
+++ b/mllib/src/test/scala/org/apache/spark/mllib/util/LocalSparkContext.scala
@@ -34,7 +34,6 @@ trait LocalSparkContext extends BeforeAndAfterAll { self: Suite =>
if (sc != null) {
sc.stop()
}
- System.clearProperty("spark.driver.port")
super.afterAll()
}
}
diff --git a/python/pyspark/tests.py b/python/pyspark/tests.py
index ed90915fcd..1f2a6ea941 100644
--- a/python/pyspark/tests.py
+++ b/python/pyspark/tests.py
@@ -57,10 +57,6 @@ class PySparkTestCase(unittest.TestCase):
def tearDown(self):
self.sc.stop()
sys.path = self._old_sys_path
- # To avoid Akka rebinding to the same port, since it doesn't unbind
- # immediately on shutdown
- self.sc._jvm.System.clearProperty("spark.driver.port")
-
class TestCheckpoint(PySparkTestCase):
diff --git a/repl/src/test/scala/org/apache/spark/repl/ReplSuite.scala b/repl/src/test/scala/org/apache/spark/repl/ReplSuite.scala
index 95460aa205..98cdfd0054 100644
--- a/repl/src/test/scala/org/apache/spark/repl/ReplSuite.scala
+++ b/repl/src/test/scala/org/apache/spark/repl/ReplSuite.scala
@@ -51,8 +51,6 @@ class ReplSuite extends FunSuite {
if (interp.sparkContext != null) {
interp.sparkContext.stop()
}
- // To avoid Akka rebinding to the same port, since it doesn't unbind immediately on shutdown
- System.clearProperty("spark.driver.port")
return out.toString
}
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/TestHive.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/TestHive.scala
index fa7d010459..041e813598 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/TestHive.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/TestHive.scala
@@ -58,7 +58,6 @@ class TestHiveContext(sc: SparkContext) extends LocalHiveContext(sc) {
// By clearing the port we force Spark to pick a new one. This allows us to rerun tests
// without restarting the JVM.
- System.clearProperty("spark.driver.port")
System.clearProperty("spark.hostPort")
override lazy val warehousePath = getTempFilePath("sparkHiveWarehouse").getCanonicalPath
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/util/MasterFailureTest.scala b/streaming/src/main/scala/org/apache/spark/streaming/util/MasterFailureTest.scala
index b3ed302db6..98e17ff92e 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/util/MasterFailureTest.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/util/MasterFailureTest.scala
@@ -187,7 +187,6 @@ object MasterFailureTest extends Logging {
setupCalled = true
// Setup the streaming computation with the given operation
- System.clearProperty("spark.driver.port")
val ssc = new StreamingContext("local[4]", "MasterFailureTest", batchDuration, null, Nil,
Map())
ssc.checkpoint(checkpointDir.toString)
diff --git a/streaming/src/test/java/org/apache/spark/streaming/LocalJavaStreamingContext.java b/streaming/src/test/java/org/apache/spark/streaming/LocalJavaStreamingContext.java
index 849bbf1299..6e1f019000 100644
--- a/streaming/src/test/java/org/apache/spark/streaming/LocalJavaStreamingContext.java
+++ b/streaming/src/test/java/org/apache/spark/streaming/LocalJavaStreamingContext.java
@@ -27,7 +27,6 @@ public abstract class LocalJavaStreamingContext {
@Before
public void setUp() {
- System.clearProperty("spark.driver.port");
System.setProperty("spark.streaming.clock", "org.apache.spark.streaming.util.ManualClock");
ssc = new JavaStreamingContext("local[2]", "test", new Duration(1000));
ssc.checkpoint("checkpoint");
@@ -37,8 +36,5 @@ public abstract class LocalJavaStreamingContext {
public void tearDown() {
ssc.stop();
ssc = null;
-
- // To avoid Akka rebinding to the same port, since it doesn't unbind immediately on shutdown
- System.clearProperty("spark.driver.port");
}
}
diff --git a/streaming/src/test/scala/org/apache/spark/streaming/CheckpointSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/CheckpointSuite.scala
index d20a7b728c..10ad3c9e1a 100644
--- a/streaming/src/test/scala/org/apache/spark/streaming/CheckpointSuite.scala
+++ b/streaming/src/test/scala/org/apache/spark/streaming/CheckpointSuite.scala
@@ -370,7 +370,6 @@ class CheckpointSuite extends TestSuiteBase {
"\n-------------------------------------------\n"
)
ssc = new StreamingContext(checkpointDir)
- System.clearProperty("spark.driver.port")
ssc.start()
val outputNew = advanceTimeWithRealDelay[V](ssc, nextNumBatches)
// the first element will be re-processed data of the last batch before restart
diff --git a/streaming/src/test/scala/org/apache/spark/streaming/TestSuiteBase.scala b/streaming/src/test/scala/org/apache/spark/streaming/TestSuiteBase.scala
index 8036f77c97..cc178fba12 100644
--- a/streaming/src/test/scala/org/apache/spark/streaming/TestSuiteBase.scala
+++ b/streaming/src/test/scala/org/apache/spark/streaming/TestSuiteBase.scala
@@ -153,8 +153,6 @@ trait TestSuiteBase extends FunSuite with BeforeAndAfter with Logging {
// Default after function for any streaming test suite. Override this
// if you want to add your stuff to "after" (i.e., don't call after { } )
def afterFunction() {
- // To avoid Akka rebinding to the same port, since it doesn't unbind immediately on shutdown
- System.clearProperty("spark.driver.port")
System.clearProperty("spark.streaming.clock")
}
--
GitLab