-
Notifications
You must be signed in to change notification settings - Fork 28.3k
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.
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
[SPARK-6304][Streaming] Fix checkpointing doesn't retain driver port issue. #5060
Changes from all commits
bbc1c9c
45befaa
2624723
7cc146d
275d252
89b01f5
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -191,8 +191,51 @@ class CheckpointSuite extends TestSuiteBase { | |
} | ||
} | ||
|
||
// This tests if "spark.driver.host" and "spark.driver.port" is set by user, can be recovered | ||
// with correct value. | ||
test("get correct spark.driver.[host|port] from checkpoint") { | ||
val conf = Map("spark.driver.host" -> "localhost", "spark.driver.port" -> "9999") | ||
conf.foreach(kv => System.setProperty(kv._1, kv._2)) | ||
ssc = new StreamingContext(master, framework, batchDuration) | ||
val originalConf = ssc.conf | ||
assert(originalConf.get("spark.driver.host") === "localhost") | ||
assert(originalConf.get("spark.driver.port") === "9999") | ||
|
||
val cp = new Checkpoint(ssc, Time(1000)) | ||
ssc.stop() | ||
|
||
// Serialize/deserialize to simulate write to storage and reading it back | ||
val newCp = Utils.deserialize[Checkpoint](Utils.serialize(cp)) | ||
|
||
val newCpConf = newCp.createSparkConf() | ||
assert(newCpConf.contains("spark.driver.host")) | ||
assert(newCpConf.contains("spark.driver.port")) | ||
assert(newCpConf.get("spark.driver.host") === "localhost") | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This just tests whether its correctly set in the new conf when it is set as system property. you should also test the other case, where the new conf does not have them when they are not in the property, even though it was present in the original conf. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. OK, I will update the test |
||
assert(newCpConf.get("spark.driver.port") === "9999") | ||
|
||
// Check if all the parameters have been restored | ||
ssc = new StreamingContext(null, newCp, null) | ||
val restoredConf = ssc.conf | ||
assert(restoredConf.get("spark.driver.host") === "localhost") | ||
assert(restoredConf.get("spark.driver.port") === "9999") | ||
ssc.stop() | ||
|
||
// If spark.driver.host and spark.driver.host is not set in system property, these two | ||
// parameters should not be presented in the newly recovered conf. | ||
conf.foreach(kv => System.clearProperty(kv._1)) | ||
val newCpConf1 = newCp.createSparkConf() | ||
assert(!newCpConf1.contains("spark.driver.host")) | ||
assert(!newCpConf1.contains("spark.driver.port")) | ||
|
||
// Spark itself will dispatch a random, not-used port for spark.driver.port if it is not set | ||
// explicitly. | ||
ssc = new StreamingContext(null, newCp, null) | ||
val restoredConf1 = ssc.conf | ||
assert(restoredConf1.get("spark.driver.host") === "localhost") | ||
assert(restoredConf1.get("spark.driver.port") !== "9999") | ||
} | ||
|
||
// This tests whether the systm can recover from a master failure with simple | ||
// This tests whether the system can recover from a master failure with simple | ||
// non-stateful operations. This assumes as reliable, replayable input | ||
// source - TestInputDStream. | ||
test("recovery with map and reduceByKey operations") { | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could you keep these alphabetically sorted. Looks cleaner.