Spark multiversion support

cluster/build.gradle

@@ -44,16 +44,26 @@ dependencies {
     compile project(':snappy-spark:snappy-spark-streaming_' + scalaBinaryVersion)
     compile project(':snappy-spark:snappy-spark-streaming-kafka-0.10_' + scalaBinaryVersion)
     compile project(':snappy-spark:snappy-spark-sql-kafka-0.10_' + scalaBinaryVersion)
+    compile project(':snappy-spark:snappy-spark-avro_' + scalaBinaryVersion)
     compile project(':snappy-spark:snappy-spark-mllib_' + scalaBinaryVersion)
     compile project(':snappy-spark:snappy-spark-yarn_' + scalaBinaryVersion)
     compile project(':snappy-spark:snappy-spark-graphx_' + scalaBinaryVersion)
     compile project(':snappy-spark:snappy-spark-hive-thriftserver_' + scalaBinaryVersion)
     if (rootProject.hasProperty('mesos')) {
       compile project(':snappy-spark:snappy-spark-mesos_' + scalaBinaryVersion)
     }
+    if (rootProject.hasProperty('k8s')) {
+      compile project(':snappy-spark:snappy-spark-kubernetes_' + scalaBinaryVersion)
+    }
+    if (rootProject.hasProperty('flume')) {
+      compile project(':snappy-spark:snappy-spark-streaming-flume_' + scalaBinaryVersion)
+      compile project(':snappy-spark:snappy-spark-streaming-flume-sink_' + scalaBinaryVersion)
+    }
 
     testCompile project(path: ':snappy-spark:snappy-spark-sql_' + scalaBinaryVersion,
         configuration: 'testOutput')
+    testCompile project(path: ':snappy-spark:snappy-spark-sql-kafka-0.10_' + scalaBinaryVersion,
+        configuration: 'testOutput')
   } else {
     compile 'io.snappydata:snappy-spark-unsafe_' + scalaBinaryVersion + ':' + snappySparkVersion
     compile 'io.snappydata:snappy-spark-core_' + scalaBinaryVersion + ':' + snappySparkVersion
@@ -64,19 +74,29 @@ dependencies {
     compile 'io.snappydata:snappy-spark-streaming_' + scalaBinaryVersion + ':' + snappySparkVersion
     compile 'io.snappydata:snappy-spark-streaming-kafka-0.10_' + scalaBinaryVersion + ':' + snappySparkVersion
     compile 'io.snappydata:snappy-spark-sql-kafka-0.10_' + scalaBinaryVersion + ':' + snappySparkVersion
+    compile 'io.snappydata:snappy-spark-avro_' + scalaBinaryVersion + ':' + snappySparkVersion
     compile 'io.snappydata:snappy-spark-mllib_' + scalaBinaryVersion + ':' + snappySparkVersion
     compile 'io.snappydata:snappy-spark-yarn_' + scalaBinaryVersion + ':' + snappySparkVersion
     compile 'io.snappydata:snappy-spark-graphx_' + scalaBinaryVersion + ':' + snappySparkVersion
     compile 'io.snappydata:snappy-spark-hive-thriftserver_' + scalaBinaryVersion + ':' + snappySparkVersion
     if (rootProject.hasProperty('mesos')) {
       compile 'io.snappydata:snappy-spark-mesos_' + scalaBinaryVersion + ':' + snappySparkVersion
     }
+    if (rootProject.hasProperty('k8s')) {
+      compile 'io.snappydata:snappy-spark-kubernetes_' + scalaBinaryVersion + ':' + snappySparkVersion
+    }
+    if (rootProject.hasProperty('flume')) {
+      compile 'io.snappydata:snappy-spark-streaming-flume_' + scalaBinaryVersion + ':' + snappySparkVersion
+      compile 'io.snappydata:snappy-spark-streaming-flume-sink_' + scalaBinaryVersion + ':' + snappySparkVersion
+    }
 
     testCompile group: 'io.snappydata', name: 'snappy-spark-sql_' + scalaBinaryVersion,
                 version: snappySparkVersion, classifier: 'tests'
+    testCompile group: 'io.snappydata', name: 'snappy-spark-sql-kafka-0.10_' + scalaBinaryVersion,
+                version: snappySparkVersion, classifier: 'tests'
   }
 
-  compile (project(':snappy-core_' + scalaBinaryVersion)) {
+  compile (project(coreProjectName)) {
     exclude(group: 'org.apache.spark', module: 'spark-unsafe_' + scalaBinaryVersion)
     exclude(group: 'org.apache.spark', module: 'spark-core_' + scalaBinaryVersion)
     exclude(group: 'org.apache.spark', module: 'spark-catalyst_' + scalaBinaryVersion)
@@ -85,10 +105,13 @@ dependencies {
     exclude(group: 'org.apache.spark', module: 'spark-streaming_' + scalaBinaryVersion)
     exclude(group: 'org.apache.spark', module: 'spark-streaming-kafka-0-10_' + scalaBinaryVersion)
     exclude(group: 'org.apache.spark', module: 'spark-sql-kafka-0-10_' + scalaBinaryVersion)
+    exclude(group: 'org.apache.spark', module: 'spark-avro_' + scalaBinaryVersion)
     exclude(group: 'org.apache.spark', module: 'spark-mllib_' + scalaBinaryVersion)
     exclude(group: 'org.eclipse.jetty', module: 'jetty-servlet')
   }
-  testCompile (project(path: ':snappy-core_' + scalaBinaryVersion, configuration: 'testOutput')) {
+  compile project(compatProjectName)
+
+  testCompile (project(path: coreProjectName, configuration: 'testOutput')) {
     exclude(group: 'org.apache.spark', module: 'spark-unsafe_' + scalaBinaryVersion)
     exclude(group: 'org.apache.spark', module: 'spark-core_' + scalaBinaryVersion)
     exclude(group: 'org.apache.spark', module: 'spark-catalyst_' + scalaBinaryVersion)
@@ -97,6 +120,7 @@ dependencies {
     exclude(group: 'org.apache.spark', module: 'spark-streaming_' + scalaBinaryVersion)
     exclude(group: 'org.apache.spark', module: 'spark-streaming-kafka-0-10_' + scalaBinaryVersion)
     exclude(group: 'org.apache.spark', module: 'spark-sql-kafka-0-10_' + scalaBinaryVersion)
+    exclude(group: 'org.apache.spark', module: 'spark-avro_' + scalaBinaryVersion)
     exclude(group: 'org.apache.spark', module: 'spark-mllib_' + scalaBinaryVersion)
     exclude(group: 'org.eclipse.jetty', module: 'jetty-servlet')
   }
@@ -138,36 +162,15 @@ dependencies {
   testCompile "org.scalatest:scalatest_${scalaBinaryVersion}:${scalatestVersion}"
 
   if (new File(rootDir, 'aqp/build.gradle').exists() && rootProject.hasProperty('snappydata.enterprise')) {
-    testRuntime project(':snappy-aqp_' + scalaBinaryVersion)
+    testRuntime project(aqpProjectName)
   }
   testRuntime files("${projectDir}/../tests/common/src/main/resources")
   testRuntime "org.pegdown:pegdown:${pegdownVersion}"
 }
 
 // Creates the version properties file and writes it to the resources dir
-task createVersionPropertiesFile(dependsOn: 'processResources') {
-  def propertiesDir = file("${sourceSets.main.scala.outputDir}/io/snappydata")
-  outputs.file "${propertiesDir}/SnappyDataVersion.properties"
-  inputs.file "${rootProject.projectDir}/build.gradle"
-
-  doLast {
-
-    def props = [
-      'Product-Name'      : productName,
-      'Product-Version'   : version,
-      'Build-Id'          : buildIdPrefix + buildNumber,
-      'Build-Date'        : buildDate,
-      'Build-Platform'    : osName.getName() + osVersion + osArch,
-      'Build-Java-Version': jdkVersion,
-      'Source-Date'       : sourceDate,
-      'Source-Revision'   : commitId,
-      'Source-Repository' : gitBranch,
-    ]
-
-   writeProperties(propertiesDir, 'SnappyDataVersion.properties',
-        "Properties that control what version ${productName} will think it is. Changing these values may cause ${productName} to no longer function.", props)
-  }
-}
+createVersionPropertiesTask(project, 'SnappyDataVersion.properties', productName,
+    sourceDate, commitId, gitBranch)
 
 compileJava.dependsOn createVersionPropertiesFile
 

cluster/src/dunit/scala/io/snappydata/cluster/SplitSnappyClusterDUnitTest.scala

@@ -26,21 +26,23 @@ import scala.concurrent.{Await, Future}
 import scala.language.postfixOps
 import scala.reflect.io.Path
 import scala.util.{Failure, Success, Try}
+
 import com.gemstone.gemfire.internal.cache.PartitionedRegion
 import com.pivotal.gemfirexd.internal.engine.Misc
 import io.snappydata.core.{TestData, TestData2}
 import io.snappydata.test.dunit.{AvailablePortHelper, SerializableRunnable}
 import io.snappydata.util.TestUtils
 import io.snappydata.{ColumnUpdateDeleteTests, ConcurrentOpsTests, Property, SnappyTableStatsProviderService}
 import org.junit.Assert
+
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.encoders.{ExpressionEncoder, RowEncoder}
 import org.apache.spark.sql.execution.CatalogStaleException
 import org.apache.spark.sql.execution.columnar.impl.ColumnFormatRelation
 import org.apache.spark.sql.kafka010.KafkaTestUtils
 import org.apache.spark.sql.store.{SnappyJoinSuite, StoreUtils}
-import org.apache.spark.sql.streaming.ProcessingTime
+import org.apache.spark.sql.streaming.Trigger
 import org.apache.spark.sql.types.{DateType, StringType, StructField, StructType}
 import org.apache.spark.sql.udf.UserDefinedFunctionsDUnitTest
 import org.apache.spark.{Logging, SparkConf, SparkContext}
@@ -49,8 +51,7 @@ import org.apache.spark.{Logging, SparkConf, SparkContext}
  * Basic tests for non-embedded mode connections to an embedded cluster.
  */
 class SplitSnappyClusterDUnitTest(s: String)
-    extends ClusterManagerTestBase(s) with SplitClusterDUnitTestBase
-      with Serializable {
+    extends ClusterManagerTestBase(s) with SplitClusterDUnitTestBase with Serializable {
 
   override val locatorNetPort: Int = testObject.locatorNetPort
 
@@ -255,7 +256,7 @@ class SplitSnappyClusterDUnitTest(s: String)
     if (jars.count() > 0) {
       var str = msg
       jars.collect().foreach(x => str += s"$x,")
-      assert(false, str)
+      assert(assertion = false, str)
     }
   }
 
@@ -291,13 +292,16 @@ class SplitSnappyClusterDUnitTest(s: String)
     sns.sql("deploy package testsch.mongo-spark_v1.2  'org.mongodb.spark:mongo-spark" +
         "-connector_2.11:2.2.2'")
     sns.sql("undeploy  testsch.mongo-spark_v1.2")
-    sns.sql(s"""deploy package "testsch"."mongo-spark_v1.3"  'org.mongodb.spark:mongo""" +
+    sns.sql(
+      s"""deploy package "testsch"."mongo-spark_v1.3"  'org.mongodb.spark:mongo""" +
           "-spark-connector_2.11:2.2.2'")
     sns.sql(s"""undeploy  "testsch"."mongo-spark_v1.3" """)
-    sns.sql(s"""deploy package testsch."mongo-spark_v1.4"  'org.mongodb.spark:mongo""" +
+    sns.sql(
+      s"""deploy package testsch."mongo-spark_v1.4"  'org.mongodb.spark:mongo""" +
           "-spark-connector_2.11:2.2.2'")
     sns.sql(s"""undeploy  testsch."mongo-spark_v1.4" """)
-    sns.sql(s"""deploy package "testsch".mongo-spark_v1.5  'org.mongodb.spark:mongo""" +
+    sns.sql(
+      s"""deploy package "testsch".mongo-spark_v1.5  'org.mongodb.spark:mongo""" +
           "-spark-connector_2.11:2.2.2'")
     sns.sql(s"""undeploy "testsch".mongo-spark_v1.5 """)
     assert(sns.sql("list packages").count() == 0)
@@ -346,7 +350,7 @@ class SplitSnappyClusterDUnitTest(s: String)
         "Deploy command should have failed because of the duplicate alias.")
       case Failure(error) =>
         assert(error.getMessage.contains("Name 'akka-v1' specified in" +
-          " context 'of deploying jars/packages' is not unique."))
+            " context 'of deploying jars/packages' is not unique."))
     }
     sns.sql("undeploy  akka-v1")
     functionCheck(sns, "Some jars/packages are not cleaned up! ")
@@ -927,7 +931,7 @@ object SplitSnappyClusterDUnitTest
         .set("snappydata.connection", connectionURL)
         .set("snapptdata.sql.planCaching", random.nextBoolean().toString)
         .set(Property.TestDisableCodeGenFlag.name, "false")
-    logInfo("Spark conf:" + conf.getAll.toString)
+    logInfo("Spark conf: " + conf.getAll.mkString(", "))
 
     val sc = SparkContext.getOrCreate(conf)
     //      sc.setLogLevel("DEBUG")
@@ -1411,7 +1415,7 @@ object SplitSnappyClusterDUnitTest
           .writeStream
           .format("snappysink")
           .queryName(tableName)
-          .trigger(ProcessingTime("1 seconds"))
+          .trigger(Trigger.ProcessingTime("1 seconds"))
           .option("tableName", tableName)
           .option("checkpointLocation", s"$testTempDir/checkpoint")
           .start()

cluster/src/dunit/scala/io/snappydata/externalstore/ColumnTableDUnitTest.scala

@@ -556,15 +556,15 @@ class ColumnTableDUnitTest(s: String) extends ClusterManagerTestBase(s) {
         "BUCKETS '1'," +
         "REDUNDANCY '2')")
 
-    snc.sql("insert into COLUMNTABLE4 VALUES(1,11)")
-    snc.sql("insert into COLUMNTABLE4 VALUES(2,11)")
-    snc.sql("insert into COLUMNTABLE4 VALUES(3,11)")
+    snc.sql("insert into COLUMNTABLE4 VALUES(1,11,null,null)")
+    snc.sql("insert into COLUMNTABLE4 VALUES(2,11,null,null)")
+    snc.sql("insert into COLUMNTABLE4 VALUES(3,11,null,null)")
 
-    snc.sql("insert into COLUMNTABLE4 VALUES(4,11)")
-    snc.sql("insert into COLUMNTABLE4 VALUES(5,11)")
-    snc.sql("insert into COLUMNTABLE4 VALUES(6,11)")
+    snc.sql("insert into COLUMNTABLE4 VALUES(4,11,null,null)")
+    snc.sql("insert into COLUMNTABLE4 VALUES(5,11,null,null)")
+    snc.sql("insert into COLUMNTABLE4 VALUES(6,11,null,null)")
 
-    snc.sql("insert into COLUMNTABLE4 VALUES(7,11)")
+    snc.sql("insert into COLUMNTABLE4 VALUES(7,11,null,null)")
 
     var data =
       Seq(Seq(1, 2, 3, 4), Seq(7, 8, 9, 10), Seq(9, 2, 3, 4), Seq(4, 2, 5, 7), Seq(5, 6, 2, 3))