Add failures field to UpdateByQueryResponse and `DeleteByQueryRes…

…ponse` (#2896) Also update `ReindexResponse` to make its `failures` field have the same type as the one in `UpdateByQueryResponse` and `DeleteByQueryResponse`. Resolves #2883 Co-authored-by: Alex Zolotko <oleksandr.zolotko@ibm.com> Co-authored-by: Sam <sam@sksamuel.com>
Philippus · Aug 11, 2023 · e2a9a5b · e2a9a5b
1 parent b9d39ab
commit e2a9a5b
Show file tree

Hide file tree

Showing 6 changed files with 45 additions and 9 deletions.
diff --git a/elastic4s-domain/src/main/scala/com/sksamuel/elastic4s/BulkIndexByScrollFailure.scala b/elastic4s-domain/src/main/scala/com/sksamuel/elastic4s/BulkIndexByScrollFailure.scala
@@ -0,0 +1,8 @@
+package com.sksamuel.elastic4s
+
+// https://github.com/elastic/elasticsearch-specification/blob/b8b9d95dd6f94dc4e415d37da97095278f9a3a90/specification/_types/Errors.ts#L58
+case class BulkIndexByScrollFailure(cause: ErrorCause,
+                                    id: String,
+                                    index: String,
+                                    status: Int,
+                                    `type`: String)
diff --git a/elastic4s-domain/src/main/scala/com/sksamuel/elastic4s/ErrorCause.scala b/elastic4s-domain/src/main/scala/com/sksamuel/elastic4s/ErrorCause.scala
@@ -0,0 +1,23 @@
+package com.sksamuel.elastic4s
+
+import com.fasterxml.jackson.annotation.{JsonAnySetter, JsonProperty}
+
+import scala.collection.mutable
+
+
+// https://github.com/elastic/elasticsearch-specification/blob/b8b9d95dd6f94dc4e415d37da97095278f9a3a90/specification/_types/Errors.ts#L29
+case class ErrorCause(`type`: String,
+                      reason: Option[String],
+                      @JsonProperty("stack_trace") stackTrace: Option[String],
+                      @JsonProperty("caused_by") causedBy: Option[ErrorCause],
+                      @JsonProperty("root_cause") rootCause: Option[Seq[ErrorCause]],
+                      suppressed: Option[Seq[ErrorCause]]) {
+  private val _other = mutable.HashMap[String, String]()
+
+  //noinspection ScalaUnusedSymbol
+  @JsonAnySetter private def setOther(k: String, v: String): Unit = _other.put(k, v)
+
+  def other(key: String): Option[String] = _other.get(key)
+
+  override def toString: String = s"ErrorCause(${`type`},$reason,${_other})"
+}
diff --git a/...-domain/src/main/scala/com/sksamuel/elastic4s/requests/delete/DeleteByQueryResponse.scala b/...-domain/src/main/scala/com/sksamuel/elastic4s/requests/delete/DeleteByQueryResponse.scala
@@ -1,6 +1,7 @@
 package com.sksamuel.elastic4s.requests.delete
 
 import com.fasterxml.jackson.annotation.JsonProperty
+import com.sksamuel.elastic4s.BulkIndexByScrollFailure
 
 case class DeleteByQueryResponse(took: Long,
                                  @JsonProperty("timed_out") timedOut: Boolean,
@@ -11,4 +12,5 @@ case class DeleteByQueryResponse(took: Long,
                                  noops: Long,
                                  @JsonProperty("throttled_millis") throttledMillis: Long,
                                  @JsonProperty("requests_per_second") requestsPerSecond: Long,
-                                 @JsonProperty("throttled_until_millis") throttledUntilMillis: Long)
+                                 @JsonProperty("throttled_until_millis") throttledUntilMillis: Long,
+                                 failures: Option[Seq[BulkIndexByScrollFailure]])
diff --git a/...-domain/src/main/scala/com/sksamuel/elastic4s/requests/update/UpdateByQueryResponse.scala b/...-domain/src/main/scala/com/sksamuel/elastic4s/requests/update/UpdateByQueryResponse.scala
@@ -1,5 +1,7 @@
 package com.sksamuel.elastic4s.requests.update
 
+import com.sksamuel.elastic4s.BulkIndexByScrollFailure
+
 case class UpdateByQueryResponse(took: Long,
                                  timedOut: Boolean,
                                  total: Long,
@@ -10,4 +12,5 @@ case class UpdateByQueryResponse(took: Long,
                                  noops: Long,
                                  throttledMillis: Long,
                                  requestsPerSecond: Long,
-                                 throttledUntilMillis: Long)
+                                 throttledUntilMillis: Long,
+                                 failures: Option[Seq[BulkIndexByScrollFailure]])
diff --git a/...c4s-handlers/src/main/scala/com/sksamuel/elastic4s/handlers/reindex/ReindexHandlers.scala b/...c4s-handlers/src/main/scala/com/sksamuel/elastic4s/handlers/reindex/ReindexHandlers.scala
@@ -6,14 +6,12 @@ import com.sksamuel.elastic4s.handlers.ElasticErrorParser
 import com.sksamuel.elastic4s.requests.common.RefreshPolicyHttpValue
 import com.sksamuel.elastic4s.requests.reindex.ReindexRequest
 import com.sksamuel.elastic4s.requests.task.CreateTaskResponse
-import com.sksamuel.elastic4s.{ElasticError, ElasticRequest, Handler, HttpEntity, HttpResponse, ResponseHandler}
+import com.sksamuel.elastic4s.{BulkIndexByScrollFailure, ElasticError, ElasticRequest, Handler, HttpEntity, HttpResponse, ResponseHandler}
 
 import scala.concurrent.duration._
 
 case class Retries(bulk: Long, search: Long)
 
-case class ReindexFailure()
-
 case class ReindexResponse(took: Long,
                            timed_out: Boolean,
                            total: Long,
@@ -27,7 +25,7 @@ case class ReindexResponse(took: Long,
                            @JsonProperty("throttled_millis") throttledMillis: Long,
                            @JsonProperty("requests_per_second") requestsPerSecond: Long,
                            @JsonProperty("throttled_until_millis") throttledUntilMillis: Long,
-                           failures: Seq[ReindexFailure]) {
+                           failures: Option[Seq[BulkIndexByScrollFailure]]) {
   def throttled: Duration = throttledMillis.millis
   def throttledUntil: Duration = throttledUntilMillis.millis
 }

diff --git a/...tic4s-handlers/src/main/scala/com/sksamuel/elastic4s/handlers/update/UpdateHandlers.scala b/...tic4s-handlers/src/main/scala/com/sksamuel/elastic4s/handlers/update/UpdateHandlers.scala
@@ -9,7 +9,7 @@ import com.sksamuel.elastic4s.json.{XContentBuilder, XContentFactory}
 import com.sksamuel.elastic4s.requests.common.RefreshPolicyHttpValue
 import com.sksamuel.elastic4s.requests.task.GetTask
 import com.sksamuel.elastic4s.requests.update.{BaseUpdateByQueryRequest, UpdateByQueryAsyncRequest, UpdateByQueryAsyncResponse, UpdateByQueryRequest, UpdateByQueryResponse, UpdateByQueryTask, UpdateRequest, UpdateResponse}
-import com.sksamuel.elastic4s.{ElasticError, ElasticRequest, ElasticUrlEncoder, Handler, HttpEntity, HttpResponse, ResponseHandler}
+import com.sksamuel.elastic4s.{BulkIndexByScrollFailure, ElasticError, ElasticRequest, ElasticUrlEncoder, Handler, HttpEntity, HttpResponse, ResponseHandler}
 
 object UpdateByQueryBodyFn {
   def apply(request: BaseUpdateByQueryRequest): XContentBuilder = {
@@ -110,7 +110,8 @@ trait UpdateHandlers {
         resp.noops,
         resp.throttled_millis,
         resp.requests_per_second,
-        resp.throttled_until_millis
+        resp.throttled_until_millis,
+        resp.failures
       )
     }
   }
@@ -141,5 +142,6 @@ case class UpdateByQuerySnakeCase(
                                    noops: Long,
                                    throttled_millis: Long,
                                    requests_per_second: Long,
-                                   throttled_until_millis: Long
+                                   throttled_until_millis: Long,
+                                   failures: Option[Seq[BulkIndexByScrollFailure]]
                                  )