Apache Spark Streaming失敗的集成測試

Question

我一直在試圖追蹤一些我一直在爲Apache Spark項目編寫的單元/集成測試的問題。

當使用Spark 1.1.1時，我的測試通過了。當我嘗試升級到1.4.0（也嘗試1.4.1）時，測試開始失敗。

我設法減少了將問題重現到下面的小集成測試所需的代碼。

有趣的是，如果我在測試中註釋掉@RunWith註解，測試就會正確傳遞。很明顯，我不需要@RunWith註釋來進行這個減少測試，但真正的測試使用了相當廣泛的mock，所以我寧願不必使用PowerMock。

package com.example; 

import org.apache.spark.SparkConf; 
import org.apache.spark.streaming.Duration; 
import org.apache.spark.streaming.api.java.JavaStreamingContext; 
import org.junit.Before; 
import org.junit.Test; 
import org.junit.runner.RunWith; 
import org.powermock.modules.junit4.PowerMockRunner; 

@RunWith(PowerMockRunner.class) 
public class SampleTest { 

    @Before 
    public void setup() throws Exception { 
     SparkConf conf = new  SparkConf(false).setMaster("local[2]").setAppName("My app"); 
     JavaStreamingContext jsc = new JavaStreamingContext(conf, new Duration(1000)); 
    } 

    @Test 
    public void exampleTest() { 
    } 
} 

下面是我看到

java.io.IOException: failure to login 
    at org.apache.hadoop.security.UserGroupInformation.loginUserFromSubject(UserGroupInformation.java:796) 
    at org.apache.hadoop.security.UserGroupInformation.getLoginUser(UserGroupInformation.java:748) 
    at org.apache.hadoop.security.UserGroupInformation.getCurrentUser(UserGroupInformation.java:621) 
    at org.apache.spark.util.Utils$$anonfun$getCurrentUserName$1.apply(Utils.scala:2162) 
    at org.apache.spark.util.Utils$$anonfun$getCurrentUserName$1.apply(Utils.scala:2162) 
    at scala.Option.getOrElse(Option.scala:120) 
    at org.apache.spark.util.Utils$.getCurrentUserName(Utils.scala:2162) 
    at org.apache.spark.SparkContext.<init>(SparkContext.scala:301) 
    at org.apache.spark.streaming.StreamingContext$.createNewSparkContext(StreamingContext.scala:842) 
    at org.apache.spark.streaming.StreamingContext.<init>(StreamingContext.scala:80) 
    at org.apache.spark.streaming.api.java.JavaStreamingContext.<init>(JavaStreamingContext.scala:133) 
    at com.example.SampleTest.setup(SampleTest.java:19) 
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) 
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
    at java.lang.reflect.Method.invoke(Method.java:606) 
    at org.junit.internal.runners.MethodRoadie.runBefores(MethodRoadie.java:133) 
    at org.junit.internal.runners.MethodRoadie.runBeforesThenTestThenAfters(MethodRoadie.java:96) 
    at org.powermock.modules.junit4.internal.impl.PowerMockJUnit44RunnerDelegateImpl$PowerMockJUnit44MethodRunner.executeTest(PowerMockJUnit44RunnerDelegateImpl.java:294) 
    at org.powermock.modules.junit4.internal.impl.PowerMockJUnit47RunnerDelegateImpl$PowerMockJUnit47MethodRunner.executeTestInSuper(PowerMockJUnit47RunnerDelegateImpl.java:127) 
    at org.powermock.modules.junit4.internal.impl.PowerMockJUnit47RunnerDelegateImpl$PowerMockJUnit47MethodRunner.executeTest(PowerMockJUnit47RunnerDelegateImpl.java:82) 
    at org.powermock.modules.junit4.internal.impl.PowerMockJUnit44RunnerDelegateImpl$PowerMockJUnit44MethodRunner.runBeforesThenTestThenAfters(PowerMockJUnit44RunnerDelegateImpl.java:282) 
    at org.junit.internal.runners.MethodRoadie.runTest(MethodRoadie.java:87) 
    at org.junit.internal.runners.MethodRoadie.run(MethodRoadie.java:50) 
    at org.powermock.modules.junit4.internal.impl.PowerMockJUnit44RunnerDelegateImpl.invokeTestMethod(PowerMockJUnit44RunnerDelegateImpl.java:207) 
    at org.powermock.modules.junit4.internal.impl.PowerMockJUnit44RunnerDelegateImpl.runMethods(PowerMockJUnit44RunnerDelegateImpl.java:146) 
    at org.powermock.modules.junit4.internal.impl.PowerMockJUnit44RunnerDelegateImpl$1.run(PowerMockJUnit44RunnerDelegateImpl.java:120) 
    at org.junit.internal.runners.ClassRoadie.runUnprotected(ClassRoadie.java:34) 
    at org.junit.internal.runners.ClassRoadie.runProtected(ClassRoadie.java:44) 
    at org.powermock.modules.junit4.internal.impl.PowerMockJUnit44RunnerDelegateImpl.run(PowerMockJUnit44RunnerDelegateImpl.java:122) 
    at org.powermock.modules.junit4.common.internal.impl.JUnit4TestSuiteChunkerImpl.run(JUnit4TestSuiteChunkerImpl.java:106) 
    at org.powermock.modules.junit4.common.internal.impl.AbstractCommonPowerMockRunner.run(AbstractCommonPowerMockRunner.java:53) 
    at org.powermock.modules.junit4.PowerMockRunner.run(PowerMockRunner.java:59) 
    at org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run(JUnit4TestReference.java:50) 
    at org.eclipse.jdt.internal.junit.runner.TestExecution.run(TestExecution.java:38) 
    at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:467) 
    at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:683) 
    at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run(RemoteTestRunner.java:390) 
    at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main(RemoteTestRunner.java:197) 
Caused by: javax.security.auth.login.LoginException: Can't find user name 
    at org.apache.hadoop.security.UserGroupInformation$HadoopLoginModule.commit(UserGroupInformation.java:197) 
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) 
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
    at java.lang.reflect.Method.invoke(Method.java:606) 
    at javax.security.auth.login.LoginContext.invoke(LoginContext.java:784) 
    at javax.security.auth.login.LoginContext.access$000(LoginContext.java:203) 
    at javax.security.auth.login.LoginContext$5.run(LoginContext.java:721) 
    at javax.security.auth.login.LoginContext$5.run(LoginContext.java:719) 
    at java.security.AccessController.doPrivileged(Native Method) 
    at javax.security.auth.login.LoginContext.invokeCreatorPriv(LoginContext.java:718) 
    at javax.security.auth.login.LoginContext.login(LoginContext.java:591) 
    at org.apache.hadoop.security.UserGroupInformation.loginUserFromSubject(UserGroupInformation.java:771) 
    ... 38 more 

各種依賴關係的版本之外顯示如下

• Hadoop的客戶端2.6
• 阿帕奇火花1.4.0/1.4.1
• junit 4.12
• 容易模仿3.31
• 功率模擬1.6.2

我已經與星火的各種版本嘗試這個。上述測試用火花

以下版本傳遞

• 1.1.1
• 1.2.2

它開始從火花1.3.0起故障。

任何想法，我需要改變才能使這個工作？

Answer 1

查看SparkContext的源代碼，在嘗試獲取當前用戶名時導致異常的行。在version 1.2有一個備用的默認SparkContext.SPARK_UNKNOWN_USER，並沒有在用戶需要登錄currectly：

// Set SPARK_USER for user who is running SparkContext. 
val sparkUser = Option { 
    Option(System.getenv("SPARK_USER")).getOrElse(System.getProperty("user.name")) 
    }.getOrElse { 
     SparkContext.SPARK_UNKNOWN_USER 
    } 

這段代碼在version 1.3推出沒有默認的用戶了，因此你爲什麼不與早期版本出現此錯誤：

// Set SPARK_USER for user who is running SparkContext. 
val sparkUser = Utils.getCurrentUserName() 

這將調用following code在Utils：

/** 
    * Returns the current user name. This is the currently logged in user, unless that's been 
    * overridden by the `SPARK_USER` environment variable. 
    */ 
    def getCurrentUserName(): String = { 
    Option(System.getenv("SPARK_USER")) 
     .getOrElse(UserGroupInformation.getCurrentUser().getShortUserName()) 
    } 

如果您設置環境變量SPARK_USER，則應該能夠防止分支到導致您的異常的UserGroupInformation。

UserGroupInformation是一個Hadoop安全類，它看起來像使用PowerMock阻止它正常工作。