Use the Executor's ClassLoader in sc.objectFile().

[darabos]

This makes it possible to read classes from the object file which were specified in the user-provided jars. (By default ObjectInputStream uses latestUserDefinedLoader, which may or may not be the right one.)

I created this because I ran into the following problem. I have x:RDD[X] with X being defined in the jar that I provide to SparkContext. I save it with x.saveAsObjectFile("x"). I try to load it with sc.objectFileX. It fails with ClassNotFoundException.

After a good while of debugging I figured out that Utils.deserialize() most likely uses the ClassLoader of Utils. This is the bootstrap ClassLoader, so it is not aware of the dynamically added jars. This patch fixes the issue.

A more robust fix would be to always default to Thread.currentThread.getContextClassLoader. This would prevent this problem from biting anyone in the future. It would be a bit harder to test though. On the topic of testing, if you'd like to see tests for this, I will need some hand-holding. Thanks!

[mateiz]

Jenkins, test this please

[mateiz]

@darabos unfortunately testing this is kind of tricky because we need to create a .jar with a class that is not normally on the classpath. There's some TestUtils code to create this in https://github.com/apache/spark/pull/119/files, which compiles a new class. Maybe we can get for that to get merged. We can also try to create a dummy JAR through our own build process and find it, but it will be kind of brittle.

Once we have such a JAR, I'd add a test to FileServerSuite that tries to use those classes in tasks. This is already used to test that addJar ships the JAR over.

[mateiz]

BTW the "test failure" is actually a style error if you click it -- you have a line longer than 100 characters.

[darabos]

I'll fix the style error and look at writing a test. Thanks for the pointers!

What do you think about using Thread.currentThread.getContextClassLoader in deserialize by default? Since deserialize is not part of the public interface I think this may be safe and useful to do.

[darabos]

Sorry, the delay is my fault. I was too busy to get around to the test so far, but I still intend to do it. At least I've read some Spark tests :).

[darabos]

I couldn't figure out how to get a separate jar built for use in this test. (I'm new to Java/Scala build systems.) Anyway you say it would be brittle. I'll just wait for TestUtils from #119. It looks like a generally handy tool for testing jar-related behavior.

[pwendell]

Hey forgot to ping this, but I included that test utility a while ago in a pull request that we merged.

[pwendell]

Hey forgot to ping this, but I included that test utility a while ago in a pull request that we merged.

[rxin]

@darabos do you mind picking this up now the test util was merged?

[darabos]

Sorry for leaving this hanging. I'll take a look at the test.

[darabos]

I was so slow, Bogdan has already fixed this in #821. Anyway, here's the belated test. It's probably still useful to avoid regressions. I tested the test by reverting Bogdan's change, and the test then fails with ClassNotFoundException: FileSuiteObjectFileTest. Either both the fix and the test are correct, or they are both bugged :).

[pwendell]

Cool - thanks for contributing the test! Jenkins, test this please. LGTM pending tests.

[SparkQA]

QA tests have started for PR 181. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16563/consoleFull

[darabos]

I think Jenkins means to say it's all good.

Attempting to post to Github:
{"body": "QA results for PR 181:<br>
          - This patch PASSES unit tests.<br>
          - This patch merges cleanly<br>
          - This patch adds the following public classes (experimental):<br>
          -      "public class " className + " { @Override public String toString() { " +<br>
                 "public class " + className + " implements java.io.Serializable {" +<br>
          <br>
          For more information see test ouptut:<br>
          https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16563/consoleFull"}
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed

  0     0    0     0    0     0      0      0 --:--:-- --:--:-- --:--:--     0
100   553  100    99  100   454    288   1322 --:--:-- --:--:-- --:--:--  1351
HTTP/1.1 400 Bad Request
Server: GitHub.com
Date: Fri, 11 Jul 2014 09:38:58 GMT
Content-Type: application/json; charset=utf-8
Status: 400 Bad Request

(Extra linebreaks by me.) Not sure what went wrong with the request. Perhaps the quotes in the mistakenly identified class names got encoded wrong. But it looks like I probably didn't break anything :).

[pwendell]

@darabos yeah you're just seeing a bug in some new code we wrote to post information about pull requests to github. I think we don't properly escape quotes in this code. Anyways, no problem, tests have passed so I can merge this!

[ch0ice]

This problem arose again for me, and I reproduced it when I converted byte into protoBuf after redis checked the data.
The following code in the deserialization (Utils deserialize (value, Utils getContextOrSparkClassLoader ())) when there is an error

        Map<byte[],byte[]> hResult = jedisUtil.hgetAll((REDIS_LASTPOINT_PREFIX + serviceId));
        hResult.forEach((key,value)->{
            LocationPackage.LocationMessage lastLocationPackage = Utils.deserialize(value,Utils.getContextOrSparkClassLoader());
            if(lastLocationPackage.getLocation().getLocationTime() >= sTime && lastLocationPackage.getLocation().getLocationTime() <= eTime){
                String gpsRedisKey = REDIS_GPSLIST_PREFIX + serviceId + "_" + lastLocationPackage.getLocation().getTerminalId();
                redisKeys.add(gpsRedisKey);
            }

        });

ERROR:
java.lang.RuntimeException: Unable to find proto buffer class
at com.google.protobuf.GeneratedMessageLite$SerializedForm.readResolve(GeneratedMessageLite.java:775)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at java.io.ObjectStreamClass.invokeReadResolve(ObjectStreamClass.java:1104)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1810)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371)
at org.apache.spark.util.Utils$.deserialize(Utils.scala:161)
at org.apache.spark.util.Utils.deserialize(Utils.scala)
at com.location.platform.analysis.util.AnalysisUtil.lambda$getRedisData$0(AnalysisUtil.java:310)
at com.location.platform.analysis.util.AnalysisUtil$$Lambda$17/489334328.accept(Unknown Source)
at java.util.Map.forEach(Map.java:630)
at com.location.platform.analysis.util.AnalysisUtil.getRedisData(AnalysisUtil.java:307)
at com.location.platform.analysis.util.AnalysisUtil.getMongoAndRedisDataset(AnalysisUtil.java:372)
at com.location.platform.analysis.bi.TodayOdometerAnalysis.lambda$main$0(TodayOdometerAnalysis.java:96)
at com.location.platform.analysis.bi.TodayOdometerAnalysis$$Lambda$1/1413730361.accept(Unknown Source)
at java.util.ArrayList.forEach(ArrayList.java:1249)
at com.location.platform.analysis.bi.TodayOdometerAnalysis.main(TodayOdometerAnalysis.java:76)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:775)
at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:180)
at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:205)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:119)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.lang.ClassNotFoundException: com.location.platform.common.protobuf.LocationPackage$LocationMessage
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:264)
at com.google.protobuf.GeneratedMessageLite$SerializedForm.readResolve(GeneratedMessageLite.java:768)
... 28 more

Only if there is a problem with the cluster, local mode is fine.