Merge pull request #319 from github/jherns/rethrottle

add rethrottle support to reindex api

CHANGELOG.md

@@ -1,6 +1,9 @@
+## 6.2.1 (2024-12-10)
+- Add support for rethrottling reindex tasks
+
 ## 6.2.0 (2024-11-06)
-- Added support for reindex API
-- Removed CI checks for ES 5.6.15
+- Add support for reindex API
+- Remove CI checks for ES 5.6.15
 
 ## 6.1.1 (2024-06-05)
 - Unlock faraday_middleware version to allow 1.x

lib/elastomer_client/client/reindex.rb

@@ -24,6 +24,11 @@ def reindex(body, params = {})
         response.body
       end
 
+      def rethrottle(task_id, params = {})
+        response = client.post "/_reindex/#{task_id}/_rethrottle", params.merge(params, action: "rethrottle", rest_api: "reindex")
+        response.body
+      end
+
     end
   end
 end

lib/elastomer_client/version.rb

@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 module ElastomerClient
-  VERSION = "6.2.0"
+  VERSION = "6.2.1"
 
   def self.version
     VERSION

test/client/reindex_test.rb

@@ -49,6 +49,30 @@
     assert_equal "Book 1", doc["_source"]["title"]
   end
 
+  it "successfully rethrottles a reindex task" do
+    reindex = $client.reindex
+    body = {
+      source: { index: @source_index.name },
+      dest: { index: @dest_index.name }
+    }
+    response = reindex.reindex(body, requests_per_second: 0.01, wait_for_completion: false)
+    task_id = response["task"]
+
+    reindex.rethrottle(task_id, requests_per_second: 1)
+
+    tasks = $client.tasks
+    node_id = task_id.split(":").first
+    task_id = task_id.split(":").last.to_i
+
+    # wait for the task to complete
+    tasks.wait_by_id(node_id, task_id, "30s")
+
+    # Verify that the document has been reindexed
+    doc = @dest_index.docs.get(id: 1, type: "book")
+
+    assert_equal "Book 1", doc["_source"]["title"]
+  end
+
   it "creates a new index when the destination index does not exist" do
     reindex = $client.reindex
     body = {
@@ -69,7 +93,7 @@
     }
 
     exception = assert_raises(ElastomerClient::Client::RequestError) do
-      response = reindex.reindex(body)
+      reindex.reindex(body)
     end
     assert_equal(404, exception.status)
   end