From e9ef3baef3dc49d04f49074db124a1ff0d2438ca Mon Sep 17 00:00:00 2001 From: angerszhu Date: Tue, 19 Nov 2019 15:20:47 +0800 Subject: [PATCH 01/13] Update pom.xml --- pom.xml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 44593b78c9a0..c4e16b76c76d 100644 --- a/pom.xml +++ b/pom.xml @@ -123,6 +123,7 @@ 1.7.16 1.2.17 2.7.4 + 3.2.0 2.5.0 ${hadoop.version} 3.4.14 @@ -1024,7 +1025,7 @@ org.apache.hadoop hadoop-minikdc - ${hadoop.version} + ${miniKdc.version} test From 3ec289102c412c16a0f429caf27b2607e8540ac6 Mon Sep 17 00:00:00 2001 From: angerszhu Date: Sun, 24 Nov 2019 09:20:38 +0800 Subject: [PATCH 02/13] miniKdc -> minikdc --- pom.xml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index c4e16b76c76d..fb5db619166c 100644 --- a/pom.xml +++ b/pom.xml @@ -123,7 +123,7 @@ 1.7.16 1.2.17 2.7.4 - 3.2.0 + 3.2.0 2.5.0 ${hadoop.version} 3.4.14 @@ -1025,7 +1025,7 @@ org.apache.hadoop hadoop-minikdc - ${miniKdc.version} + ${minikdc.version} test From 2e282f264c6eceaf60e1cf36115d0b6c021faf22 Mon Sep 17 00:00:00 2001 From: angerszhu Date: Sun, 24 Nov 2019 16:29:19 +0800 Subject: [PATCH 03/13] Update KafkaTestUtils.scala --- .../scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala index 6c745987b4c2..99674d2b3c74 100644 --- a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala +++ b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala @@ -46,6 +46,7 @@ import org.apache.kafka.common.header.internals.RecordHeader import org.apache.kafka.common.network.ListenerName import org.apache.kafka.common.security.auth.SecurityProtocol.{PLAINTEXT, SASL_PLAINTEXT} import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer} +import org.apache.kerby.kerberos.kerb.admin.Krb5Conf import org.apache.zookeeper.server.{NIOServerCnxnFactory, ZooKeeperServer} import org.apache.zookeeper.server.auth.SASLAuthenticationProvider import org.scalatest.concurrent.Eventually._ @@ -173,6 +174,7 @@ class KafkaTestUtils( | useKeyTab=true | storeKey=true | useTicketCache=false + | refreshKrb5Config=true | keyTab="${zkServerKeytabFile.getAbsolutePath()}" | principal="$zkServerUser@$realm"; |}; @@ -182,6 +184,7 @@ class KafkaTestUtils( | useKeyTab=true | storeKey=true | useTicketCache=false + | refreshKrb5Config=true | keyTab="${zkClientKeytabFile.getAbsolutePath()}" | principal="$zkClientUser@$realm"; |}; @@ -243,6 +246,7 @@ class KafkaTestUtils( if (secure) { setupKrbDebug() setUpMiniKdc() + System.setProperty(Krb5Conf.KRB5_CONF, kdc.getKrb5conf.getAbsolutePath) val jaasConfigFile = createKeytabsAndJaasConfigFile() System.setProperty(JAVA_AUTH_CONFIG, jaasConfigFile) Configuration.getConfiguration.refresh() From a224fad9e9d38b7238ca192beadb6b798d1dd5c9 Mon Sep 17 00:00:00 2001 From: angerszhu Date: Sun, 24 Nov 2019 16:35:34 +0800 Subject: [PATCH 04/13] Update KafkaTestUtils.scala --- .../scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala | 2 -- 1 file changed, 2 deletions(-) diff --git a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala index 99674d2b3c74..95f41c6f0aea 100644 --- a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala +++ b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala @@ -46,7 +46,6 @@ import org.apache.kafka.common.header.internals.RecordHeader import org.apache.kafka.common.network.ListenerName import org.apache.kafka.common.security.auth.SecurityProtocol.{PLAINTEXT, SASL_PLAINTEXT} import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer} -import org.apache.kerby.kerberos.kerb.admin.Krb5Conf import org.apache.zookeeper.server.{NIOServerCnxnFactory, ZooKeeperServer} import org.apache.zookeeper.server.auth.SASLAuthenticationProvider import org.scalatest.concurrent.Eventually._ @@ -246,7 +245,6 @@ class KafkaTestUtils( if (secure) { setupKrbDebug() setUpMiniKdc() - System.setProperty(Krb5Conf.KRB5_CONF, kdc.getKrb5conf.getAbsolutePath) val jaasConfigFile = createKeytabsAndJaasConfigFile() System.setProperty(JAVA_AUTH_CONFIG, jaasConfigFile) Configuration.getConfiguration.refresh() From 3456ff9c718619d6ca3a3af736ffd3f9603e1eee Mon Sep 17 00:00:00 2001 From: angerszhu Date: Tue, 26 Nov 2019 13:22:33 +0800 Subject: [PATCH 05/13] Follow comment --- .../spark/sql/kafka010/KafkaTestUtils.scala | 17 ++++++++++++++++- pom.xml | 3 +-- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala index 95f41c6f0aea..81125eaf9f3b 100644 --- a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala +++ b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala @@ -17,7 +17,7 @@ package org.apache.spark.sql.kafka010 -import java.io.{File, IOException} +import java.io.{File, IOException, PrintWriter} import java.lang.{Integer => JInt} import java.net.{InetAddress, InetSocketAddress} import java.nio.charset.StandardCharsets @@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit import javax.security.auth.login.Configuration import scala.collection.JavaConverters._ +import scala.io.Source import scala.util.Random import com.google.common.io.Files @@ -138,6 +139,20 @@ class KafkaTestUtils( kdcConf.setProperty(MiniKdc.DEBUG, "true") kdc = new MiniKdc(kdcConf, kdcDir) kdc.start() + val krb5Conf = Source.fromFile(kdc.getKrb5conf, "UTF-8").getLines() + val rewriteKrb5Conf = krb5Conf.map(s => if (s.contains("libdefaults")) { + s + "\n" + + " default_tkt_enctypes=aes128-cts-hmac-sha1-96\n" + + " default_tgs_enctypes=aes128-cts-hmac-sha1-96 " + } else { + s + }) + kdc.getKrb5conf.delete() + val writer = new PrintWriter(kdc.getKrb5conf) + // scalastyle:off + rewriteKrb5Conf.foreach(writer.println) + writer.flush() + writer.close() kdcReady = true } diff --git a/pom.xml b/pom.xml index fb5db619166c..44593b78c9a0 100644 --- a/pom.xml +++ b/pom.xml @@ -123,7 +123,6 @@ 1.7.16 1.2.17 2.7.4 - 3.2.0 2.5.0 ${hadoop.version} 3.4.14 @@ -1025,7 +1024,7 @@ org.apache.hadoop hadoop-minikdc - ${minikdc.version} + ${hadoop.version} test From 7ab61923880fbe1483559c92dea1f8e2ef99d341 Mon Sep 17 00:00:00 2001 From: angerszhu Date: Thu, 28 Nov 2019 18:57:34 +0800 Subject: [PATCH 06/13] Update KafkaTestUtils.scala --- .../spark/sql/kafka010/KafkaTestUtils.scala | 28 +++++++++++++------ 1 file changed, 19 insertions(+), 9 deletions(-) diff --git a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala index 030f2c6f632c..e64223295305 100644 --- a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala +++ b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala @@ -137,21 +137,31 @@ class KafkaTestUtils( kdcConf.setProperty(MiniKdc.DEBUG, "true") kdc = new MiniKdc(kdcConf, kdcDir) kdc.start() + rewriteKrb5Conf() + kdcReady = true + } + + private def rewriteKrb5Conf(): Unit = { val krb5Conf = Source.fromFile(kdc.getKrb5conf, "UTF-8").getLines() - val rewriteKrb5Conf = krb5Conf.map(s => if (s.contains("libdefaults")) { - s + "\n" + - " default_tkt_enctypes=aes128-cts-hmac-sha1-96\n" + - " default_tgs_enctypes=aes128-cts-hmac-sha1-96 " - } else { - s - }) + val rewriteKrb5Conf = krb5Conf.map(s => + if (s.contains("libdefaults")) { + val addedConfig = + addedKrb5Config("default_tkt_enctypes", "aes128-cts-hmac-sha1-96") + + addedKrb5Config("default_tgs_enctypes", "aes128-cts-hmac-sha1-96") + s + addedConfig + } else { + s + }) kdc.getKrb5conf.delete() val writer = new PrintWriter(kdc.getKrb5conf) // scalastyle:off rewriteKrb5Conf.foreach(writer.println) - writer.flush() + // scalastyle:on writer.close() - kdcReady = true + } + + private def addedKrb5Config(key: String, value: String): String = { + System.lineSeparator() + s" $key=$value" } private def createKeytabsAndJaasConfigFile(): String = { From 476d60e469c213951e4a2ce16572328b62cb85ba Mon Sep 17 00:00:00 2001 From: angerszhu Date: Fri, 29 Nov 2019 18:35:51 +0800 Subject: [PATCH 07/13] Update KafkaTestUtils.scala --- .../scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala index e64223295305..4ffe16204613 100644 --- a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala +++ b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala @@ -154,9 +154,9 @@ class KafkaTestUtils( }) kdc.getKrb5conf.delete() val writer = new PrintWriter(kdc.getKrb5conf) - // scalastyle:off + // scalastyle:off println rewriteKrb5Conf.foreach(writer.println) - // scalastyle:on + // scalastyle:on println writer.close() } From b782e997b893fb4791d08058fc8c24301940ff25 Mon Sep 17 00:00:00 2001 From: angerszhu Date: Wed, 4 Dec 2019 16:59:59 +0800 Subject: [PATCH 08/13] Update KafkaTestUtils.scala --- .../apache/spark/sql/kafka010/KafkaTestUtils.scala | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala index 4ffe16204613..cbbc944b8c86 100644 --- a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala +++ b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala @@ -143,15 +143,20 @@ class KafkaTestUtils( private def rewriteKrb5Conf(): Unit = { val krb5Conf = Source.fromFile(kdc.getKrb5conf, "UTF-8").getLines() - val rewriteKrb5Conf = krb5Conf.map(s => + var rewritten = false + val addedConfig = + addedKrb5Config("default_tkt_enctypes", "aes128-cts-hmac-sha1-96") + + addedKrb5Config("default_tgs_enctypes", "aes128-cts-hmac-sha1-96") + var rewriteKrb5Conf = krb5Conf.map(s => if (s.contains("libdefaults")) { - val addedConfig = - addedKrb5Config("default_tkt_enctypes", "aes128-cts-hmac-sha1-96") + - addedKrb5Config("default_tgs_enctypes", "aes128-cts-hmac-sha1-96") + rewritten = true s + addedConfig } else { s }) + if (!rewritten) { + rewriteKrb5Conf = rewriteKrb5Conf ++ Array("[libdefaults]" + addedConfig) + } kdc.getKrb5conf.delete() val writer = new PrintWriter(kdc.getKrb5conf) // scalastyle:off println From 10a000668a33eb6234b4ac3621992f179319a30b Mon Sep 17 00:00:00 2001 From: angerszhu Date: Wed, 4 Dec 2019 17:06:01 +0800 Subject: [PATCH 09/13] Update KafkaTestUtils.scala --- .../org/apache/spark/sql/kafka010/KafkaTestUtils.scala | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala index cbbc944b8c86..c43ce51fb1cd 100644 --- a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala +++ b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala @@ -137,10 +137,16 @@ class KafkaTestUtils( kdcConf.setProperty(MiniKdc.DEBUG, "true") kdc = new MiniKdc(kdcConf, kdcDir) kdc.start() + // TODO https://issues.apache.org/jira/browse/SPARK-30037 + // Need to customize krb5.conf like Kafka rewriteKrb5Conf() kdcReady = true } + /** + * In this method we rewrite krb5.conf to make kdc nad client use the same enctypes + * + */ private def rewriteKrb5Conf(): Unit = { val krb5Conf = Source.fromFile(kdc.getKrb5conf, "UTF-8").getLines() var rewritten = false From 03f7fb1c117da3852c379c20a7eaabb0adeeb5ca Mon Sep 17 00:00:00 2001 From: angerszhu Date: Wed, 4 Dec 2019 17:33:48 +0800 Subject: [PATCH 10/13] Update KafkaTestUtils.scala --- .../spark/sql/kafka010/KafkaTestUtils.scala | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala index c43ce51fb1cd..4d928bd0afd2 100644 --- a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala +++ b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala @@ -153,20 +153,25 @@ class KafkaTestUtils( val addedConfig = addedKrb5Config("default_tkt_enctypes", "aes128-cts-hmac-sha1-96") + addedKrb5Config("default_tgs_enctypes", "aes128-cts-hmac-sha1-96") - var rewriteKrb5Conf = krb5Conf.map(s => + val rewriteKrb5Conf = krb5Conf.map(s => if (s.contains("libdefaults")) { rewritten = true s + addedConfig } else { s - }) - if (!rewritten) { - rewriteKrb5Conf = rewriteKrb5Conf ++ Array("[libdefaults]" + addedConfig) + }).filter(!_.trim.startsWith("#")).mkString(System.lineSeparator()) + + val krb5confStr = if (!rewritten) { + "[libdefaults]" + addedConfig + System.lineSeparator() + + System.lineSeparator() + rewriteKrb5Conf + } else { + rewriteKrb5Conf } + kdc.getKrb5conf.delete() val writer = new PrintWriter(kdc.getKrb5conf) // scalastyle:off println - rewriteKrb5Conf.foreach(writer.println) + writer.println(krb5confStr) // scalastyle:on println writer.close() } From e252dd0ff4b8b327db531810168caa66a7a4584f Mon Sep 17 00:00:00 2001 From: angerszhu Date: Thu, 5 Dec 2019 09:38:04 +0800 Subject: [PATCH 11/13] Update KafkaTestUtils.scala --- .../scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala index 4d928bd0afd2..f124850bc6ce 100644 --- a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala +++ b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala @@ -138,14 +138,13 @@ class KafkaTestUtils( kdc = new MiniKdc(kdcConf, kdcDir) kdc.start() // TODO https://issues.apache.org/jira/browse/SPARK-30037 - // Need to customize krb5.conf like Kafka + // Need to build spark's own MiniKDC and customize krb5.conf like Kafka rewriteKrb5Conf() kdcReady = true } /** - * In this method we rewrite krb5.conf to make kdc nad client use the same enctypes - * + * In this method we rewrite krb5.conf to make kdc and client use the same enctypes */ private def rewriteKrb5Conf(): Unit = { val krb5Conf = Source.fromFile(kdc.getKrb5conf, "UTF-8").getLines() From 4facbeeb374da4098064c9bf68e10931a5922fbd Mon Sep 17 00:00:00 2001 From: angerszhu Date: Fri, 6 Dec 2019 12:26:55 +0800 Subject: [PATCH 12/13] Update KafkaTestUtils.scala --- .../org/apache/spark/sql/kafka010/KafkaTestUtils.scala | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala index f124850bc6ce..a333379020da 100644 --- a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala +++ b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala @@ -168,11 +168,7 @@ class KafkaTestUtils( } kdc.getKrb5conf.delete() - val writer = new PrintWriter(kdc.getKrb5conf) - // scalastyle:off println - writer.println(krb5confStr) - // scalastyle:on println - writer.close() + Files.write(krb5confStr, kdc.getKrb5conf, StandardCharsets.UTF_8) } private def addedKrb5Config(key: String, value: String): String = { From 223533fa13551b54f3ac10b666babb379f8601a9 Mon Sep 17 00:00:00 2001 From: angerszhu Date: Fri, 6 Dec 2019 12:28:04 +0800 Subject: [PATCH 13/13] Update KafkaTestUtils.scala --- .../scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala index a333379020da..47ae7be85ce0 100644 --- a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala +++ b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala @@ -17,7 +17,7 @@ package org.apache.spark.sql.kafka010 -import java.io.{File, IOException, PrintWriter} +import java.io.{File, IOException} import java.net.{InetAddress, InetSocketAddress} import java.nio.charset.StandardCharsets import java.util.{Collections, Properties, UUID}