Revert "Move DeleteByQuery and Reindex requests into core (#24578)"

This reverts commit 6ea2ae3.
elastic · May 11, 2017 · 952feb5 · 952feb5
1 parent 57fddce
commit 952feb5
Show file tree

Hide file tree

Showing 77 changed files with 182 additions and 90 deletions.
diff --git a/buildSrc/src/main/resources/checkstyle_suppressions.xml b/buildSrc/src/main/resources/checkstyle_suppressions.xml
@@ -127,6 +127,7 @@
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]bulk[/\\]Retry.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]bulk[/\\]TransportBulkAction.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]bulk[/\\]TransportShardBulkAction.java" checks="LineLength" />
+  <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]bulk[/\\]byscroll[/\\]AbstractAsyncBulkByScrollAction.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]delete[/\\]DeleteRequest.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]explain[/\\]TransportExplainAction.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]get[/\\]GetRequest.java" checks="LineLength" />

diff --git a/...ndex/AbstractAsyncBulkByScrollAction.java → ...roll/AbstractAsyncBulkByScrollAction.java b/...ndex/AbstractAsyncBulkByScrollAction.java → ...roll/AbstractAsyncBulkByScrollAction.java
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.index.reindex;
+package org.elasticsearch.action.bulk.byscroll;
 
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
@@ -31,7 +31,7 @@
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.bulk.Retry;
-import org.elasticsearch.index.reindex.ScrollableHitSource.SearchFailure;
+import org.elasticsearch.action.bulk.byscroll.ScrollableHitSource.SearchFailure;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.client.ParentTaskAssigningClient;
@@ -77,7 +77,7 @@
 import static java.util.Collections.emptyList;
 import static java.util.Collections.unmodifiableList;
 import static org.elasticsearch.action.bulk.BackoffPolicy.exponentialBackoff;
-import static org.elasticsearch.index.reindex.AbstractBulkByScrollRequest.SIZE_ALL_MATCHES;
+import static org.elasticsearch.action.bulk.byscroll.AbstractBulkByScrollRequest.SIZE_ALL_MATCHES;
 import static org.elasticsearch.common.unit.TimeValue.timeValueNanos;
 import static org.elasticsearch.rest.RestStatus.CONFLICT;
 import static org.elasticsearch.search.sort.SortBuilders.fieldSort;
@@ -116,8 +116,8 @@ public abstract class AbstractAsyncBulkByScrollAction<Request extends AbstractBu
     private final BiFunction<RequestWrapper<?>, ScrollableHitSource.Hit, RequestWrapper<?>> scriptApplier;
 
     public AbstractAsyncBulkByScrollAction(WorkingBulkByScrollTask task, Logger logger, ParentTaskAssigningClient client,
-                                           ThreadPool threadPool, Request mainRequest, ScriptService scriptService,
-                                           ClusterState clusterState, ActionListener<BulkByScrollResponse> listener) {
+                                           ThreadPool threadPool, Request mainRequest, ScriptService scriptService, ClusterState clusterState,
+                                           ActionListener<BulkByScrollResponse> listener) {
         this(task, logger, client, threadPool, mainRequest, scriptService, clusterState, listener, client.settings());
     }
 
@@ -741,7 +741,7 @@ public DeleteRequest self() {
     /**
      * Wraps a {@link DeleteRequest} in a {@link RequestWrapper}
      */
-    public static RequestWrapper<DeleteRequest> wrap(DeleteRequest request) {
+    static RequestWrapper<DeleteRequest> wrap(DeleteRequest request) {
         return new DeleteRequestWrapper(request);
     }
 

diff --git a/.../reindex/AbstractBulkByScrollRequest.java → ...byscroll/AbstractBulkByScrollRequest.java b/.../reindex/AbstractBulkByScrollRequest.java → ...byscroll/AbstractBulkByScrollRequest.java
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.index.reindex;
+package org.elasticsearch.action.bulk.byscroll;
 
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionRequest;
@@ -355,7 +355,7 @@ public int getSlices() {
     /**
      * Build a new request for a slice of the parent request.
      */
-    public abstract Self forSlice(TaskId slicingTask, SearchRequest slice);
+    protected abstract Self forSlice(TaskId slicingTask, SearchRequest slice);
 
     /**
      * Setup a clone of this request with the information needed to process a slice of it.

diff --git a/...x/AbstractBulkByScrollRequestBuilder.java → ...l/AbstractBulkByScrollRequestBuilder.java b/...x/AbstractBulkByScrollRequestBuilder.java → ...l/AbstractBulkByScrollRequestBuilder.java
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.index.reindex;
+package org.elasticsearch.action.bulk.byscroll;
 
 import org.elasticsearch.action.Action;
 import org.elasticsearch.action.ActionRequestBuilder;

diff --git a/...dex/reindex/AsyncDeleteByQueryAction.java → ...lk/byscroll/AsyncDeleteByQueryAction.java b/...dex/reindex/AsyncDeleteByQueryAction.java → ...lk/byscroll/AsyncDeleteByQueryAction.java
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.index.reindex;
+package org.elasticsearch.action.bulk.byscroll;
 
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
@@ -32,8 +32,8 @@
  */
 public class AsyncDeleteByQueryAction extends AbstractAsyncBulkByScrollAction<DeleteByQueryRequest> {
     public AsyncDeleteByQueryAction(WorkingBulkByScrollTask task, Logger logger, ParentTaskAssigningClient client,
-                                    ThreadPool threadPool, DeleteByQueryRequest request, ScriptService scriptService,
-                                    ClusterState clusterState, ActionListener<BulkByScrollResponse> listener) {
+            ThreadPool threadPool, DeleteByQueryRequest request, ScriptService scriptService, ClusterState clusterState,
+            ActionListener<BulkByScrollResponse> listener) {
         super(task, logger, client, threadPool, request, scriptService, clusterState, listener);
     }
 

diff --git a/...ex/BulkByScrollParallelizationHelper.java → ...ll/BulkByScrollParallelizationHelper.java b/...ex/BulkByScrollParallelizationHelper.java → ...ll/BulkByScrollParallelizationHelper.java
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.index.reindex;
+package org.elasticsearch.action.bulk.byscroll;
 
 import org.elasticsearch.action.Action;
 import org.elasticsearch.action.ActionListener;
@@ -32,13 +32,13 @@
 /**
  * Helps parallelize reindex requests using sliced scrolls.
  */
-class BulkByScrollParallelizationHelper {
+public class BulkByScrollParallelizationHelper {
     private BulkByScrollParallelizationHelper() {}
 
-    public static <Request extends AbstractBulkByScrollRequest<Request>> void startSlices(Client client, TaskManager taskManager,
-                               Action<Request, BulkByScrollResponse, ?> action,
-                               String localNodeId, ParentBulkByScrollTask task, Request request,
-                               ActionListener<BulkByScrollResponse> listener) {
+    public static <
+                Request extends AbstractBulkByScrollRequest<Request>
+            > void startSlices(Client client, TaskManager taskManager, Action<Request, BulkByScrollResponse, ?> action,
+                    String localNodeId, ParentBulkByScrollTask task, Request request, ActionListener<BulkByScrollResponse> listener) {
         TaskId parentTaskId = new TaskId(localNodeId, task.getId());
         for (final SearchRequest slice : sliceIntoSubRequests(request.getSearchRequest(), UidFieldMapper.NAME, request.getSlices())) {
             // TODO move the request to the correct node. maybe here or somehow do it as part of startup for reindex in general....

diff --git a/...h/index/reindex/BulkByScrollResponse.java → ...n/bulk/byscroll/BulkByScrollResponse.java b/...h/index/reindex/BulkByScrollResponse.java → ...n/bulk/byscroll/BulkByScrollResponse.java
@@ -17,10 +17,11 @@
  * under the License.
  */
 
-package org.elasticsearch.index.reindex;
+package org.elasticsearch.action.bulk.byscroll;
 
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.bulk.BulkItemResponse.Failure;
+import org.elasticsearch.action.bulk.byscroll.ScrollableHitSource.SearchFailure;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -44,14 +45,14 @@ public class BulkByScrollResponse extends ActionResponse implements ToXContent {
     private TimeValue took;
     private BulkByScrollTask.Status status;
     private List<Failure> bulkFailures;
-    private List<ScrollableHitSource.SearchFailure> searchFailures;
+    private List<SearchFailure> searchFailures;
     private boolean timedOut;
 
     public BulkByScrollResponse() {
     }
 
     public BulkByScrollResponse(TimeValue took, BulkByScrollTask.Status status, List<Failure> bulkFailures,
-                                List<ScrollableHitSource.SearchFailure> searchFailures, boolean timedOut) {
+                                     List<SearchFailure> searchFailures, boolean timedOut) {
         this.took = took;
         this.status = requireNonNull(status, "Null status not supported");
         this.bulkFailures = bulkFailures;
@@ -138,7 +139,7 @@ public List<Failure> getBulkFailures() {
     /**
      * All search failures.
      */
-    public List<ScrollableHitSource.SearchFailure> getSearchFailures() {
+    public List<SearchFailure> getSearchFailures() {
         return searchFailures;
     }
 
@@ -165,7 +166,7 @@ public void readFrom(StreamInput in) throws IOException {
         took = new TimeValue(in);
         status = new BulkByScrollTask.Status(in);
         bulkFailures = in.readList(Failure::new);
-        searchFailures = in.readList(ScrollableHitSource.SearchFailure::new);
+        searchFailures = in.readList(SearchFailure::new);
         timedOut = in.readBoolean();
     }
 
@@ -180,7 +181,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             failure.toXContent(builder, params);
             builder.endObject();
         }
-        for (ScrollableHitSource.SearchFailure failure: searchFailures) {
+        for (SearchFailure failure: searchFailures) {
             failure.toXContent(builder, params);
         }
         builder.endArray();
@@ -198,4 +199,4 @@ public String toString() {
         builder.append(",search_failures=").append(getSearchFailures().subList(0, min(3, getSearchFailures().size())));
         return builder.append(']').toString();
     }
-}
+}
diff --git a/...earch/index/reindex/BulkByScrollTask.java → ...ction/bulk/byscroll/BulkByScrollTask.java b/...earch/index/reindex/BulkByScrollTask.java → ...ction/bulk/byscroll/BulkByScrollTask.java
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.index.reindex;
+package org.elasticsearch.action.bulk.byscroll;
 
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;

diff --git a/...ex/reindex/ClientScrollableHitSource.java → ...k/byscroll/ClientScrollableHitSource.java b/...ex/reindex/ClientScrollableHitSource.java → ...k/byscroll/ClientScrollableHitSource.java
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.index.reindex;
+package org.elasticsearch.action.bulk.byscroll;
 
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;

diff --git a/...h/index/reindex/DeleteByQueryRequest.java → ...n/bulk/byscroll/DeleteByQueryRequest.java b/...h/index/reindex/DeleteByQueryRequest.java → ...n/bulk/byscroll/DeleteByQueryRequest.java
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.index.reindex;
+package org.elasticsearch.action.bulk.byscroll;
 
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.IndicesRequest;
@@ -81,7 +81,7 @@ public ActionRequestValidationException validate() {
     }
 
     @Override
-    public DeleteByQueryRequest forSlice(TaskId slicingTask, SearchRequest slice) {
+    protected DeleteByQueryRequest forSlice(TaskId slicingTask, SearchRequest slice) {
         return doForSlice(new DeleteByQueryRequest(slice, false), slicingTask);
     }
 

diff --git a/...index/reindex/ParentBulkByScrollTask.java → ...bulk/byscroll/ParentBulkByScrollTask.java b/...index/reindex/ParentBulkByScrollTask.java → ...bulk/byscroll/ParentBulkByScrollTask.java
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.index.reindex;
+package org.elasticsearch.action.bulk.byscroll;
 
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.common.collect.Tuple;
@@ -104,7 +104,7 @@ public void onSliceResponse(ActionListener<BulkByScrollResponse> listener, int s
     /**
      * Record a failure from a slice and respond to the listener if the request is finished.
      */
-    public void onSliceFailure(ActionListener<BulkByScrollResponse> listener, int sliceId, Exception e) {
+    void onSliceFailure(ActionListener<BulkByScrollResponse> listener, int sliceId, Exception e) {
         results.setOnce(sliceId, new Result(sliceId, e));
         recordSliceCompletionAndRespondIfAllDone(listener);
         // TODO cancel when a slice fails?

diff --git a/...ch/index/reindex/ScrollableHitSource.java → ...on/bulk/byscroll/ScrollableHitSource.java b/...ch/index/reindex/ScrollableHitSource.java → ...on/bulk/byscroll/ScrollableHitSource.java
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.index.reindex;
+package org.elasticsearch.action.bulk.byscroll;
 
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ElasticsearchException;
@@ -111,7 +111,7 @@ public final void close(Runnable onCompletion) {
     /**
      * Set the id of the last scroll. Used for debugging.
      */
-    public final void setScroll(String scrollId) {
+    final void setScroll(String scrollId) {
         this.scrollId.set(scrollId);
     }
 

diff --git a/.../index/reindex/SuccessfullyProcessed.java → .../bulk/byscroll/SuccessfullyProcessed.java b/.../index/reindex/SuccessfullyProcessed.java → .../bulk/byscroll/SuccessfullyProcessed.java
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.index.reindex;
+package org.elasticsearch.action.bulk.byscroll;
 
 /**
  * Implemented by {@link BulkByScrollTask} and {@link BulkByScrollTask.Status} to consistently implement

diff --git a/...ndex/reindex/WorkingBulkByScrollTask.java → ...ulk/byscroll/WorkingBulkByScrollTask.java b/...ndex/reindex/WorkingBulkByScrollTask.java → ...ulk/byscroll/WorkingBulkByScrollTask.java
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.index.reindex;
+package org.elasticsearch.action.bulk.byscroll;
 
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.common.logging.ESLoggerFactory;
@@ -115,15 +115,15 @@ TimeValue throttledUntil() {
         return timeValueNanos(max(0, delayed.future.getDelay(TimeUnit.NANOSECONDS)));
     }
 
-    public void setTotal(long totalHits) {
+    void setTotal(long totalHits) {
         total.set(totalHits);
     }
 
-    public void countBatch() {
+    void countBatch() {
         batch.incrementAndGet();
     }
 
-    public void countNoop() {
+    void countNoop() {
         noops.incrementAndGet();
     }
 
@@ -132,7 +132,7 @@ public long getCreated() {
         return created.get();
     }
 
-    public void countCreated() {
+    void countCreated() {
         created.incrementAndGet();
     }
 
@@ -141,7 +141,7 @@ public long getUpdated() {
         return updated.get();
     }
 
-    public void countUpdated() {
+    void countUpdated() {
         updated.incrementAndGet();
     }
 
@@ -150,15 +150,15 @@ public long getDeleted() {
         return deleted.get();
     }
 
-    public void countDeleted() {
+    void countDeleted() {
         deleted.incrementAndGet();
     }
 
-    public void countVersionConflict() {
+    void countVersionConflict() {
         versionConflicts.incrementAndGet();
     }
 
-    public void countBulkRetry() {
+    void countBulkRetry() {
         bulkRetries.incrementAndGet();
     }
 
@@ -174,8 +174,8 @@ float getRequestsPerSecond() {
      * Schedule prepareBulkRequestRunnable to run after some delay. This is where throttling plugs into reindexing so the request can be
      * rescheduled over and over again.
      */
-    public void delayPrepareBulkRequest(ThreadPool threadPool, TimeValue lastBatchStartTime, int lastBatchSize,
-                                        AbstractRunnable prepareBulkRequestRunnable) {
+    void delayPrepareBulkRequest(ThreadPool threadPool, TimeValue lastBatchStartTime, int lastBatchSize,
+            AbstractRunnable prepareBulkRequestRunnable) {
         // Synchronize so we are less likely to schedule the same request twice.
         synchronized (delayedPrepareBulkRequestReference) {
             TimeValue delay = throttleWaitTime(lastBatchStartTime, timeValueNanos(System.nanoTime()), lastBatchSize);
@@ -184,7 +184,7 @@ public void delayPrepareBulkRequest(ThreadPool threadPool, TimeValue lastBatchSt
         }
     }
 
-    public TimeValue throttleWaitTime(TimeValue lastBatchStartTime, TimeValue now, int lastBatchSize) {
+    TimeValue throttleWaitTime(TimeValue lastBatchStartTime, TimeValue now, int lastBatchSize) {
         long earliestNextBatchStartTime = now.nanos() + (long) perfectlyThrottledBatchTime(lastBatchSize);
         return timeValueNanos(max(0, earliestNextBatchStartTime - System.nanoTime()));
     }

diff --git a/...ticsearch/index/reindex/package-info.java → ...ch/action/bulk/byscroll/package-info.java b/...ticsearch/index/reindex/package-info.java → ...ch/action/bulk/byscroll/package-info.java
@@ -18,7 +18,6 @@
  */
 
 /**
- * Infrastructure for actions that modify documents based on the results of a scrolling query
- * like reindex, update by query or delete by query.
+ * Infrastructure for actions that modify documents based on the results of a scrolling query.
  */
-package org.elasticsearch.index.reindex;
+package org.elasticsearch.action.bulk.byscroll;
diff --git a/...reindex/AsyncBulkByScrollActionTests.java → ...yscroll/AsyncBulkByScrollActionTests.java b/...reindex/AsyncBulkByScrollActionTests.java → ...yscroll/AsyncBulkByScrollActionTests.java
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.index.reindex;
+package org.elasticsearch.action.bulk.byscroll;
 
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ExceptionsHelper;
@@ -36,8 +36,8 @@
 import org.elasticsearch.action.bulk.BulkItemResponse.Failure;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
-import org.elasticsearch.index.reindex.ScrollableHitSource.Hit;
-import org.elasticsearch.index.reindex.ScrollableHitSource.SearchFailure;
+import org.elasticsearch.action.bulk.byscroll.ScrollableHitSource.Hit;
+import org.elasticsearch.action.bulk.byscroll.ScrollableHitSource.SearchFailure;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.delete.DeleteResponse;
 import org.elasticsearch.action.index.IndexRequest;
@@ -696,7 +696,7 @@ private static class DummyAbstractBulkByScrollRequest extends AbstractBulkByScro
         }
 
         @Override
-        public DummyAbstractBulkByScrollRequest forSlice(TaskId slicingTask, SearchRequest slice) {
+        protected DummyAbstractBulkByScrollRequest forSlice(TaskId slicingTask, SearchRequest slice) {
             throw new UnsupportedOperationException();
         }