update executeIndex so as to wait for next refresh or not

Author: fupelaqu

core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala

@@ -551,9 +551,10 @@ trait ElasticClientDelegator extends ElasticClientApi with BulkTypes {
   override private[client] def executeIndex(
     index: String,
     id: String,
-    source: String
+    source: String,
+    wait: Boolean
   ): ElasticResult[Boolean] =
-    delegate.executeIndex(index, id, source)
+    delegate.executeIndex(index, id, source, wait)
 
   override private[client] def executeIndexAsync(
     index: String,
@@ -1127,7 +1128,7 @@ trait ElasticClientDelegator extends ElasticClientApi with BulkTypes {
 
   override private[client] def pitSearchAfter(
     elasticQuery: ElasticQuery,
-    fieldAliases: Map[JSONResults, JSONResults],
+    fieldAliases: Map[String, String],
     config: ScrollConfig,
     hasSorts: Boolean
   )(implicit system: ActorSystem) = {

core/src/main/scala/app/softnetwork/elastic/client/IndexApi.scala

@@ -91,11 +91,9 @@ trait IndexApi extends ElasticClientHelpers { _: RefreshApi with SerializationAp
 
     logger.debug(s"Indexing document with id '$id' in index '$index'")
 
-    executeIndex(index, id, source) match {
+    executeIndex(index, id, source, wait=false) match {
       case success @ ElasticSuccess(true) =>
         logger.info(s"✅ Document with id '$id' indexed successfully in index '$index'")
-        // Refresh the index to make sure the document is available for search
-        this.refresh(index)
         success
       case success @ ElasticSuccess(_) =>
         logger.info(s"✅ Document with id '$id' not indexed in index '$index'")
@@ -206,7 +204,8 @@ trait IndexApi extends ElasticClientHelpers { _: RefreshApi with SerializationAp
   private[client] def executeIndex(
     index: String,
     id: String,
-    source: String
+    source: String,
+    wait: Boolean
   ): ElasticResult[Boolean]
 
   private[client] def executeIndexAsync(

es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestIndexApi.scala

@@ -36,15 +36,22 @@ trait JestIndexApi extends IndexApi with JestClientHelpers {
   override private[client] def executeIndex(
     index: String,
     id: String,
-    source: String
+    source: String,
+    wait: Boolean
   ): ElasticResult[Boolean] =
     executeJestBooleanAction(
       operation = "index",
       index = Some(index),
       retryable = true
-    )(
-      new Index.Builder(source).index(index).`type`("_doc").id(id).build()
-    )
+    ) {
+      val refresh = if (wait) "wait_for" else "false"
+      new Index.Builder(source)
+        .index(index)
+        .`type`("_doc")
+        .id(id)
+        .setParameter("refresh", refresh)
+        .build()
+    }
 
   /** Index a document in the given index asynchronously.
     * @see

es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala

@@ -48,6 +48,7 @@ import org.elasticsearch.action.search.{
   SearchResponse,
   SearchScrollRequest
 }
+import org.elasticsearch.action.support.WriteRequest
 import org.elasticsearch.action.support.master.AcknowledgedResponse
 import org.elasticsearch.action.update.{UpdateRequest, UpdateResponse}
 import org.elasticsearch.action.{ActionListener, DocWriteRequest}
@@ -490,18 +491,22 @@ trait RestHighLevelClientIndexApi extends IndexApi with RestHighLevelClientHelpe
   override private[client] def executeIndex(
     index: String,
     id: String,
-    source: String
+    source: String,
+    wait: Boolean
   ): result.ElasticResult[Boolean] =
     executeRestAction[IndexRequest, IndexResponse, Boolean](
       operation = "index",
       index = Some(index),
       retryable = false
-    )(
-      request = new IndexRequest(index)
+    )(request = {
+      val refresh =
+        if (wait) WriteRequest.RefreshPolicy.WAIT_UNTIL else WriteRequest.RefreshPolicy.NONE
+      new IndexRequest(index)
         .`type`("_doc")
         .id(id)
         .source(source, XContentType.JSON)
-    )(
+        .setRefreshPolicy(refresh)
+    })(
       executor = req => apply().index(req, RequestOptions.DEFAULT)
     )(response => response.status().getStatus < 400)
 

es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala

@@ -49,6 +49,7 @@ import org.elasticsearch.action.search.{
   SearchResponse,
   SearchScrollRequest
 }
+import org.elasticsearch.action.support.WriteRequest
 import org.elasticsearch.action.support.master.AcknowledgedResponse
 import org.elasticsearch.action.update.{UpdateRequest, UpdateResponse}
 import org.elasticsearch.action.{ActionListener, DocWriteRequest}
@@ -496,17 +497,23 @@ trait RestHighLevelClientIndexApi extends IndexApi with RestHighLevelClientHelpe
   override private[client] def executeIndex(
     index: String,
     id: String,
-    source: String
+    source: String,
+    wait: Boolean
   ): result.ElasticResult[Boolean] =
     executeRestAction[IndexRequest, IndexResponse, Boolean](
       operation = "index",
       index = Some(index),
       retryable = false
     )(
-      request = new IndexRequest(index)
-        .`type`("_doc")
-        .id(id)
-        .source(source, XContentType.JSON)
+      request = {
+        val refresh =
+          if (wait) WriteRequest.RefreshPolicy.WAIT_UNTIL else WriteRequest.RefreshPolicy.NONE
+        new IndexRequest(index)
+          .`type`("_doc")
+          .id(id)
+          .source(source, XContentType.JSON)
+          .setRefreshPolicy(refresh)
+      }
     )(
       executor = req => apply().index(req, RequestOptions.DEFAULT)
     )(response => response.status().getStatus < 400)

es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala

@@ -27,7 +27,14 @@ import app.softnetwork.elastic.sql.bridge._
 import app.softnetwork.elastic.sql.query.{SQLAggregation, SQLSearchRequest}
 import app.softnetwork.elastic.client
 import app.softnetwork.elastic.client.result.{ElasticFailure, ElasticResult, ElasticSuccess}
-import co.elastic.clients.elasticsearch._types.{FieldSort, FieldValue, SortOptions, SortOrder, Time}
+import co.elastic.clients.elasticsearch._types.{
+  FieldSort,
+  FieldValue,
+  Refresh,
+  SortOptions,
+  SortOrder,
+  Time
+}
 import co.elastic.clients.elasticsearch.core.bulk.{
   BulkOperation,
   DeleteOperation,
@@ -510,22 +517,25 @@ trait JavaClientIndexApi extends IndexApi with JavaClientHelpers {
   override private[client] def executeIndex(
     index: String,
     id: String,
-    source: String
+    source: String,
+    wait: Boolean
   ): result.ElasticResult[Boolean] =
     executeJavaBooleanAction(
       operation = "index",
       index = Some(index),
       retryable = false
-    )(
+    ) {
+      val refresh = if (wait) Refresh.WaitFor else Refresh.False
       apply()
         .index(
           new IndexRequest.Builder()
             .index(index)
             .id(id)
             .withJson(new StringReader(source))
+            .refresh(refresh)
             .build()
         )
-    )(
+    }(
       _.shards()
         .failed()
         .intValue() == 0

es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala

@@ -27,7 +27,14 @@ import app.softnetwork.elastic.sql.bridge._
 import app.softnetwork.elastic.sql.query.{SQLAggregation, SQLSearchRequest}
 import app.softnetwork.elastic.client
 import app.softnetwork.elastic.client.result.{ElasticFailure, ElasticResult, ElasticSuccess}
-import co.elastic.clients.elasticsearch._types.{FieldSort, FieldValue, SortOptions, SortOrder, Time}
+import co.elastic.clients.elasticsearch._types.{
+  FieldSort,
+  FieldValue,
+  Refresh,
+  SortOptions,
+  SortOrder,
+  Time
+}
 import co.elastic.clients.elasticsearch.core.bulk.{
   BulkOperation,
   DeleteOperation,
@@ -46,7 +53,6 @@ import com.google.gson.JsonParser
 import _root_.java.io.{IOException, StringReader}
 import _root_.java.util.{Map => JMap}
 import scala.jdk.CollectionConverters._
-
 import scala.concurrent.{ExecutionContext, Future}
 import scala.language.implicitConversions
 import scala.util.{Failure, Success, Try}
@@ -508,22 +514,25 @@ trait JavaClientIndexApi extends IndexApi with JavaClientHelpers {
   override private[client] def executeIndex(
     index: String,
     id: String,
-    source: String
+    source: String,
+    wait: Boolean
   ): result.ElasticResult[Boolean] =
     executeJavaBooleanAction(
       operation = "index",
       index = Some(index),
       retryable = false
-    )(
+    ) {
+      val refresh = if (wait) Refresh.WaitFor else Refresh.False
       apply()
         .index(
           new IndexRequest.Builder()
             .index(index)
             .id(id)
             .withJson(new StringReader(source))
+            .refresh(refresh)
             .build()
         )
-    )(
+    }(
       _.shards()
         .failed()
         .intValue() == 0