From 6130d151725c202fd11dfdbd267478edefc83f56 Mon Sep 17 00:00:00 2001 From: Tanguy Leroux Date: Wed, 23 Jan 2019 11:08:54 +0100 Subject: [PATCH] Adapt SyncedFlushService (#37691) --- .../indices/flush/SyncedFlushService.java | 15 +++++-------- .../flush/SyncedFlushSingleNodeTests.java | 22 +++++++++++++------ 2 files changed, 21 insertions(+), 16 deletions(-) diff --git a/server/src/main/java/org/elasticsearch/indices/flush/SyncedFlushService.java b/server/src/main/java/org/elasticsearch/indices/flush/SyncedFlushService.java index 9bc4e4ead12..0423559aaf5 100644 --- a/server/src/main/java/org/elasticsearch/indices/flush/SyncedFlushService.java +++ b/server/src/main/java/org/elasticsearch/indices/flush/SyncedFlushService.java @@ -32,7 +32,6 @@ import org.elasticsearch.cluster.ClusterState; 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; @@ -290,16 +289,14 @@ public class SyncedFlushService implements IndexEventListener { 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); } diff --git a/server/src/test/java/org/elasticsearch/indices/flush/SyncedFlushSingleNodeTests.java b/server/src/test/java/org/elasticsearch/indices/flush/SyncedFlushSingleNodeTests.java index b9e0bd13f35..9d7f3d5e253 100644 --- a/server/src/test/java/org/elasticsearch/indices/flush/SyncedFlushSingleNodeTests.java +++ b/server/src/test/java/org/elasticsearch/indices/flush/SyncedFlushSingleNodeTests.java @@ -20,11 +20,13 @@ package org.elasticsearch.indices.flush; 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.List; 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 class SyncedFlushSingleNodeTests extends ESSingleNodeTestCase { } 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();