java - Elasticsearch 索引创建后立即执行读取操作导致异常

Question

我尝试在创建 Elasticsearch 索引后立即对其执行读取操作。下面是重现这种情况的简单代码:

import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import static java.net.InetAddress.getLoopbackAddress;

public class ElasticIssue {

  static String index = "my_index";

  public static void main(String[] args) {
    final Client c = getClient();
    deleteIndexIfExists(c);
    createIndex(c);
    //refresh(c);
    //flush(c);
    //delay();
    //indexDoc(c);
    getDoc(c);
  }

  static void getDoc(Client client) {
    client.prepareGet(index, "some-type", "1").get();
  }

  static void indexDoc(Client client) {
    client.prepareIndex(index, "another-type", "25").setSource("{}").get();
  }

  static void createIndex(Client client) {
    client.admin().indices().prepareCreate(index).get();
  }

  static void delay() {
    try {Thread.sleep(3000);} catch (InterruptedException e) {}
  }

  static void flush(Client client) {
    client.admin().indices().prepareFlush(index).get();
  }

  private static void refresh(Client client) {
    client.admin().indices().prepareRefresh(index).get();
  }

  static void deleteIndexIfExists(Client client) {
    final IndicesExistsResponse response = client.admin().indices().prepareExists(index).get();
    if (response.isExists()) {
      deleteIndex(client);
    }
  }

  static void deleteIndex(Client client) {
    client.admin().indices().prepareDelete(index).get();
  }

  static Client getClient() {
    final Settings settings = Settings.builder()
        .put("cluster.name", "elasticsearch") //default name
        .put("node.name", "my-node")
        .build();
    return TransportClient.builder()
        .settings(settings)
        .build()
        .addTransportAddress(new InetSocketTransportAddress(getLoopbackAddress(), 9300));
  }
}

然后我收到以下错误:

Exception in thread "main" NoShardAvailableActionException[No shard available for [get [my_index][some-type][1]: routing [null]]]; nested: RemoteTransportException[[my-node][172.17.0.2:9300][indices:data/read/get[s]]]; nested: IllegalIndexShardStateException[CurrentState[RECOVERING] operations only allowed when shard state is one of [POST_RECOVERY, STARTED, RELOCATED]];
    at org.elasticsearch.action.support.single.shard.TransportSingleShardAction$AsyncSingleAction.perform(TransportSingleShardAction.java:199)
    at org.elasticsearch.action.support.single.shard.TransportSingleShardAction$AsyncSingleAction.onFailure(TransportSingleShardAction.java:186)
    at org.elasticsearch.action.support.single.shard.TransportSingleShardAction$AsyncSingleAction.access$1300(TransportSingleShardAction.java:115)
    at org.elasticsearch.action.support.single.shard.TransportSingleShardAction$AsyncSingleAction$2.handleException(TransportSingleShardAction.java:240)
    at org.elasticsearch.transport.TransportService$DirectResponseChannel.processException(TransportService.java:855)
    at org.elasticsearch.transport.TransportService$DirectResponseChannel.sendResponse(TransportService.java:833)
    at org.elasticsearch.transport.TransportService$4.onFailure(TransportService.java:387)
    at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:39)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: RemoteTransportException[[my-node][172.17.0.2:9300][indices:data/read/get[s]]]; nested: IllegalIndexShardStateException[CurrentState[RECOVERING] operations only allowed when shard state is one of [POST_RECOVERY, STARTED, RELOCATED]];
Caused by: [my_index][[my_index][3]] IllegalIndexShardStateException[CurrentState[RECOVERING] operations only allowed when shard state is one of [POST_RECOVERY, STARTED, RELOCATED]]
    at org.elasticsearch.index.shard.IndexShard.readAllowed(IndexShard.java:1035)
    at org.elasticsearch.index.shard.IndexShard.get(IndexShard.java:651)
    at org.elasticsearch.index.get.ShardGetService.innerGet(ShardGetService.java:173)
    at org.elasticsearch.index.get.ShardGetService.get(ShardGetService.java:86)
    at org.elasticsearch.action.get.TransportGetAction.shardOperation(TransportGetAction.java:101)
    at org.elasticsearch.action.get.TransportGetAction.shardOperation(TransportGetAction.java:44)
    at org.elasticsearch.action.support.single.shard.TransportSingleShardAction$ShardTransportHandler.messageReceived(TransportSingleShardAction.java:282)
    at org.elasticsearch.action.support.single.shard.TransportSingleShardAction$ShardTransportHandler.messageReceived(TransportSingleShardAction.java:275)
    at org.elasticsearch.transport.TransportRequestHandler.messageReceived(TransportRequestHandler.java:33)
    at org.elasticsearch.transport.RequestHandlerRegistry.processMessageReceived(RequestHandlerRegistry.java:75)
    at org.elasticsearch.transport.TransportService$4.doRun(TransportService.java:376)
    at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:37)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

尽管已返回响应，但 Elasticsearch 索引创建似乎尚未完成。这有点令人沮丧。如果我执行以下任何操作:延迟、索引任何文档、刷新索引、刷新索引(取消注释任何行)；然后读取操作成功。

此行为的解释是什么？确保索引准备好工作的推荐方法是什么？列出的解决方案是通过实验找到的。

我正在使用 Elasticsearch 2.3.3 和 Java 8。与 Elasticsearch 的所有通信都是使用传输协议(protocol)(使用 Java api)完成的。

为了更轻松地设置，这里使用 docker 命令来获取具有所有必要设置的容器:

docker run -p 9200:9200 -p 9300:9300 elasticsearch:2.3.3 -Des.node.name="my-node"

这是 Elasticsearch Java API 的 Maven 依赖项:

<dependency>
  <groupId>org.elasticsearch</groupId>
  <artifactId>elasticsearch</artifactId>
  <version>2.3.3</version>
</dependency>

Answer 1

您需要等待索引创建完成。您可以执行此操作，直到索引的运行状况处于黄色状态。

索引创建函数后调用以下函数:

static void indexStatusCheck(Client client) {
     ClusterHealthResponse response = client.admin().cluster().prepareHealth().setIndices(index).setWaitForYellowStatus().get();
if (response.getStatus() == ClusterHealthStatus.RED) {
 throw Exception("Index not ready");
}
    }

然后您可以继续调用 getDoc()。