| // Copyright (C) 2019 GerritForge Ltd |
| // |
| // Licensed under the Apache License, Version 2.0 (the "License"); |
| // you may not use this file except in compliance with the License. |
| // You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, software |
| // distributed under the License is distributed on an "AS IS" BASIS, |
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| // See the License for the specific language governing permissions and |
| // limitations under the License. |
| |
| package com.gerritforge.analytics.infrastructure |
| import com.gerritforge.analytics.SparkTestSupport |
| import com.gerritforge.analytics.support.ops.ElasticsearchTestITSupport |
| import org.apache.spark.sql.Dataset |
| import org.scalatest.{BeforeAndAfterAll, FlatSpec, Matchers} |
| |
| import scala.concurrent.Await |
| import scala.concurrent.duration._ |
| |
| class ElasticSearchPimpedWriterIT |
| extends FlatSpec |
| with Matchers |
| with BeforeAndAfterAll |
| with SparkTestSupport |
| with ElasticsearchTestITSupport { |
| |
| override lazy val elasticSearchConnPort: Int = esHostHTTPExtPortMapping |
| import ESSparkWriterImplicits.withAliasSwap |
| import spark.implicits._ |
| |
| "Saving and reading from same ES alias" must "work while changing indices mapping" in { |
| |
| val aliasName = "the_alias" |
| val documentName = "doc" |
| |
| // Writing into the first index |
| val dataIntoIndexOne: Dataset[String] = "Content in the first index".split(" ").toList.toDS() |
| Await.result( |
| dataIntoIndexOne.saveToEsWithAliasSwap(aliasName, documentName).futureAction, |
| 2 seconds |
| ) |
| // Reading from the alias |
| val resultFromAliasFirst: Dataset[String] = |
| spark.read.format("es").load(s"$aliasName/$documentName").as[String] |
| |
| // Written should equal Read |
| dataIntoIndexOne |
| .collect() |
| .toList should contain only (resultFromAliasFirst.collect().toList: _*) |
| |
| // Writing into the second index |
| val dataIntoIndexTwo: Dataset[String] = "Content in the second index".split(" ").toList.toDS() |
| Await.result( |
| dataIntoIndexTwo.saveToEsWithAliasSwap(aliasName, documentName).futureAction, |
| 2 seconds |
| ) |
| // Reading from the alias |
| val resultFromAliasSecond: Dataset[String] = |
| spark.read.format("es").load(s"$aliasName/$documentName").as[String] |
| |
| // Written should equal Read |
| dataIntoIndexTwo |
| .collect() |
| .toList should contain only (resultFromAliasSecond.collect().toList: _*) |
| } |
| |
| "Saving data into ES" must "succeed even if alias creation fails" in { |
| import org.elasticsearch.spark.sql._ |
| val indexWithAliasName = "alias_name" |
| val documentName = "doc" |
| val indexWithAliasNameData = List("This", "index", "will", "make", "alias", "creation", "fail") |
| |
| indexWithAliasNameData.toDS.saveToEs(s"$indexWithAliasName/$documentName") |
| |
| val indexWithAliasData = List("An", "index", "with", "alias") |
| val aliasActionResponse: EnrichedAliasActionResponse = |
| indexWithAliasData.toDS.saveToEsWithAliasSwap(indexWithAliasName, documentName) |
| |
| val oldIndexData = |
| spark.read.format("es").load(s"$indexWithAliasName/$documentName").as[String] |
| val newIndexData = |
| spark.read.format("es").load(aliasActionResponse.path).as[String] |
| |
| assertThrows[NoSuchElementException] { |
| Await.result(aliasActionResponse.futureAction, 2 seconds) |
| } |
| newIndexData.collect().toList should contain only (indexWithAliasData: _*) |
| oldIndexData.collect().toList should contain only (indexWithAliasNameData: _*) |
| } |
| } |