[SNAP-1190] Changes for updates to spark layer (#435)

- register LaunchTasks with PooledKryoSerializer if present
- avoid sending useless TaskState.RUNNING updates (heartbeat messages are separate)
- adding the new kafka-sql project

cluster/src/main/scala/org/apache/spark/executor/SnappyCoarseGrainedExecutorBackend.scala

@@ -17,14 +17,15 @@
 package org.apache.spark.executor
 
 import java.net.URL
+import java.nio.ByteBuffer
 
 import com.pivotal.gemfirexd.internal.engine.store.GemFireStore
 import io.snappydata.cluster.ExecutorInitiator
 
-import org.apache.spark.SparkEnv
 import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.rpc.RpcEnv
 import org.apache.spark.sql.SnappyContext
+import org.apache.spark.{SparkEnv, TaskState}
 
 class SnappyCoarseGrainedExecutorBackend(
     override val rpcEnv: RpcEnv,
@@ -36,6 +37,7 @@ class SnappyCoarseGrainedExecutorBackend(
     env: SparkEnv)
     extends CoarseGrainedExecutorBackend(rpcEnv, driverUrl,
       executorId, hostName, cores, userClassPath, env) {
+
   override def onStop() {
     SnappyContext.clearStaticArtifacts()
     exitWithoutRestart()
@@ -50,6 +52,16 @@ class SnappyCoarseGrainedExecutorBackend(
       userClassPath, new SnappyUncaughtExceptionHandler(this),
       isLocal = false)
 
+  /**
+   * Avoid sending any message for TaskState.RUNNING which serves no purpose.
+   */
+  override def statusUpdate(taskId: Long, state: TaskState.TaskState,
+      data: ByteBuffer): Unit = {
+    if ((state ne TaskState.RUNNING) || data.hasRemaining) {
+      super.statusUpdate(taskId, state, data)
+    }
+  }
+
   /**
    * Snappy addition (Replace System.exit with exitExecutor). We could have
    * added functions calling System.exit to SnappyCoarseGrainedExecutorBackend

core/src/main/scala/org/apache/spark/serializer/PooledKryoSerializer.scala

@@ -136,6 +136,14 @@ final class PooledKryoSerializer(conf: SparkConf)
     kryo.register(classOf[MultiBucketExecutorPartition],
       new KryoSerializableSerializer)
 
+    try {
+      val launchTasksClass = Utils.classForName(
+        "org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.LaunchTasks")
+      kryo.register(launchTasksClass, new KryoSerializableSerializer)
+    } catch {
+      case _: ClassNotFoundException => // ignore
+    }
+
     kryo
   }
 

settings.gradle

@@ -54,6 +54,7 @@ if (new File(rootDir, 'spark/build.gradle').exists()) {
   include ':snappy-spark:snappy-spark-streaming-flume-sink_' + scalaBinaryVersion
   include ':snappy-spark:snappy-spark-streaming-kafka-0.8_' + scalaBinaryVersion
   include ':snappy-spark:snappy-spark-streaming-kafka-0.10_' + scalaBinaryVersion
+  include ':snappy-spark:snappy-spark-sql-kafka-0.10_' + scalaBinaryVersion
   include ':snappy-spark:snappy-spark-examples_' + scalaBinaryVersion
   include ':snappy-spark:snappy-spark-repl_' + scalaBinaryVersion
   include ':snappy-spark:snappy-spark-launcher_' + scalaBinaryVersion
@@ -89,6 +90,7 @@ if (new File(rootDir, 'spark/build.gradle').exists()) {
   project(':snappy-spark:snappy-spark-streaming-flume-sink_' + scalaBinaryVersion).projectDir = "$rootDir/spark/external/flume-sink" as File
   project(':snappy-spark:snappy-spark-streaming-kafka-0.8_' + scalaBinaryVersion).projectDir = "$rootDir/spark/external/kafka-0-8" as File
   project(':snappy-spark:snappy-spark-streaming-kafka-0.10_' + scalaBinaryVersion).projectDir = "$rootDir/spark/external/kafka-0-10" as File
+  project(':snappy-spark:snappy-spark-sql-kafka-0.10_' + scalaBinaryVersion).projectDir = "$rootDir/spark/external/kafka-0-10-sql" as File
   project(':snappy-spark:snappy-spark-examples_' + scalaBinaryVersion).projectDir = "$rootDir/spark/examples" as File
   project(':snappy-spark:snappy-spark-repl_' + scalaBinaryVersion).projectDir = "$rootDir/spark/repl" as File
   project(':snappy-spark:snappy-spark-launcher_' + scalaBinaryVersion).projectDir = "$rootDir/spark/launcher" as File