Merge pull request #282 from alexott/spark3

Update code to use Spark 3.0 & Scala 2.12

Author: fe2s

README.md

@@ -22,6 +22,7 @@ The master branch contains the recent development for the next release.
 
 | Spark-Redis                                                     | Spark         | Redis            | Supported Scala Versions | 
 | ----------------------------------------------------------------| ------------- | ---------------- | ------------------------ |
+| [master](https://github.com/RedisLabs/spark-redis/)             | 3.0.x         | >=2.9.0          | 2.12                     | 
 | [2.4, 2.5, 2.6](https://github.com/RedisLabs/spark-redis/tree/branch-2.4) | 2.4.x         | >=2.9.0          | 2.11, 2.12               | 
 | [2.3](https://github.com/RedisLabs/spark-redis/tree/branch-2.3) | 2.3.x         | >=2.9.0          | 2.11                     | 
 | [1.4](https://github.com/RedisLabs/spark-redis/tree/branch-1.4) | 1.4.x         |                  | 2.10                     | 

dev/change-scala-version.sh

@@ -2,10 +2,9 @@
 
 set -e
 
-VALID_VERSIONS=( 2.11 2.12 )
+VALID_VERSIONS=( 2.12 )
 
-SCALA_211_MINOR_VERSION="12"
-SCALA_212_MINOR_VERSION="9"
+SCALA_212_MINOR_VERSION="12"
 
 usage() {
   echo "Usage: $(basename $0) [-h|--help] <version>

doc/dev.md

@@ -26,11 +26,3 @@ To build Spark-Redis skipping tests, run:
 mvn clean package -DskipTests
 ```
 
-To change scala version use `./dev/change-scala-version.sh` script. It will change scala version in `pom.xml`. For example:
-```
-./dev/change-scala-version.sh 2.12
-```
-
-```
-./dev/change-scala-version.sh 2.11
-```

pom.xml

@@ -2,8 +2,8 @@
 	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>com.redislabs</groupId>
-	<artifactId>spark-redis_2.11</artifactId>
-	<version>2.6.0-SNAPSHOT</version>
+	<artifactId>spark-redis_2.12</artifactId>
+	<version>3.0.0-SNAPSHOT</version>
 	<name>Spark-Redis</name>
 	<description>A Spark library for Redis</description>
 	<url>http://github.com/RedisLabs/spark-redis</url>
@@ -47,10 +47,10 @@
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 		<java.version>1.8</java.version>
-		<scala.major.version>2.11</scala.major.version>
+		<scala.major.version>2.12</scala.major.version>
 		<scala.complete.version>${scala.major.version}.12</scala.complete.version>
-		<jedis.version>3.2.0</jedis.version>
-		<spark.version>2.4.1</spark.version>
+		<jedis.version>3.4.1</jedis.version>
+		<spark.version>3.0.1</spark.version>
 		<plugins.scalatest.version>1.0</plugins.scalatest.version>
 	</properties>
 
@@ -89,7 +89,7 @@
 			<plugin>
 				<groupId>net.alchim31.maven</groupId>
 				<artifactId>scala-maven-plugin</artifactId>
-				<version>3.2.2</version>
+				<version>3.4.6</version>
 				<configuration>
 					<scalaVersion>${scala.complete.version}</scalaVersion>
 					<javacArgs>
@@ -287,16 +287,19 @@
             <groupId>org.scala-lang</groupId>
             <artifactId>scala-library</artifactId>
             <version>${scala.complete.version}</version>
+	    <scope>provided</scope>
         </dependency>
 		<dependency>
 			<groupId>org.scala-lang</groupId>
 			<artifactId>scalap</artifactId>
 			<version>${scala.complete.version}</version>
+			<scope>provided</scope>
 		</dependency>
 		<dependency>
 			<groupId>org.scala-lang</groupId>
 			<artifactId>scala-compiler</artifactId>
 			<version>${scala.complete.version}</version>
+			<scope>provided</scope>
 		</dependency>
 		<dependency>
             <groupId>org.apache.spark</groupId>

src/main/scala/com/redislabs/provider/redis/streaming/RedisStreamReceiver.scala

@@ -1,15 +1,14 @@
 package com.redislabs.provider.redis.streaming
 
 import java.util.AbstractMap.SimpleEntry
-
 import com.redislabs.provider.redis.util.PipelineUtils.foreachWithPipeline
 import com.redislabs.provider.redis.util.{Logging, StreamUtils}
 import com.redislabs.provider.redis.{ReadWriteConfig, RedisConfig}
 import org.apache.curator.utils.ThreadUtils
 import org.apache.spark.storage.StorageLevel
 import org.apache.spark.streaming.receiver.Receiver
-import org.spark_project.guava.util.concurrent.RateLimiter
-import redis.clients.jedis.{StreamEntryID, Jedis, StreamEntry}
+import org.sparkproject.guava.util.concurrent.RateLimiter
+import redis.clients.jedis.{Jedis, StreamEntry, StreamEntryID}
 
 import scala.collection.JavaConversions._
 

src/main/scala/org/apache/spark/sql/redis/stream/RedisSource.scala

@@ -4,7 +4,7 @@ import com.redislabs.provider.redis.RedisConfig
 import com.redislabs.provider.redis.util.CollectionUtils.RichCollection
 import com.redislabs.provider.redis.util.ConnectionUtils.{JedisExt, XINFO, withConnection}
 import com.redislabs.provider.redis.util.StreamUtils.{createConsumerGroupIfNotExist, resetConsumerGroup}
-import com.redislabs.provider.redis.util.{ConnectionUtils, Logging, ParseUtils}
+import com.redislabs.provider.redis.util.{Logging, ParseUtils}
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.execution.streaming.{Offset, Source}
 import org.apache.spark.sql.redis.stream.RedisSource._

src/test/scala/com/redislabs/provider/redis/env/RedisStandaloneEnv.scala

@@ -14,6 +14,7 @@ trait RedisStandaloneEnv extends Env {
     .set("spark.redis.port", s"$redisPort")
     .set("spark.redis.auth", redisAuth)
     .set("spark.streaming.stopGracefullyOnShutdown", "true")
+    .set("spark.sql.streaming.forceDeleteTempCheckpointLocation", "true")
     .set("spark.driver.bindAddress", "127.0.0.1")
 
   override val redisConfig: RedisConfig =

src/test/scala/org/apache/spark/sql/redis/stream/RedisStreamSourceSuite.scala

@@ -228,7 +228,7 @@ trait RedisStreamSourceSuite extends FunSuite with Matchers with Env with Loggin
 
     val (spark, query) = readStream2(streamKey, extraReadOptions, extraWriteOptions, writeFormat)
     // give some time for spark query to start
-    Thread.sleep(50)
+    Thread.sleep(2000)
     try {
       body(spark)
     } finally {