SPARKC-458: Fix for 2.12.10 Build Issues and Review

The Spark Packages plugin was automatically adding a default spark
version of 1.4.0 to all projects which didn't inherit the
sparkPackagesSettings group. With this setting added to the
cassandra-embedded project the Scala 2.12 integration test can compile.
Additionally the SparkRepl class exists in Scala Specific source
directories, there was no Scala 2.12 version so that is added as well.

The Cassandra Java Util test also required a small fix because it was
doing a strict string comparison of typetag's string representations. In
Scala 2.12 java.lang is no longer added to Java types (getSimpleName) vs
(getName) causing a bunch of mismatches. We remove all "java.lang." from
the strings before comparing to fix the mismatch and be compatible with
both Scala versions.

In Scala 2.12 there seems to be a sensitivity to untyped implicits and their
position in the code. Previously, the unannotated implicit connector in
the CassandraSourceRelation was declared after it's usage in
"saveToCassandra", in Scala 2.12 this caused the implicit to be ignored
and the default of (CassandraConnector(sparkConf)) to be used instead.
This broke configuring the DataSource using an alternative connection
conf. In Scala 2.11 there is no problem with the ordering.

To fix this we move and explicitly type the implicit val declarations to the
top of the file.

Adds travis configuration for running 2.12 Integration Tests

Author: RussellSpitzer

.travis.yml

@@ -7,6 +7,7 @@ sudo: required
 dist: trusty
 scala:
   - 2.11.12
+  - 2.12.10
 
 env:
  - CASSANDRA_VERSION=2.1.15

project/SparkCassandraConnectorBuild.scala

@@ -39,7 +39,7 @@ object CassandraSparkBuild extends Build {
   lazy val cassandraServerProject = Project(
     id = "cassandra-server",
     base = file(namespace),
-    settings = defaultSettings ++ Seq(
+    settings = defaultSettings ++ sparkPackageSettings ++ Seq(
       libraryDependencies ++= Seq(Artifacts.cassandraServer % "it", Artifacts.airlift),
       Testing.cassandraServerClasspath := {
         (fullClasspath in IntegrationTest).value.map(_.data.getAbsoluteFile).mkString(File.pathSeparator)

spark-cassandra-connector-embedded/scala-2.12/src/main/scala/com/datastax/spark/connector/embedded/SparkRepl.scala

@@ -0,0 +1,42 @@
+package com.datastax.spark.connector.embedded
+
+import java.io._
+import java.net.URLClassLoader
+
+import org.apache.spark.SparkConf
+import org.apache.spark.repl.{Main, SparkILoop}
+
+import scala.collection.mutable.ArrayBuffer
+import scala.tools.nsc.GenericRunnerSettings
+
+object SparkRepl {
+
+  def runInterpreter(input: String, conf: SparkConf): String = {
+    val in = new BufferedReader(new StringReader(input + "\n"))
+    val out = new StringWriter()
+    val cl = getClass.getClassLoader
+    var paths = new ArrayBuffer[String]
+    cl match {
+      case urlLoader: URLClassLoader =>
+        for (url <- urlLoader.getURLs) {
+          if (url.getProtocol == "file") {
+            paths += url.getFile
+          }
+        }
+      case _ =>
+    }
+
+    Main.conf.setAll(conf.getAll)
+    val interp = new SparkILoop(Some(in), new PrintWriter(out))
+    Main.interp = interp
+    val separator = System.getProperty("path.separator")
+    val settings = new GenericRunnerSettings(s => throw new RuntimeException(s"Scala options error: $s"))
+    settings.processArguments(List("-classpath", paths.mkString(separator)), true)
+    interp.process(settings) // Repl starts and goes in loop of R.E.P.L
+    Main.interp = null
+    Option(Main.sparkContext).foreach(_.stop())
+    System.clearProperty("spark.driver.port")
+    out.toString
+  }
+
+}

spark-cassandra-connector/src/it/java/com/datastax/spark/connector/CassandraJavaUtilTest.java

@@ -5,6 +5,10 @@
 import java.util.Map;
 import java.util.Set;
 
+import com.datastax.spark.connector.japi.CassandraJavaUtil;
+import org.hamcrest.BaseMatcher;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
 import scala.reflect.api.TypeTags;
 
 import static com.datastax.spark.connector.japi.CassandraJavaUtil.*;
@@ -31,61 +35,71 @@
 @SuppressWarnings("unchecked")
 public class CassandraJavaUtilTest {
 
+    /**
+     * Scala refelection type tags change the string reprsentation of some types, in scala 2.11 java.lang
+     * is included, in scala 2.12 it is removed. To remove this conflict we just always remove the java.lang
+     * portion
+     */
+    private String removeJavaLang(String target) {
+        return target.replaceAll("java.lang.", "");
+    }
+
+    private final String STRING = removeJavaLang(String.class.getName());
+    private final String LIST_STRING =
+        removeJavaLang(String.format("%s[%s]", List.class.getName(), String.class.getName()));
+    private final String MAP_STRING_INT =
+        removeJavaLang(String.format("%s[%s,%s]", Map.class.getName(), String.class.getName(), Integer.class.getName()));
+    private final String LIST_SET_MAP_STRING_INT =
+        removeJavaLang(String.format("%s[%s[%s[%s,%s]]]", List.class.getName(), Set.class.getName(), Map.class.getName(), String.class.getName(), Integer.class.getName()));
+
     @Test
     public void testTypeTag1() throws Exception {
         TypeTags.TypeTag<String> tt = typeTag(String.class);
-        assertThat(tt.tpe().toString(), is(String.class.getName()));
+        assertThat(removeJavaLang(tt.tpe().toString()), is(STRING));
     }
 
     @Test
     public void testTypeTag2() throws Exception {
         TypeTags.TypeTag<List> tt1 = typeTag(List.class, String.class);
-        assertThat(tt1.tpe().toString(), is(String.format("%s[%s]",
-                List.class.getName(), String.class.getName())));
+        assertThat(removeJavaLang(removeJavaLang(tt1.tpe().toString())), is(LIST_STRING));
 
         TypeTags.TypeTag<Map> tt2 = typeTag(Map.class, String.class, Integer.class);
-        assertThat(tt2.tpe().toString(), is(String.format("%s[%s,%s]",
-                Map.class.getName(), String.class.getName(), Integer.class.getName())));
+        assertThat(removeJavaLang(removeJavaLang(tt2.tpe().toString())), is(MAP_STRING_INT));
     }
 
     @Test
     public void testTypeTag3() throws Exception {
         TypeTags.TypeTag<List> tt = typeTag(List.class, typeTag(Set.class, typeTag(Map.class, typeTag(String.class), typeTag(Integer.class))));
-        assertThat(tt.tpe().toString(), is(String.format("%s[%s[%s[%s,%s]]]",
-                List.class.getName(), Set.class.getName(), Map.class.getName(), String.class.getName(), Integer.class.getName())));
+        assertThat(removeJavaLang(tt.tpe().toString()), is(LIST_SET_MAP_STRING_INT));
     }
 
     @Test
     public void testTypeConverter1() throws Exception {
         TypeConverter<List<String>> tc = typeConverter(String.class);
-        assertThat(tc.targetTypeName(), is(String.class.getSimpleName()));
+        assertThat(removeJavaLang(tc.targetTypeName()), is(STRING));
     }
 
     @Test
     public void testTypeConverter2() throws Exception {
         TypeConverter<List<String>> tc1 = typeConverter(List.class, String.class);
-        assertThat(tc1.targetTypeName(), is(String.format("%s[%s]",
-                List.class.getName(), String.class.getSimpleName())));
+        assertThat(removeJavaLang(tc1.targetTypeName()), is(LIST_STRING));
 
         TypeConverter<Map<String, Integer>> tc2 = typeConverter(Map.class, String.class, Integer.class);
-        assertThat(tc2.targetTypeName(), is(String.format("%s[%s,%s]",
-                Map.class.getName(), String.class.getSimpleName(), Integer.class.getName())));
+        assertThat(removeJavaLang(tc2.targetTypeName()), is(MAP_STRING_INT));
 
     }
 
     @Test
     public void testTypeConverter3() throws Exception {
         TypeConverter<List> tc = typeConverter(List.class, typeTag(Set.class, typeTag(Map.class, typeTag(String.class), typeTag(Integer.class))));
-        assertThat(tc.targetTypeName(), is(String.format("%s[%s[%s[%s,%s]]]",
-                List.class.getName(), Set.class.getName(), Map.class.getName(), String.class.getSimpleName(), Integer.class.getName())));
+        assertThat(removeJavaLang(tc.targetTypeName()), is(LIST_SET_MAP_STRING_INT));
     }
 
     @Test
     public void testTypeConverter4() throws Exception {
         TypeTags.TypeTag<List> tt = typeTag(List.class, typeTag(Set.class, typeTag(Map.class, typeTag(String.class), typeTag(Integer.class))));
         TypeConverter<List> tc = typeConverter(tt);
-        assertThat(tc.targetTypeName(), is(String.format("%s[%s[%s[%s,%s]]]",
-                List.class.getName(), Set.class.getName(), Map.class.getName(), String.class.getSimpleName(), Integer.class.getName())));
+        assertThat(removeJavaLang(tc.targetTypeName()), is(LIST_SET_MAP_STRING_INT));
     }
 
     @Test

spark-cassandra-connector/src/main/scala/org/apache/spark/sql/cassandra/CassandraSourceRelation.scala

@@ -10,7 +10,7 @@ import com.datastax.spark.connector.rdd.partitioner.dht.TokenFactory.forSystemLo
 import com.datastax.spark.connector.types.{InetType, UUIDType, VarIntType}
 import com.datastax.spark.connector.util.Quote._
 import com.datastax.spark.connector.util.{ConfigParameter, Logging, ReflectionUtil}
-import com.datastax.spark.connector.writer.{SqlRowWriter, WriteConf}
+import com.datastax.spark.connector.writer.{RowWriterFactory, SqlRowWriter, WriteConf}
 import com.datastax.spark.connector.{SomeColumns, _}
 import org.apache.spark.SparkConf
 import org.apache.spark.rdd.RDD
@@ -43,6 +43,10 @@ private[cassandra] class CassandraSourceRelation(
   with PrunedFilteredScan
   with Logging {
 
+  implicit val readconf: ReadConf = readConf
+  implicit val rwf: RowWriterFactory[Row] = SqlRowWriter.Factory
+  implicit val cassandraConnector: CassandraConnector = connector
+
   private[this] val tableDef = Schema.tableFromCassandra(
     connector,
     tableRef.keyspace,
@@ -52,6 +56,7 @@ private[cassandra] class CassandraSourceRelation(
     userSpecifiedSchema.getOrElse(StructType(tableDef.columns.map(toStructField)))
   }
 
+
   override def insert(data: DataFrame, overwrite: Boolean): Unit = {
     if (overwrite) {
       if (confirmTruncate) {
@@ -71,18 +76,17 @@ private[cassandra] class CassandraSourceRelation(
 
     }
 
-    implicit val rwf = SqlRowWriter.Factory
     val columns = SomeColumns(data.columns.map(x => x: ColumnRef): _*)
     data.rdd.saveToCassandra(tableRef.keyspace, tableRef.table, columns, writeConf)
   }
 
+
+
   override def sizeInBytes: Long = {
     // If it's not found, use SQLConf default setting
     tableSizeInBytes.getOrElse(sqlContext.conf.defaultSizeInBytes)
   }
 
-  implicit val cassandraConnector = connector
-  implicit val readconf = readConf
   private[this] val baseRdd =
     sqlContext.sparkContext.cassandraTable[CassandraSQLRow](tableRef.keyspace, tableRef.table)