Fixes issue caused by http in history server config property (linkedi…

…n#217)

app/com/linkedin/drelephant/spark/fetchers/SparkRestClient.scala

@@ -51,7 +51,14 @@ class SparkRestClient(sparkConf: SparkConf) {
 
  private val historyServerUri: URI = sparkConf.getOption(HISTORY_SERVER_ADDRESS_KEY) match {
  case Some(historyServerAddress) =>
- val baseUri = new URI(s"http://${historyServerAddress}")
+ val baseUri: URI =
+ // Latest versions of CDH include http in their history server address configuration.
+ // However, it is not recommended by Spark documentation(http://spark.apache.org/docs/latest/running-on-yarn.html)
+ if (historyServerAddress.contains(s"http://")) {
+ new URI(historyServerAddress)
+ } else {
+ new URI(s"http://${historyServerAddress}")
+ }
  require(baseUri.getPath == "")
  baseUri
  case None =>

test/com/linkedin/drelephant/spark/fetchers/SparkRestClientTest.scala

@@ -112,6 +112,40 @@ class SparkRestClientTest extends AsyncFunSpec with Matchers {
  assertion
  }
  }
+
+ it("returns the desired data from the Spark REST API for cluster mode application when http in jobhistory address") {
+ import ExecutionContext.Implicits.global
+ val fakeJerseyServer = new FakeJerseyServer() {
+ override def configure(): Application = super.configure() match {
+ case resourceConfig: ResourceConfig =>
+ resourceConfig
+ .register(classOf[FetchClusterModeDataFixtures.ApiResource])
+ .register(classOf[FetchClusterModeDataFixtures.ApplicationResource])
+ .register(classOf[FetchClusterModeDataFixtures.JobsResource])
+ .register(classOf[FetchClusterModeDataFixtures.StagesResource])
+ .register(classOf[FetchClusterModeDataFixtures.ExecutorsResource])
+ case config => config
+ }
+ }
+
+ fakeJerseyServer.setUp()
+
+ val historyServerUri = fakeJerseyServer.target.getUri
+
+ val sparkConf = new SparkConf().set("spark.yarn.historyServer.address", s"http://${historyServerUri.getHost}:${historyServerUri.getPort}")
+ val sparkRestClient = new SparkRestClient(sparkConf)
+
+ sparkRestClient.fetchData(FetchClusterModeDataFixtures.APP_ID) map { restDerivedData =>
+ restDerivedData.applicationInfo.id should be(FetchClusterModeDataFixtures.APP_ID)
+ restDerivedData.applicationInfo.name should be(FetchClusterModeDataFixtures.APP_NAME)
+ restDerivedData.jobDatas should not be(None)
+ restDerivedData.stageDatas should not be(None)
+ restDerivedData.executorSummaries should not be(None)
+ } andThen { case assertion: Try[Assertion] =>
+ fakeJerseyServer.tearDown()
+ assertion
+ }
+ }
  }
 }