Skip to content

[SPARK-16787] SparkContext.addFile() should not throw if called twice with the same file #14396

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.

Sign up for GitHub

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

Closed
JoshRosen wants to merge 6 commits into from
Closed

[SPARK-16787] SparkContext.addFile() should not throw if called twice with the same file #14396

Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
99d9855
Add failing regression test.
JoshRosen Jul 29, 2016
3ecdb88
Remove catch case which masked error for addJar.
JoshRosen Jul 29, 2016
c412f99
Fix bug.
JoshRosen Jul 29, 2016
b98d149
Add back require but weaken it to only accept identical paths.
JoshRosen Jul 29, 2016
0d7dd0d
Improve thread-safety; do not update timestamp for file under assumpt…
JoshRosen Jul 29, 2016
9aa32b3
Attempt to fix PySpark tests by re-adding driver download of added fi…
JoshRosen Jul 29, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Fix bug.
  • Loading branch information
@JoshRosen
JoshRosen committed Jul 29, 2016
commit c412f991c145cf02affd7c2d38de016a9b7f548d
21 changes: 14 additions & 7 deletions core/src/main/scala/org/apache/spark/rpc/netty/NettyStreamManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ package org.apache.spark.rpc.netty
import java.io.File
import java.util.concurrent.ConcurrentHashMap

import org.apache.spark.internal.Logging
import org.apache.spark.network.buffer.{FileSegmentManagedBuffer, ManagedBuffer}
import org.apache.spark.network.server.StreamManager
import org.apache.spark.rpc.RpcEnvFileServer
Expand All @@ -37,7 +38,7 @@ import org.apache.spark.util.Utils
* Only streaming (openStream) is supported.
*/
private[netty] class NettyStreamManager(rpcEnv: NettyRpcEnv)
extends StreamManager with RpcEnvFileServer {
extends StreamManager with RpcEnvFileServer with Logging {

private val files = new ConcurrentHashMap[String, File]()
private val jars = new ConcurrentHashMap[String, File]()
Expand Down Expand Up @@ -66,15 +67,21 @@ private[netty] class NettyStreamManager(rpcEnv: NettyRpcEnv)
}

override def addFile(file: File): String = {
require(files.putIfAbsent(file.getName(), file) == null,
s"File ${file.getName()} already registered.")
val oldFile = files.put(file.getName, file)
if (oldFile != file) {
logInfo(
s"File ${file.getName} will now be served from $file (was previously served from $oldFile)")
}
s"${rpcEnv.address.toSparkURL}/files/${Utils.encodeFileNameToURIRawPath(file.getName())}"
}

override def addJar(file: File): String = {
require(jars.putIfAbsent(file.getName(), file) == null,
s"JAR ${file.getName()} already registered.")
s"${rpcEnv.address.toSparkURL}/jars/${Utils.encodeFileNameToURIRawPath(file.getName())}"
override def addJar(jar: File): String = {
val oldJar = jars.put(jar.getName, jar)
if (oldJar != jar) {
logInfo(
s"JAR ${jar.getName} will now be served from $jar (was previously served from $oldJar)")
}
s"${rpcEnv.address.toSparkURL}/jars/${Utils.encodeFileNameToURIRawPath(jar.getName())}"
}

override def addDirectory(baseUri: String, path: File): String = {
Expand Down