- Set conflicts to true when Indexer performs the trim operation.

magda-io · Aug 30, 2024 · 2101f9f · 2101f9f
1 parent 8d5ba80
commit 2101f9f
Show file tree

Hide file tree

Showing 2 changed files with 38 additions and 12 deletions.
diff --git a/CHANGES.md b/CHANGES.md
@@ -1,5 +1,9 @@
 # CHANGELOG
 
+## v4.2.4
+
+- Set `conflicts` to true when Indexer performs the trim operation.
+
 ## v4.2.3
 
 - #3553: Make Chart Preview & Table Preview Configurable per Dataset

diff --git a/.../main/scala/au/csiro/data61/magda/indexer/search/elasticsearch/ElasticSearchIndexer.scala b/.../main/scala/au/csiro/data61/magda/indexer/search/elasticsearch/ElasticSearchIndexer.scala
@@ -579,23 +579,45 @@ class ElasticSearchIndexer(
       setupFuture
         .flatMap { client =>
           client.execute(
-            deleteIn(idxName).by(rangeQuery("indexed").lt(before.toString))
+            deleteIn(idxName)
+              .by(rangeQuery("indexed").lt(before.toString))
+              .proceedOnConflicts(true)
           )
         }
-        .map {
-          case results: RequestSuccess[DeleteByQueryResponse] =>
-            logger.info(
-              "Trimmed index {} for {} old datasets",
-              idxName,
-              results.result.deleted
-            )
-          case ESGenericException(e) =>
-            logger.info(
+        .map { res =>
+          if (res.isError) {
+            logger.error(
               "Failed to Trimmed index {} old datasets: {}",
               idxName,
-              e.getMessage
+              res.body.getOrElse(res.error.reason)
             )
-        }
+            throw res.error.asException
+          } else {
+            res.result match {
+              case Left(r) =>
+                logger.info(
+                  "Trimmed index {} for {} old datasets",
+                  idxName,
+                  r.deleted
+                )
+              case Right(r) =>
+                logger.info(
+                  "A task has been created for trimmed index {}. Task Id: {}. Node Id: {}",
+                  idxName,
+                  r.taskId,
+                  r.nodeId
+                )
+            }
+          }
+        } recover {
+        case e: Throwable =>
+          logger.error(
+            "Failed to Trimmed index {} old datasets: {}",
+            idxName,
+            e.getMessage
+          )
+          throw e
+      }
     }
 
     Future.sequence(trimIndexFutureList).map(_ => Unit)