spark 3.0.0 with hdp 3.2

Author: avnerl

.gitignore

@@ -20,3 +20,4 @@ metastore_db
 /spark/sql-13/with_meta_*
 out/
 localRepo/
+*.sha1

.travis.yml

@@ -1,12 +1,17 @@
 language: java
 jdk:
- - oraclejdk8
-
+ - openjdk12
 
 dist: trusty
 
+addons:
+ apt:
+  packages:
+   - oracle-java8-installer
+
 before_install:
  - "chmod +x gradlew"
  - export JAVA_OPTS="-Xmx2048m -XX:MaxPermSize=386m"
+ - export JAVA8_HOME=$(update-alternatives --list java | grep java-8-oracle | sed 's|/bin/java$||' | sed 's|/jre$||')
 
-install: ./gradlew assemble -Pskip.signing
+install: ./gradlew assemble -Pskip.signing updateSHAs

buildSrc/src/main/groovy/org/elasticsearch/hadoop/gradle/BaseBuildPlugin.groovy

@@ -124,6 +124,13 @@ class BaseBuildPlugin implements Plugin<Project> {
                     project.rootProject.ext.hadoopVersion = project.hadoop22Version
                     println "Using Apache Hadoop [$project.hadoop22Version]"
                     break
+            // Hadoop YARN/3.2.x
+                case "hadoopYarn3":
+                    String version = project.hadoop32Version
+                    project.rootProject.ext.hadoopVersion = version
+                    project.rootProject.ext.hadoopClient = ["org.apache.hadoop:hadoop-client:$version"]
+                    println "Using Apache Hadoop on YARN [$version]"
+                    break
                 default:
                     throw new GradleException("Invalid [hadoopDistro] setting: [$project.rootProject.ext.hadoopDistro]")
             }

dist/licenses/hadoop-client-2.7.6.jar.sha1

@@ -8,6 +8,13 @@ log4jVersion = 2.6.2
 # Hadoop versions
 hadoop2Version  = 2.7.6
 hadoop22Version = 2.2.0
+hadoop32Version = 3.2.0
+
+# `distro` property can be one of the following: [hadoopYarn, hadoopYarn3, hadoopStable]
+# default: hadoopYarn => build with hadoop2Version
+# hadoopStable => build with hadoop22Version
+# hadoopYarn3 => build with hadoop32Version
+# distro = hadoopYarn3
 
 # Common libraries
 hiveVersion = 1.2.1
@@ -19,11 +26,14 @@ jacksonVersion = 1.8.8
 # Spark
 spark13Version = 1.6.2
 spark20Version = 2.3.0
+spark30Version = 3.0.0
 # same as Spark's
 scala210Version = 2.10.7
 scala210MajorVersion = 2.10
 scala211Version = 2.11.12
 scala211MajorVersion = 2.11
+scala212Version = 2.12.10
+scala212MajorVersion = 2.12
 
 stormVersion = 1.0.6
 

dist/licenses/hadoop-common-2.7.6.jar.sha1

@@ -85,7 +85,7 @@ public void generateEventHiveRecordLimited() throws Exception {
         SerializationFailure iaeFailure = new SerializationFailure(new IllegalArgumentException("garbage"), tuple, new ArrayList<String>());
 
         String rawEvent = eventConverter.getRawEvent(iaeFailure);
-        assertThat(rawEvent, startsWith("HiveType{object=org.apache.hadoop.io.MapWritable@"));
+        assertTrue(rawEvent.matches("HiveType\\{object=\\{three=3, one=1, two=2\\}.*|^HiveType\\{object=org.apache.hadoop.io.MapWritable@.*"));
         String timestamp = eventConverter.getTimestamp(iaeFailure);
         assertTrue(StringUtils.hasText(timestamp));
         assertTrue(DateUtils.parseDate(timestamp).getTime().getTime() > 1L);

dist/licenses/hadoop-mapreduce-client-core-2.7.6.jar.sha1

@@ -52,13 +52,13 @@ public NodeInfo(String id, Map<String, Object> map) {
                 this.isData = true;
             } else {
                 String data = (String) attributes.get("data");
-                this.isClient = data == null ? true : !Boolean.parseBoolean(data);
-                this.isData = data == null ? true : Boolean.parseBoolean(data);
+                this.isClient = data == null || !Boolean.parseBoolean(data);
+                this.isData = data == null || Boolean.parseBoolean(data);
             }
             this.isIngest = false;
         } else {
             List<String> roles = (List<String>) map.get("roles");
-            this.isClient = roles.contains("data") == false;
+            this.isClient = !roles.contains("data");
             this.isData = roles.contains("data");
             this.isIngest = roles.contains("ingest");
         }

gradle.properties

@@ -69,7 +69,7 @@ public void generateEventWritable() throws Exception {
         SerializationFailure iaeFailure = new SerializationFailure(new IllegalArgumentException("garbage"), document, new ArrayList<String>());
 
         String rawEvent = eventConverter.getRawEvent(iaeFailure);
-        assertThat(rawEvent, Matchers.startsWith("org.apache.hadoop.io.MapWritable@"));
+        assertTrue(rawEvent.matches("\\{field=value\\}|^org.apache.hadoop.io.MapWritable@.*"));
         String timestamp = eventConverter.getTimestamp(iaeFailure);
         assertTrue(StringUtils.hasText(timestamp));
         assertTrue(DateUtils.parseDate(timestamp).getTime().getTime() > 1L);

hive/src/test/java/org/elasticsearch/hadoop/serialization/handler/write/impl/HiveSerializationEventConverterTest.java

@@ -15,6 +15,10 @@ include 'sql-20'
 project(":sql-20").projectDir = new File(settingsDir, "spark/sql-20")
 project(":sql-20").name = "elasticsearch-spark-20"
 
+include 'sql-30'
+project(":sql-30").projectDir = new File(settingsDir, "spark/sql-30")
+project(":sql-30").name = "elasticsearch-spark-30"
+
 include 'storm'
 project(":storm").name = "elasticsearch-storm"
 

licenses/hadoop-client-2.7.6.jar.sha1

@@ -18,8 +18,7 @@
  */
 package org.elasticsearch.spark.rdd;
 
-import scala.collection.JavaConversions.collectionAsScalaIterable
-import scala.collection.JavaConversions.mapAsJavaMap
+import scala.collection.JavaConverters._
 import scala.reflect.ClassTag
 import org.apache.commons.logging.LogFactory
 import org.apache.spark.Partition
@@ -45,7 +44,7 @@ private[spark] abstract class AbstractEsRDD[T: ClassTag](
   @transient protected lazy val logger = LogFactory.getLog(this.getClass())
 
   override def getPartitions: Array[Partition] = {
-    esPartitions.zipWithIndex.map { case(esPartition, idx) =>
+    esPartitions.asScala.zipWithIndex.map { case(esPartition, idx) =>
       new EsPartition(id, idx, esPartition)
     }.toArray
   }
@@ -70,7 +69,7 @@ private[spark] abstract class AbstractEsRDD[T: ClassTag](
 
   @transient private[spark] lazy val esCfg = {
     val cfg = new SparkSettingsManager().load(sc.getConf).copy();
-    cfg.merge(params)
+    cfg.merge(params.asJava)
     InitializationUtils.setUserProviderIfNotSet(cfg, classOf[HadoopUserProvider], logger)
     cfg
   }

licenses/hadoop-common-2.7.6.jar.sha1

@@ -21,6 +21,7 @@ package org.elasticsearch.spark.rdd
 import org.apache.commons.logging.Log
 import org.apache.commons.logging.LogFactory
 import org.apache.spark.TaskContext
+import org.apache.spark.util.TaskCompletionListener
 import org.elasticsearch.hadoop.cfg.PropertiesSettings
 import org.elasticsearch.hadoop.cfg.Settings
 import org.elasticsearch.hadoop.mr.security.HadoopUserProvider
@@ -64,7 +65,10 @@ private[spark] class EsRDDWriter[T: ClassTag](val serializedSettings: String,
   def write(taskContext: TaskContext, data: Iterator[T]): Unit = {
     val writer = RestService.createWriter(settings, taskContext.partitionId.toLong, -1, log)
 
-    taskContext.addTaskCompletionListener((TaskContext) => writer.close())
+    val taskCompletionListener = new TaskCompletionListener {
+      override def onTaskCompletion(context: TaskContext): Unit = writer.close()
+    }
+    taskContext.addTaskCompletionListener(taskCompletionListener)
 
     if (runtimeMetadata) {
       writer.repository.addRuntimeFieldExtractor(metaExtractor)