Merge pull request #3560 from magda-io/fix-index-trim

Fix index trim opereation
magda-io · Aug 30, 2024 · 2913644 · 2913644
2 parents 8d5ba80 + 3fd211e
commit 2913644
Show file tree

Hide file tree

Showing 3 changed files with 43 additions and 12 deletions.
diff --git a/CHANGES.md b/CHANGES.md
@@ -1,5 +1,10 @@
 # CHANGELOG
 
+## v4.2.4
+
+- #3559: Set `conflicts` to true when Indexer performs the trim operation.
+- Increase indexer client connection idle-timeout to avoid encountering connection reset error for downloading large region files
+
 ## v4.2.3
 
 - #3553: Make Chart Preview & Table Preview Configurable per Dataset

diff --git a/magda-indexer/src/main/resources/application.conf b/magda-indexer/src/main/resources/application.conf
@@ -33,6 +33,10 @@ akka {
             idle-timeout = 120s
         }
 
+        client {
+            # User might increase this setting when encounter connection reset error for downloading large region files
+            idle-timeout = 180s
+        }
 
     }
 }

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)