upgrade to Spark 2.0

[skaarthik]

No description provided.

[codecov-io]

Current coverage is 81.91% (diff: 79.31%)

Merging #546 into master will decrease coverage by 0.21%

@@             master       #546   diff @@
==========================================
  Files            66         67     +1   
  Lines          7341       7447   +106   
  Methods        1165       1171     +6   
  Messages          0          0          
  Branches        945        953     +8   
==========================================
+ Hits           6029       6100    +71   
- Misses          967        996    +29   
- Partials        345        351     +6   

Powered by Codecov. Last update ad7cecf...7157cd7

[qualiu]

Failed in Kafka test

git checkout master
git pull upstream master
git branch rev2.0
git checkout rev2.0
git pull https://github.com/skaarthik/Mobius sparkupgradev3

build\Build.cmd
build\localmode\RunSamples.cmd

git clone https://github.com/qualiu/testMobius
testMobius\csharp\Build.cmd

testMobius\csharp\testArgsQuotes\test.bat %CD%  arg2@*#:,+.-\/~  Pi* d:\tmp
testMobius\csharp\testKeyValueStream\test-by-starting-socket.bat -p 9123 -w 5 -s 1 -d 1

testMobius\csharp\txtStreamTest\test.bat D:\cosmos\download-stream\tenant\csv-2015-10-01

testMobius\tools\start-zookeeper-kafka.bat
testMobius\csharp\kafkaStreamTest\test.bat WindowSlideTest -Topics test -d 1 -w 5 -s 1

last test -- Kafka failed : (succeeded in previous test)

testMobius\csharp\kafkaStreamTest\test.bat WindowSlideTest -Topics test -d 1 -w 5 -s 1 2>&1 | lzmw -it "exception|failed"
:64: [CSharpBackendHandler] <init> on object of type NullObject failed
:66: java.lang.reflect.InvocationTargetException
:93: Caused by: java.lang.IllegalArgumentException: requirement failed: Spark Streaming cannot be initialized with both SparkContext and checkpoint as null
:108: public void org.apache.spark.streaming.api.java.JavaStreamingContext.awaitTermination() throws java.lang.InterruptedException
:135: public boolean org.apache.spark.streaming.api.java.JavaStreamingContext.awaitTerminationOrTimeout(long) throws java.lang.InterruptedException
:136: public final void java.lang.Object.wait(long,int) throws java.lang.InterruptedException
:137: public final native void java.lang.Object.wait(long) throws java.lang.InterruptedException
:138: public final void java.lang.Object.wait() throws java.lang.InterruptedException
:147: [2016-08-24 10:30:58,560] [1] [ERROR] [Microsoft.Spark.CSharp.Interop.Ipc.JvmBridge] - JVM method execution failed: Constructor failed for class org.apache.spark.streaming.api.java.JavaStreamingContext when called with 1 parameters ([Index=1, Type=String, Value=checkDir], )
:148: [2016-08-24 10:30:58,560] [1] [ERROR] [Microsoft.Spark.CSharp.Interop.Ipc.JvmBridge] - java.lang.IllegalArgumentException: requirement failed: Spark Streaming cannot be initialized with both SparkContext and checkpoint as null
:180: [2016-08-24 10:30:58,576] [1] [ERROR] [Microsoft.Spark.CSharp.Interop.Ipc.JvmBridge] - JVM method execution failed: Constructor failed for class org.apache.spark.streaming.api.java.JavaStreamingContext when called with 1 parameters ([Index=1, Type=String, Value=checkDir], )
:186: Unhandled Exception: System.Exception: JVM method execution failed: Constructor failed for class org.apache.spark.streaming.api.java.JavaStreamingContext when called with 1 parameters ([Index=1, Type=String, Value=checkDir], )
:194: Exception caught: An existing connection was forcibly closed by the remote host
:195: java.io.IOException: An existing connection was forcibly closed by the remote host
matched 14 lines(6.22%) from stdin, read 225 lines. Used 13.773301 s. from 2016-08-24 10:30:49.830576 to 10:31:03.603877

[skaarthik]

@qualiu we need a simpler repro. Share a simple driver program that uses Kafka and share the error message if any. The error message "requirement failed: Spark Streaming cannot be initialized with both SparkContext and checkpoint as null" from your comment indicates the possibility of user error

[qualiu]

@skaarthik : the repro just as I did is simple : clone the test, build, run : (most easy way is clone/build/run in the Mobius code directory)

git clone https://github.com/qualiu/testMobius
testMobius\csharp\Build.cmd
testMobius\tools\start-zookeeper-kafka.bat
testMobius\csharp\kafkaStreamTest\test.bat WindowSlideTest -Topics test -d 1 -w 5 -s 1

And I'm looking for it now.
(I've updated the testMobius scripts and set INFO level to avoid too many logs; Please update if you've pulled/cloned testMobius)

[qualiu]

I updated the Kafka test case, and found issue point : d:\msgit\revMobius\csharp\adapter\microsoft.spark.csharp\streaming\streamingcontext.cs
Line 75 : public static StreamingContext GetOrCreate(string checkpointPath, Func creatingFunc)
Will come back to dig after lunch..

csharp\kafkaStreamTest\test.bat WindowSlideTest -Topics test -w 5 -s 1 -d 1

[2016-08-24 12:22:18,453] [1] [INFO ] [kafkaStreamTest.WindowSlideTest] - begin test[1]-1 , sparkContext = Microsoft.Spark.CSharp.Core.SparkContext
will CreateStreamingContext as not exist  checkpointPath : checkDir
[2016-08-24 12:22:18,586] [1] [INFO ] [Microsoft.Spark.CSharp.Proxy.Ipc.StreamingContextIpcProxy] - Callback server port number is 2309
[CSharpBackendHandler] Connecting to a callback server at port 2309
[CSharpBackendHandler] createDirectStreamWithoutMessageHandler on object of type org.apache.spark.streaming.kafka.KafkaUtilsPythonHelper failed
null
java.lang.reflect.InvocationTargetException
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:606)
        at org.apache.spark.api.csharp.CSharpBackendHandler.handleMethodCall(CSharpBackendHandler.scala:159)
        at org.apache.spark.api.csharp.CSharpBackendHandler.handleBackendRequest(CSharpBackendHandler.scala:106)

[skaarthik]

Thanks - that was helpful.

You need to provide kafka jar in --jars parameter when you submit the job.

I will update the download tool to get this jar under dependencies folder.

[skaarthik]

I do not know what caused codecov drop in the latest iteration. There was no C#/scala/C++ code changes between the previous build iteration where codecov check succeeded and the last one where it failed. For now, I am going to ignore that failure.

[qualiu]

It works now, lost the jar you mentioned. : need to download/use spark-streaming-kafka-0-8-assembly_2.11-2.0.0.jar instead of the old spark-streaming-kafka-assembly_2.10-1.6.1.jar in build\dependencies

And I provided the jars to --jars : In fact, when you run the testMobius\csharp\kafkaStreamTest\test.bat it'll automatically find and add the jars in %SPARKCLR_HOME% . (You can change it if you defined %SparkOptions% )

The test tools in https://github.com/qualiu/testMobius is designed to provide fast/consistent/simple experience to test/repro/validate Mobius.

[qualiu]

LGTM

[skaarthik]

Thanks @qualiu for the validation