Core: Remove ability to run optimize and upgrade async

This has been very trappy. Rather than continue to allow buggy behavior
of having upgrade/optimize requests sidestep the single shard per node
limits optimize is supposed to be subject to, this removes
the ability to run the upgrade/optimize async.

closes #9638

Author: rjernst

docs/reference/indices/optimize.asciidoc

@@ -7,6 +7,10 @@ operations (and relates to the number of segments a Lucene index holds
 within each shard). The optimize operation allows to reduce the number
 of segments by merging them.
 
+This call will block until the optimize is complete. If the http connection
+is lost, the request will continue in the background, and
+any new requests will block until the previous optimize is complete.
+
 [source,js]
 --------------------------------------------------
 $ curl -XPOST 'http://localhost:9200/twitter/_optimize'
@@ -33,10 +37,6 @@ deletes. Defaults to `false`.  Note that this won't override the
 `flush`::  Should a flush be performed after the optimize. Defaults to
 `true`.
 
-`wait_for_merge`:: Should the request wait for the merge to end. Defaults
-to `true`. Note, a merge can potentially be a very heavy operation, so
-it might make sense to run it set to `false`.
-
 [float]
 [[optimize-multi-index]]
 === Multi Index

docs/reference/indices/upgrade.asciidoc

@@ -17,15 +17,9 @@ NOTE: Upgrading is an I/O intensive operation, and is limited to processing a
 single shard per node at a time.  It also is not allowed to run at the same
 time as optimize.
 
-[float]
-[[upgrade-parameters]]
-==== Request Parameters
-
-The `upgrade` API accepts the following request parameters:
-
-[horizontal]
-`wait_for_completion`:: Should the request wait for the upgrade to complete. Defaults
-to `false`.
+This call will block until the upgrade is complete. If the http connection
+is lost, the request will continue in the background, and
+any new requests will block until the previous upgrade is complete.
 
 [float]
 === Check upgrade status

src/main/java/org/elasticsearch/action/admin/indices/optimize/OptimizeRequest.java

@@ -30,9 +30,6 @@
  * A request to optimize one or more indices. In order to optimize on all the indices, pass an empty array or
  * <tt>null</tt> for the indices.
  * <p/>
- * <p>{@link #waitForMerge(boolean)} allows to control if the call will block until the optimize completes and
- * defaults to <tt>true</tt>.
- * <p/>
  * <p>{@link #maxNumSegments(int)} allows to control the number of segments to optimize down to. By default, will
  * cause the optimize process to optimize down to half the configured number of segments.
  *
@@ -43,14 +40,12 @@
 public class OptimizeRequest extends BroadcastOperationRequest<OptimizeRequest> {
 
     public static final class Defaults {
-        public static final boolean WAIT_FOR_MERGE = true;
         public static final int MAX_NUM_SEGMENTS = -1;
         public static final boolean ONLY_EXPUNGE_DELETES = false;
         public static final boolean FLUSH = true;
         public static final boolean UPGRADE = false;
     }
-
-    private boolean waitForMerge = Defaults.WAIT_FOR_MERGE;
+    
     private int maxNumSegments = Defaults.MAX_NUM_SEGMENTS;
     private boolean onlyExpungeDeletes = Defaults.ONLY_EXPUNGE_DELETES;
     private boolean flush = Defaults.FLUSH;
@@ -69,21 +64,6 @@ public OptimizeRequest() {
 
     }
 
-    /**
-     * Should the call block until the optimize completes. Defaults to <tt>true</tt>.
-     */
-    public boolean waitForMerge() {
-        return waitForMerge;
-    }
-
-    /**
-     * Should the call block until the optimize completes. Defaults to <tt>true</tt>.
-     */
-    public OptimizeRequest waitForMerge(boolean waitForMerge) {
-        this.waitForMerge = waitForMerge;
-        return this;
-    }
-
     /**
      * Will optimize the index down to <= maxNumSegments. By default, will cause the optimize
      * process to optimize down to half the configured number of segments.
@@ -168,7 +148,6 @@ public OptimizeRequest upgrade(boolean upgrade) {
 
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        waitForMerge = in.readBoolean();
         maxNumSegments = in.readInt();
         onlyExpungeDeletes = in.readBoolean();
         flush = in.readBoolean();
@@ -179,7 +158,6 @@ public void readFrom(StreamInput in) throws IOException {
 
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeBoolean(waitForMerge);
         out.writeInt(maxNumSegments);
         out.writeBoolean(onlyExpungeDeletes);
         out.writeBoolean(flush);
@@ -191,8 +169,7 @@ public void writeTo(StreamOutput out) throws IOException {
     @Override
     public String toString() {
         return "OptimizeRequest{" +
-                "waitForMerge=" + waitForMerge +
-                ", maxNumSegments=" + maxNumSegments +
+                "maxNumSegments=" + maxNumSegments +
                 ", onlyExpungeDeletes=" + onlyExpungeDeletes +
                 ", flush=" + flush +
                 ", upgrade=" + upgrade +

src/main/java/org/elasticsearch/action/admin/indices/optimize/OptimizeRequestBuilder.java

@@ -27,9 +27,6 @@
  * A request to optimize one or more indices. In order to optimize on all the indices, pass an empty array or
  * <tt>null</tt> for the indices.
  * <p/>
- * <p>{@link #setWaitForMerge(boolean)} allows to control if the call will block until the optimize completes and
- * defaults to <tt>true</tt>.
- * <p/>
  * <p>{@link #setMaxNumSegments(int)} allows to control the number of segments to optimize down to. By default, will
  * cause the optimize process to optimize down to half the configured number of segments.
  */
@@ -39,14 +36,6 @@ public OptimizeRequestBuilder(IndicesAdminClient indicesClient) {
         super(indicesClient, new OptimizeRequest());
     }
 
-    /**
-     * Should the call block until the optimize completes. Defaults to <tt>true</tt>.
-     */
-    public OptimizeRequestBuilder setWaitForMerge(boolean waitForMerge) {
-        request.waitForMerge(waitForMerge);
-        return this;
-    }
-
     /**
      * Will optimize the index down to <= maxNumSegments. By default, will cause the optimize
      * process to optimize down to half the configured number of segments.

src/main/java/org/elasticsearch/action/admin/indices/optimize/ShardOptimizeRequest.java

@@ -32,8 +32,7 @@
  *
  */
 class ShardOptimizeRequest extends BroadcastShardOperationRequest {
-
-    private boolean waitForMerge = OptimizeRequest.Defaults.WAIT_FOR_MERGE;
+    
     private int maxNumSegments = OptimizeRequest.Defaults.MAX_NUM_SEGMENTS;
     private boolean onlyExpungeDeletes = OptimizeRequest.Defaults.ONLY_EXPUNGE_DELETES;
     private boolean flush = OptimizeRequest.Defaults.FLUSH;
@@ -44,17 +43,12 @@ class ShardOptimizeRequest extends BroadcastShardOperationRequest {
 
     ShardOptimizeRequest(ShardId shardId, OptimizeRequest request) {
         super(shardId, request);
-        waitForMerge = request.waitForMerge();
         maxNumSegments = request.maxNumSegments();
         onlyExpungeDeletes = request.onlyExpungeDeletes();
         flush = request.flush();
         upgrade = request.force() || request.upgrade();
     }
 
-    boolean waitForMerge() {
-        return waitForMerge;
-    }
-
     int maxNumSegments() {
         return maxNumSegments;
     }
@@ -74,7 +68,9 @@ public boolean upgrade() {
     @Override
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        waitForMerge = in.readBoolean();
+        if (in.getVersion().before(Version.V_1_5_0)) {
+            in.readBoolean(); // backcompat for waitForMerges, no longer exists
+        }
         maxNumSegments = in.readInt();
         onlyExpungeDeletes = in.readBoolean();
         flush = in.readBoolean();
@@ -86,7 +82,9 @@ public void readFrom(StreamInput in) throws IOException {
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeBoolean(waitForMerge);
+        if (out.getVersion().before(Version.V_1_5_0)) {
+            out.writeBoolean(true);
+        }
         out.writeInt(maxNumSegments);
         out.writeBoolean(onlyExpungeDeletes);
         out.writeBoolean(flush);

src/main/java/org/elasticsearch/action/admin/indices/optimize/TransportOptimizeAction.java

@@ -108,7 +108,7 @@ protected ShardOptimizeResponse newShardResponse() {
     protected ShardOptimizeResponse shardOperation(ShardOptimizeRequest request) throws ElasticsearchException {
         IndexShard indexShard = indicesService.indexServiceSafe(request.shardId().getIndex()).shardSafe(request.shardId().id());
         indexShard.optimize(new OptimizeRequest().flush(request.flush()).onlyExpungeDeletes(request.onlyExpungeDeletes())
-                .maxNumSegments(request.maxNumSegments()).waitForMerge(request.waitForMerge()).upgrade(request.upgrade()));
+                .maxNumSegments(request.maxNumSegments()).upgrade(request.upgrade()));
         return new ShardOptimizeResponse(request.shardId());
     }
 

src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -227,12 +227,12 @@ public Condition newCondition() {
     /**
      * Optimizes to 1 segment
      */
-    abstract void forceMerge(boolean flush, boolean waitForMerge);
+    abstract void forceMerge(boolean flush);
 
     /**
      * Triggers a forced merge on this engine
      */
-    public abstract void forceMerge(boolean flush, boolean waitForMerge, int maxNumSegments, boolean onlyExpungeDeletes, boolean upgrade) throws EngineException;
+    public abstract void forceMerge(boolean flush, int maxNumSegments, boolean onlyExpungeDeletes, boolean upgrade) throws EngineException;
 
     /**
      * Snapshots the index and returns a handle to it. Will always try and "commit" the

src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -843,12 +843,12 @@ private void waitForMerges(boolean flushAfter, boolean upgrade) {
     }
 
     @Override
-    public void forceMerge(boolean flush, boolean waitForMerge) {
-        forceMerge(flush, waitForMerge, 1, false, false);
+    public void forceMerge(boolean flush) {
+        forceMerge(flush, 1, false, false);
     }
 
     @Override
-    public void forceMerge(final boolean flush, boolean waitForMerge, int maxNumSegments, boolean onlyExpungeDeletes, final boolean upgrade) throws EngineException {
+    public void forceMerge(final boolean flush, int maxNumSegments, boolean onlyExpungeDeletes, final boolean upgrade) throws EngineException {
         if (optimizeMutex.compareAndSet(false, true)) {
             try (ReleasableLock _ = readLock.acquire()) {
                 ensureOpen();
@@ -881,23 +881,7 @@ public void forceMerge(final boolean flush, boolean waitForMerge, int maxNumSegm
             }
         }
 
-        // wait for the merges outside of the read lock
-        if (waitForMerge) {
-            waitForMerges(flush, upgrade);
-        } else if (flush || upgrade) {
-            // we only need to monitor merges for async calls if we are going to flush
-            engineConfig.getThreadPool().executor(ThreadPool.Names.OPTIMIZE).execute(new AbstractRunnable() {
-                @Override
-                public void onFailure(Throwable t) {
-                    logger.error("Exception while waiting for merges asynchronously after optimize", t);
-                }
-
-                @Override
-                protected void doRun() throws Exception {
-                    waitForMerges(flush, upgrade);
-                }
-            });
-        }
+        waitForMerges(flush, upgrade);
     }
 
 

src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -616,8 +616,7 @@ public void optimize(OptimizeRequest optimize) throws ElasticsearchException {
         if (logger.isTraceEnabled()) {
             logger.trace("optimize with {}", optimize);
         }
-        engine().forceMerge(optimize.flush(), optimize.waitForMerge(), optimize
-                .maxNumSegments(), optimize.onlyExpungeDeletes(), optimize.upgrade());
+        engine().forceMerge(optimize.flush(), optimize.maxNumSegments(), optimize.onlyExpungeDeletes(), optimize.upgrade());
     }
 
     public SnapshotIndexCommit snapshotIndex() throws EngineException {

src/main/java/org/elasticsearch/rest/action/admin/indices/optimize/RestOptimizeAction.java

@@ -55,7 +55,6 @@ public void handleRequest(final RestRequest request, final RestChannel channel,
         OptimizeRequest optimizeRequest = new OptimizeRequest(Strings.splitStringByCommaToArray(request.param("index")));
         optimizeRequest.listenerThreaded(false);
         optimizeRequest.indicesOptions(IndicesOptions.fromRequest(request, optimizeRequest.indicesOptions()));
-        optimizeRequest.waitForMerge(request.paramAsBoolean("wait_for_merge", optimizeRequest.waitForMerge()));
         optimizeRequest.maxNumSegments(request.paramAsInt("max_num_segments", optimizeRequest.maxNumSegments()));
         optimizeRequest.onlyExpungeDeletes(request.paramAsBoolean("only_expunge_deletes", optimizeRequest.onlyExpungeDeletes()));
         optimizeRequest.flush(request.paramAsBoolean("flush", optimizeRequest.flush()));

src/main/java/org/elasticsearch/rest/action/admin/indices/upgrade/RestUpgradeAction.java

@@ -90,7 +90,6 @@ public RestResponse buildResponse(IndicesSegmentResponse response, XContentBuild
 
     void handlePost(RestRequest request, RestChannel channel, Client client) {
         OptimizeRequest optimizeReq = new OptimizeRequest(Strings.splitStringByCommaToArray(request.param("index")));
-        optimizeReq.waitForMerge(request.paramAsBoolean("wait_for_completion", false));
         optimizeReq.flush(true);
         optimizeReq.upgrade(true);
         optimizeReq.maxNumSegments(Integer.MAX_VALUE); // we just want to upgrade the segments, not actually optimize to a single segment

src/test/java/org/elasticsearch/gateway/local/SimpleRecoveryLocalGatewayTests.java

@@ -365,7 +365,7 @@ public void testReusePeerRecovery() throws Exception {
         }
         logger.info("Running Cluster Health");
         ensureGreen();
-        client().admin().indices().prepareOptimize("test").setWaitForMerge(true).setMaxNumSegments(100).get(); // just wait for merges
+        client().admin().indices().prepareOptimize("test").setMaxNumSegments(100).get(); // just wait for merges
         client().admin().indices().prepareFlush().setWaitIfOngoing(true).setForce(true).get();
 
         logger.info("--> shutting down the nodes");

src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java

@@ -385,31 +385,10 @@ public void testSegments() throws Exception {
     public void testSegmentsWithMergeFlag() throws Exception {
 
         ConcurrentMergeSchedulerProvider mergeSchedulerProvider = new ConcurrentMergeSchedulerProvider(shardId, EMPTY_SETTINGS, threadPool, new IndexSettingsService(shardId.index(), EMPTY_SETTINGS));
-        final AtomicReference<CountDownLatch> waitTillMerge = new AtomicReference<>();
-        final AtomicReference<CountDownLatch> waitForMerge = new AtomicReference<>();
-        mergeSchedulerProvider.addListener(new MergeSchedulerProvider.Listener() {
-            @Override
-            public void beforeMerge(OnGoingMerge merge) {
-                try {
-                    if (waitTillMerge.get() != null) {
-                        waitTillMerge.get().countDown();
-                    }
-                    if (waitForMerge.get() != null) {
-                        waitForMerge.get().await();
-                    }
-                } catch (InterruptedException e) {
-                    throw ExceptionsHelper.convertToRuntime(e);
-                }
-            }
-
-            @Override
-            public void afterMerge(OnGoingMerge merge) {
-            }
-        });
-
-        final Store store = createStore();
         IndexSettingsService indexSettingsService = new IndexSettingsService(shardId.index(), ImmutableSettings.builder().put(defaultSettings).put(IndexMetaData.SETTING_VERSION_CREATED, Version.CURRENT).build());
+        final Store store = createStore();
         final Engine engine = createEngine(indexSettingsService, store, createTranslog(), mergeSchedulerProvider);
+
         ParsedDocument doc = testParsedDocument("1", "1", "test", null, -1, -1, testDocument(), Lucene.STANDARD_ANALYZER, B_1, false);
         Engine.Index index = new Engine.Index(null, newUid("1"), doc);
         engine.index(index);
@@ -430,23 +409,12 @@ public void afterMerge(OnGoingMerge merge) {
             assertThat(segment.getMergeId(), nullValue());
         }
 
-        waitTillMerge.set(new CountDownLatch(1));
-        waitForMerge.set(new CountDownLatch(1));
-        engine.forceMerge(false, false);
-        waitTillMerge.get().await();
-
-        for (Segment segment : engine.segments()) {
-            assertThat(segment.getMergeId(), notNullValue());
-        }
-
-        waitForMerge.get().countDown();
-
         index = new Engine.Index(null, newUid("4"), doc);
         engine.index(index);
         engine.flush();
         final long gen1 = store.readLastCommittedSegmentsInfo().getGeneration();
         // now, optimize and wait for merges, see that we have no merge flag
-        engine.forceMerge(true, true);
+        engine.forceMerge(true);
 
         for (Segment segment : engine.segments()) {
             assertThat(segment.getMergeId(), nullValue());
@@ -456,25 +424,14 @@ public void afterMerge(OnGoingMerge merge) {
 
         final boolean flush = randomBoolean();
         final long gen2 = store.readLastCommittedSegmentsInfo().getGeneration();
-        engine.forceMerge(flush, false);
-        waitTillMerge.get().await();
+        engine.forceMerge(flush);
         for (Segment segment : engine.segments()) {
             assertThat(segment.getMergeId(), nullValue());
         }
-        waitForMerge.get().countDown();
 
         if (flush) {
-            awaitBusy(new Predicate<Object>() {
-                @Override
-                public boolean apply(Object o) {
-                    try {
-                        // we should have had just 1 merge, so last generation should be exact
-                        return store.readLastCommittedSegmentsInfo().getLastGeneration() == gen2;
-                    } catch (IOException e) {
-                        throw ExceptionsHelper.convertToRuntime(e);
-                    }
-                }
-            });
+            // we should have had just 1 merge, so last generation should be exact
+            assertEquals(gen2 + 1, store.readLastCommittedSegmentsInfo().getLastGeneration());
         }
 
         engine.close();