Added options for request converters

Author: sohaibiftikhar

client/rest-high-level/src/main/java/org/elasticsearch/client/RequestConverters.java

@@ -816,7 +816,11 @@ static Request clusterHealth(ClusterHealthRequest healthRequest) {
     static Request reindex(ReindexRequest reindexRequest) throws IOException {
         String endpoint = new EndpointBuilder().addPathPart("_reindex").build();
         Request request = new Request(HttpPost.METHOD_NAME, endpoint);
-        Params params = new Params(request);
+        Params params = new Params(request)
+            .withRefresh(reindexRequest.isRefresh())
+            .withTimeout(reindexRequest.getTimeout())
+            .withWaitForActiveShards(reindexRequest.getWaitForActiveShards());
+
         if (reindexRequest.getScrollTime() != null) {
             params.putParam("scroll", reindexRequest.getScrollTime());
         }

client/rest-high-level/src/test/java/org/elasticsearch/client/RequestConvertersTests.java

@@ -453,6 +453,11 @@ public void testReindex() throws IOException {
             String ts = randomTimeValue();
             reindexRequest.setScroll(TimeValue.parseTimeValue(ts, "scroll"));
         }
+        if (reindexRequest.getRemoteInfo() == null && randomBoolean()) {
+            reindexRequest.setSourceQuery(new TermQueryBuilder("foo", "fooval"));
+        }
+        setRandomTimeout(reindexRequest::setTimeout, ReplicationRequest.DEFAULT_TIMEOUT, expectedParams);
+        setRandomWaitForActiveShards(reindexRequest::setWaitForActiveShards, ActiveShardCount.DEFAULT, expectedParams);
         expectedParams.put("scroll", reindexRequest.getScrollTime().getStringRep());
         Request request = RequestConverters.reindex(reindexRequest);
         assertEquals("/_reindex", request.getEndpoint());

server/src/main/java/org/elasticsearch/index/reindex/AbstractBulkByScrollRequest.java

@@ -252,6 +252,14 @@ public Self setTimeout(TimeValue timeout) {
         return self();
     }
 
+    /**
+     * Timeout to wait for the shards on to be available for each bulk request?
+     */
+    public Self setTimeout(String timeout) {
+        this.timeout = TimeValue.parseTimeValue(timeout, this.timeout, getClass().getSimpleName() + ".timeout");
+        return self();
+    }
+
     /**
      * The number of shard copies that must be active before proceeding with the write.
      */