Adapt SyncedFlushService (#37691)

Author: tlrx

server/src/main/java/org/elasticsearch/indices/flush/SyncedFlushService.java

@@ -32,7 +32,6 @@
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.node.DiscoveryNode;
-import org.elasticsearch.cluster.routing.IndexRoutingTable;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.service.ClusterService;
@@ -289,16 +288,14 @@ private void reportSuccessWithExistingSyncId(ShardId shardId,
         listener.onResponse(new ShardsSyncedFlushResult(shardId, existingSyncId, totalShards, results));
     }
 
-    final IndexShardRoutingTable getShardRoutingTable(ShardId shardId, ClusterState state) {
-        final IndexRoutingTable indexRoutingTable = state.routingTable().index(shardId.getIndexName());
-        if (indexRoutingTable == null) {
-            IndexMetaData index = state.getMetaData().index(shardId.getIndex());
-            if (index != null && index.getState() == IndexMetaData.State.CLOSE) {
-                throw new IndexClosedException(shardId.getIndex());
-            }
+    final IndexShardRoutingTable getShardRoutingTable(final ShardId shardId, final ClusterState state) {
+        final IndexMetaData indexMetaData = state.getMetaData().index(shardId.getIndex());
+        if (indexMetaData == null) {
             throw new IndexNotFoundException(shardId.getIndexName());
+        } else if (indexMetaData.getState() == IndexMetaData.State.CLOSE) {
+            throw new IndexClosedException(shardId.getIndex());
         }
-        final IndexShardRoutingTable shardRoutingTable = indexRoutingTable.shard(shardId.id());
+        final IndexShardRoutingTable shardRoutingTable = state.routingTable().index(indexMetaData.getIndex()).shard(shardId.id());
         if (shardRoutingTable == null) {
             throw new ShardNotFoundException(shardId);
         }

server/src/test/java/org/elasticsearch/indices/flush/SyncedFlushSingleNodeTests.java

@@ -20,11 +20,13 @@
 
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.lease.Releasable;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.shard.IndexShard;
@@ -38,6 +40,8 @@
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
 
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+
 public class SyncedFlushSingleNodeTests extends ESSingleNodeTestCase {
 
     public void testModificationPreventsFlushing() throws InterruptedException {
@@ -130,22 +134,26 @@ public void testSyncFailsIfOperationIsInFlight() throws InterruptedException, Ex
     }
 
     public void testSyncFailsOnIndexClosedOrMissing() throws InterruptedException {
-        createIndex("test");
+        createIndex("test", Settings.builder()
+            .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
+            .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0)
+            .build());
         IndexService test = getInstanceFromNode(IndicesService.class).indexService(resolveIndex("test"));
-        IndexShard shard = test.getShardOrNull(0);
+        final IndexShard shard = test.getShardOrNull(0);
+        assertNotNull(shard);
+        final ShardId shardId = shard.shardId();
+
+        final SyncedFlushService flushService = getInstanceFromNode(SyncedFlushService.class);
 
-        SyncedFlushService flushService = getInstanceFromNode(SyncedFlushService.class);
         SyncedFlushUtil.LatchedListener listener = new SyncedFlushUtil.LatchedListener();
-        flushService.attemptSyncedFlush(new ShardId("test", "_na_", 1), listener);
+        flushService.attemptSyncedFlush(new ShardId(shard.shardId().getIndex(), 1), listener);
         listener.latch.await();
         assertNotNull(listener.error);
         assertNull(listener.result);
         assertEquals(ShardNotFoundException.class, listener.error.getClass());
         assertEquals("no such shard", listener.error.getMessage());
 
-        final ShardId shardId = shard.shardId();
-
-        client().admin().indices().prepareClose("test").get();
+        assertAcked(client().admin().indices().prepareClose("test"));
         listener = new SyncedFlushUtil.LatchedListener();
         flushService.attemptSyncedFlush(shardId, listener);
         listener.latch.await();