Merge pull request #787 from aimansharief/KN-976

Issue #KN-976 fix: Elasticsearch version upgrade from 6.8.22 to 7.17.13

.circleci/config.yml

@@ -2,8 +2,8 @@ version: 2.0
 
 jobs:
   unit-tests:
-    docker:
-      - image: circleci/openjdk:14-jdk-buster-node-browsers-legacy
+    machine:
+      image: ubuntu-2004:202201-02
     resource_class: medium
     working_directory: ~/kp
     steps:

jobs-core/pom.xml

@@ -157,7 +157,7 @@
         <dependency>
             <groupId>org.elasticsearch.client</groupId>
             <artifactId>elasticsearch-rest-high-level-client</artifactId>
-            <version>6.8.22</version>
+            <version>7.17.13</version>
         </dependency>
         <dependency>
             <groupId>com.twitter</groupId>

jobs-core/src/main/scala/org/sunbird/job/util/ElasticSearchUtil.scala

@@ -8,20 +8,19 @@ import org.apache.commons.lang3.StringUtils
 import org.apache.http.HttpHost
 import org.apache.http.client.config.RequestConfig
 import org.elasticsearch.action.admin.indices.alias.Alias
-import org.elasticsearch.action.admin.indices.create.CreateIndexRequest
 import org.elasticsearch.action.bulk.BulkRequest
 import org.elasticsearch.action.delete.DeleteRequest
 import org.elasticsearch.action.get.GetRequest
 import org.elasticsearch.action.index.IndexRequest
 import org.elasticsearch.action.update.UpdateRequest
-import org.elasticsearch.client.{Response, RestClient, RestClientBuilder, RestHighLevelClient}
+import org.elasticsearch.client.indices.CreateIndexRequest
+import org.elasticsearch.client.{Request, RequestOptions, Response, RestClient, RestClientBuilder, RestHighLevelClient}
 import org.elasticsearch.common.settings.Settings
-import org.elasticsearch.common.xcontent.XContentType
+import org.elasticsearch.xcontent.XContentType
 import org.slf4j.LoggerFactory
-
 import scala.collection.convert.ImplicitConversions.`collection AsScalaIterable`
 
-class ElasticSearchUtil(connectionInfo: String, indexName: String, indexType: String, batchSize: Int = 1000) extends Serializable {
+class ElasticSearchUtil(connectionInfo: String, indexName: String, batchSize: Int = 1000) extends Serializable {
 
   private val resultLimit = 100
   private val esClient: RestHighLevelClient = createClient(connectionInfo)
@@ -49,7 +48,7 @@ class ElasticSearchUtil(connectionInfo: String, indexName: String, indexType: St
 
   def isIndexExists(): Boolean = {
     try {
-      val response = esClient.getLowLevelClient.performRequest("HEAD", "/" + indexName)
+      val response = esClient.getLowLevelClient.performRequest(new Request("HEAD", "/" + indexName))
       response.getStatusLine.getStatusCode == 200
     } catch {
       case e: IOException => {
@@ -59,27 +58,27 @@ class ElasticSearchUtil(connectionInfo: String, indexName: String, indexType: St
     }
   }
 
-  def addIndex(settings: String, mappings: String, alias: String = ""): Boolean = {
-    var response = false
-    val client = esClient
-    if (!isIndexExists()) {
-      val createRequest = new CreateIndexRequest(indexName)
-      if (StringUtils.isNotBlank(alias)) createRequest.alias(new Alias(alias))
-      if (StringUtils.isNotBlank(settings)) createRequest.settings(Settings.builder.loadFromSource(settings, XContentType.JSON))
-      if (StringUtils.isNotBlank(indexType) && StringUtils.isNotBlank(mappings)) createRequest.mapping(indexType, mappings, XContentType.JSON)
-      val createIndexResponse = client.indices.create(createRequest)
-      response = createIndexResponse.isAcknowledged
+    def addIndex(settings: String, mappings: String, alias: String = ""): Boolean = {
+      var response = false
+      val client = esClient
+      if (!isIndexExists()) {
+        val createRequest = new CreateIndexRequest(indexName)
+        if (StringUtils.isNotBlank(alias)) createRequest.alias(new Alias(alias))
+        if (StringUtils.isNotBlank(settings)) createRequest.settings(Settings.builder.loadFromSource(settings, XContentType.JSON))
+        if (StringUtils.isNotBlank(mappings)) createRequest.mapping(mappings, XContentType.JSON)
+        val createIndexResponse = client.indices.create(createRequest, RequestOptions.DEFAULT)
+        response = createIndexResponse.isAcknowledged
+      }
+      response
     }
-    response
-  }
 
   def addDocument(identifier: String, document: String): Unit = {
     try {
       // TODO
       // Replace mapper with JSONUtil once the JSONUtil is fixed
       val doc = mapper.readValue(document, new TypeReference[util.Map[String, AnyRef]]() {})
       val updatedDoc = checkDocStringLength(doc)
-      val response = esClient.index(new IndexRequest(indexName, indexType, identifier).source(updatedDoc))
+      val response = esClient.index(new IndexRequest(indexName).id(identifier).source(updatedDoc), RequestOptions.DEFAULT)
       logger.info(s"Added ${response.getId} to index ${response.getIndex}")
     } catch {
       case e: IOException =>
@@ -94,14 +93,12 @@ class ElasticSearchUtil(connectionInfo: String, indexName: String, indexType: St
       // Replace mapper with JSONUtil once the JSONUtil is fixed
       val doc = mapper.readValue(document, new TypeReference[util.Map[String, AnyRef]]() {})
       val updatedDoc = checkDocStringLength(doc)
-      val indexRequest = if(identifier == null) new IndexRequest(indexName, indexType) else new IndexRequest(indexName, indexType, identifier)
-      val response = esClient.index(indexRequest.source(updatedDoc))
+      val indexRequest = if(identifier == null) new IndexRequest(indexName) else new IndexRequest(indexName).id(identifier)
+      val response = esClient.index(indexRequest.source(updatedDoc), RequestOptions.DEFAULT)
       logger.info(s"Added ${response.getId} to index ${response.getIndex}")
     } catch {
       case e: IOException =>
         logger.error(s"ElasticSearchUtil:: Error while adding document to index : $indexName : " + e.getMessage)
-        e.printStackTrace()
-        throw e
     }
   }
 
@@ -111,9 +108,9 @@ class ElasticSearchUtil(connectionInfo: String, indexName: String, indexType: St
       // Replace mapper with JSONUtil once the JSONUtil is fixed
       val doc = mapper.readValue(document, new TypeReference[util.Map[String, AnyRef]]() {})
       val updatedDoc = checkDocStringLength(doc)
-      val indexRequest = new IndexRequest(indexName, indexType, identifier).source(updatedDoc)
-      val request = new UpdateRequest().index(indexName).`type`(indexType).id(identifier).doc(updatedDoc).upsert(indexRequest)
-      val response = esClient.update(request)
+      val indexRequest = new IndexRequest(indexName).id(identifier).source(updatedDoc)
+      val request = new UpdateRequest().index(indexName).id(identifier).doc(updatedDoc).upsert(indexRequest)
+      val response = esClient.update(request, RequestOptions.DEFAULT)
       logger.info(s"Updated ${response.getId} to index ${response.getIndex}")
     } catch {
       case e: IOException =>
@@ -122,12 +119,12 @@ class ElasticSearchUtil(connectionInfo: String, indexName: String, indexType: St
   }
 
   def deleteDocument(identifier: String): Unit = {
-    val response = esClient.delete(new DeleteRequest(indexName, indexType, identifier))
+    val response = esClient.delete(new DeleteRequest(indexName).id(identifier), RequestOptions.DEFAULT)
     logger.info(s"Deleted ${response.getId} to index ${response.getIndex}")
   }
 
   def getDocumentAsString(identifier: String): String = {
-    val response = esClient.get(new GetRequest(indexName, indexType, identifier))
+    val response = esClient.get(new GetRequest(indexName).id(identifier), RequestOptions.DEFAULT)
     response.getSourceAsString
   }
 
@@ -152,9 +149,9 @@ class ElasticSearchUtil(connectionInfo: String, indexName: String, indexType: St
           val doc: util.Map[String, AnyRef] = mapper.readValue(document, new TypeReference[util.Map[String, AnyRef]]() {})
           val updatedDoc = checkDocStringLength(doc)
           logger.debug("ElasticSearchUtil:: bulkIndexWithIndexId:: doc: " + updatedDoc)
-          request.add(new IndexRequest(indexName, documentType, key).source(updatedDoc))
+          request.add(new IndexRequest(indexName).id(key).source(updatedDoc))
           if (count % batchSize == 0 || (count % batchSize < batchSize && count == jsonObjects.size)) {
-            val bulkResponse = esClient.bulk(request)
+            val bulkResponse = esClient.bulk(request, RequestOptions.DEFAULT)
             if (bulkResponse.hasFailures) logger.info("ElasticSearchUtil:: bulkIndexWithIndexId:: Failures in Elasticsearch bulkIndex : " + bulkResponse.buildFailureMessage)
           }
         }
@@ -165,7 +162,7 @@ class ElasticSearchUtil(connectionInfo: String, indexName: String, indexType: St
 
   def isIndexExists(indexName: String): Boolean = {
     try {
-      val response: Response = esClient.getLowLevelClient.performRequest("HEAD", "/" + indexName)
+      val response: Response = esClient.getLowLevelClient.performRequest(new Request("HEAD", "/" + indexName))
       response.getStatusLine.getStatusCode == 200
     } catch {
       case e: IOException =>  false

jobs-core/src/test/scala/org/sunbird/spec/ElasticSearchUtilSpec.scala

@@ -9,7 +9,7 @@ import org.sunbird.job.util.ElasticSearchUtil
 class ElasticSearchUtilSpec extends FlatSpec with Matchers {
 
   val config: Config = ConfigFactory.load("base-test.conf")
-  val esUtil = new ElasticSearchUtil(config.getString("es.basePath"), "compositesearch", "cs")
+  val esUtil = new ElasticSearchUtil(config.getString("es.basePath"), "compositesearch")
 
 //  "isIndexExists" should "return true if index exists" in {
 //    val indexExists = esUtil.isIndexExists("compositesearch")

publish-pipeline/content-publish/src/main/scala/org/sunbird/job/content/function/CollectionPublishFunction.scala

@@ -47,7 +47,7 @@ class CollectionPublishFunction(config: ContentPublishConfig, httpUtil: HttpUtil
     super.open(parameters)
     cassandraUtil = new CassandraUtil(config.cassandraHost, config.cassandraPort, config)
     neo4JUtil = new Neo4JUtil(config.graphRoutePath, config.graphName, config)
-    esUtil = new ElasticSearchUtil(config.esConnectionInfo, config.compositeSearchIndexName, config.compositeSearchIndexType)
+    esUtil = new ElasticSearchUtil(config.esConnectionInfo, config.compositeSearchIndexName)
     cloudStorageUtil = new CloudStorageUtil(config)
     ec = ExecutionContexts.global
     definitionCache = new DefinitionCache()

publish-pipeline/live-node-publisher/src/main/scala/org/sunbird/job/livenodepublisher/function/LiveCollectionPublishFunction.scala

@@ -47,7 +47,7 @@ class LiveCollectionPublishFunction(config: LiveNodePublisherConfig, httpUtil: H
     super.open(parameters)
     cassandraUtil = new CassandraUtil(config.cassandraHost, config.cassandraPort, config)
     neo4JUtil = new Neo4JUtil(config.graphRoutePath, config.graphName, config)
-    esUtil = new ElasticSearchUtil(config.esConnectionInfo, config.compositeSearchIndexName, config.compositeSearchIndexType)
+    esUtil = new ElasticSearchUtil(config.esConnectionInfo, config.compositeSearchIndexName)
     cloudStorageUtil = new CloudStorageUtil(config)
     ec = ExecutionContexts.global
     definitionCache = new DefinitionCache()

publish-pipeline/publish-core/src/test/scala/org/sunbird/job/publish/spec/EcarGeneratorSpec.scala

@@ -35,7 +35,7 @@ class EcarGeneratorSpec extends FlatSpec with BeforeAndAfterAll with Matchers {
   "Object Ecar Generator generateEcar" should "return a Map containing Packaging Type and its url after uploading it to cloud" in {
 
     val hierarchy = Map("identifier" -> "do_123", "children" -> List(Map("identifier" -> "do_234", "name" -> "Children-1", "objectType" -> "Question"), Map("identifier" -> "do_345", "name" -> "Children-2", "objectType" -> "Question")))
-    val metadata = Map("identifier" -> "do_123", "appIcon" -> "https://dev.knowlg.sunbird.org/content/preview/assets/icons/avatar_anonymous.png", "identifier" -> "do_123", "objectType" -> "QuestionSet", "name" -> "Test QuestionSet", "status" -> "Live")
+    val metadata = Map("identifier" -> "do_123", "appIcon" -> "https://dev.sunbirded.org/content/preview/assets/icons/avatar_anonymous.png", "identifier" -> "do_123", "objectType" -> "QuestionSet", "name" -> "Test QuestionSet", "status" -> "Live")
     val objData = new ObjectData("do_123", metadata, None, Some(hierarchy))
     val obj = new TestEcarGenerator()
     val result = obj.generateEcar(objData,List("SPINE"))
@@ -50,6 +50,6 @@ class TestEcarGenerator extends EcarGenerator {
     "src" -> "somepath/sunbird_1551961194254.jpeg",
     "baseUrl" -> "some_base_url"
   )
-  val testObj = List(Map("children" -> List(Map("identifier" -> "do_234", "name" -> "Children-1", "objectType" -> "Question"), Map("identifier" -> "do_345", "name" -> "Children-2", "objectType" -> "Question")), "name" -> "Test QuestionSet", "appIcon" -> "https://dev.knowlg.sunbird.org/content/preview/assets/icons/avatar_anonymous.png", "objectType" -> "QuestionSet", "identifier" -> "do_123", "status" -> "Live", "identifier" -> "do_123"), Map("identifier" -> "do_234", "name" -> "Children-1", "objectType" -> "Question", "media" -> ScalaJsonUtil.serialize(List(media))), Map("identifier" -> "do_345", "name" -> "Children-2", "objectType" -> "Question"))
+  val testObj = List(Map("children" -> List(Map("identifier" -> "do_234", "name" -> "Children-1", "objectType" -> "Question"), Map("identifier" -> "do_345", "name" -> "Children-2", "objectType" -> "Question")), "name" -> "Test QuestionSet", "appIcon" -> "https://dev.sunbirded.org/content/preview/assets/icons/avatar_anonymous.png", "objectType" -> "QuestionSet", "identifier" -> "do_123", "status" -> "Live", "identifier" -> "do_123"), Map("identifier" -> "do_234", "name" -> "Children-1", "objectType" -> "Question", "media" -> ScalaJsonUtil.serialize(List(media))), Map("identifier" -> "do_345", "name" -> "Children-2", "objectType" -> "Question"))
   override def getDataForEcar(obj: ObjectData): Option[List[Map[String, AnyRef]]] = Some(testObj)
 }

publish-pipeline/publish-core/src/test/scala/org/sunbird/job/publish/spec/ObjectEnrichmentSpec.scala

@@ -37,7 +37,7 @@ class ObjectEnrichmentSpec extends FlatSpec with BeforeAndAfterAll with Matchers
     when(mockNeo4JUtil.getNodesName(List("NCERT"))).thenReturn(Map("NCERT"-> "NCERT"))
 
     val hierarchy = Map("identifier" -> "do_123", "children" -> List(Map("identifier" -> "do_234", "name" -> "Children-1"), Map("identifier" -> "do_345", "name" -> "Children-2")))
-    val metadata = Map("identifier" -> "do_123", "targetFWIds" -> List("NCERT"), "boardIds" -> List("NCERT"), "appIcon" -> "https://dev.knowlg.sunbird.org/content/preview/assets/icons/avatar_anonymous.png", "IL_UNIQUE_ID" -> "do_123", "IL_FUNC_OBJECT_TYPE" -> "QuestionSet", "name" -> "Test QuestionSet", "status" -> "Live")
+    val metadata = Map("identifier" -> "do_123", "targetFWIds" -> List("NCERT"), "boardIds" -> List("NCERT"), "appIcon" -> "https://dev.sunbirded.org/content/preview/assets/icons/avatar_anonymous.png", "IL_UNIQUE_ID" -> "do_123", "IL_FUNC_OBJECT_TYPE" -> "QuestionSet", "name" -> "Test QuestionSet", "status" -> "Live")
     val objData = new ObjectData("do_123", metadata, None, Some(hierarchy))
 
     val objectEnrichment = new TestObjectEnrichment()

publish-pipeline/publish-core/src/test/scala/org/sunbird/job/publish/spec/ThumbnailGeneratorSpec.scala

@@ -25,7 +25,7 @@ class ThumbnailGeneratorSpec extends FlatSpec with BeforeAndAfterAll with Matche
   "Object Thumbnail Generator generateThumbnail" should "add the thumbnail to ObjectData" in {
 
     val hierarchy = Map("identifier" -> "do_123", "children" -> List(Map("identifier" -> "do_234", "name" -> "Children-1"), Map("identifier" -> "do_345", "name" -> "Children-2")))
-    val metadata = Map("identifier" -> "do_123", "appIcon" -> "https://dev.knowlg.sunbird.org/content/preview/assets/icons/avatar_anonymous.png", "IL_UNIQUE_ID" -> "do_123", "objectType" -> "QuestionSet", "name" -> "Test QuestionSet", "status" -> "Live")
+    val metadata = Map("identifier" -> "do_123", "appIcon" -> "https://dev.sunbirded.org/content/preview/assets/icons/avatar_anonymous.png", "IL_UNIQUE_ID" -> "do_123", "objectType" -> "QuestionSet", "name" -> "Test QuestionSet", "status" -> "Live")
     val objData = new ObjectData("do_123", metadata, None, Some(hierarchy))
 
     val thumbnailGenerator = new TestThumbnailGenerator()
@@ -35,7 +35,7 @@ class ThumbnailGeneratorSpec extends FlatSpec with BeforeAndAfterAll with Matche
     resultMetadata.isEmpty should be(false)
     resultMetadata.getOrElse("posterImage", "").asInstanceOf[String].isEmpty should be(false)
     resultMetadata.getOrElse("appIcon", "").asInstanceOf[String].isEmpty should be(false)
-    resultMetadata.getOrElse("posterImage", "").asInstanceOf[String] shouldBe "https://dev.knowlg.sunbird.org/content/preview/assets/icons/avatar_anonymous.png"
+    resultMetadata.getOrElse("posterImage", "").asInstanceOf[String] shouldBe "https://dev.sunbirded.org/content/preview/assets/icons/avatar_anonymous.png"
     resultMetadata.getOrElse("appIcon", "").asInstanceOf[String] shouldBe "https://sunbirddev.blob.core.windows.net/sunbird-content-dev/questionset/do_123/artifact/avatar_anonymous.thumb.png"
   }
 

qrcode-image-generator/src/main/scala/org/sunbird/job/qrimagegenerator/functions/QRCodeImageGeneratorFunction.scala

@@ -27,7 +27,7 @@ class QRCodeImageGeneratorFunction(config: QRCodeImageGeneratorConfig,
   override def open(parameters: Configuration): Unit = {
     cassandraUtil = new CassandraUtil(config.cassandraHost, config.cassandraPort, config)
     cloudStorageUtil = new CloudStorageUtil(config)
-    esUtil = new ElasticSearchUtil(config.esConnectionInfo, config.dialcodeExternalIndex, config.dialcodeExternalIndexType)
+    esUtil = new ElasticSearchUtil(config.esConnectionInfo, config.dialcodeExternalIndex)
     qRCodeImageGeneratorUtil = new QRCodeImageGeneratorUtil(config, cassandraUtil, cloudStorageUtil, esUtil)
     super.open(parameters)
   }

qrcode-image-generator/src/main/scala/org/sunbird/job/qrimagegenerator/functions/QRCodeIndexImageUrlFunction.scala

@@ -24,7 +24,7 @@ class QRCodeIndexImageUrlFunction(config: QRCodeImageGeneratorConfig,
 
   override def open(parameters: Configuration): Unit = {
     cassandraUtil = new CassandraUtil(config.cassandraHost, config.cassandraPort, config)
-    esUtil = new ElasticSearchUtil(config.esConnectionInfo, config.dialcodeExternalIndex, config.dialcodeExternalIndexType)
+    esUtil = new ElasticSearchUtil(config.esConnectionInfo, config.dialcodeExternalIndex)
     qRCodeImageGeneratorUtil = new QRCodeImageGeneratorUtil(config, cassandraUtil, cloudStorageUtil, esUtil)
     super.open(parameters)
   }

search-indexer/pom.xml

@@ -73,9 +73,9 @@
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>pl.allegro.tech</groupId>
-            <artifactId>embedded-elasticsearch</artifactId>
-            <version>2.7.0</version>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>elasticsearch</artifactId>
+            <version>1.19.0</version>
             <scope>test</scope>
         </dependency>
     </dependencies>