Java 类org.elasticsearch.client.transport.NoNodeAvailableException 实例源码

public static ElasticsearchClient getIndex() {
    if (index == null) {
        // create index
        String elasticsearchAddress = config.getOrDefault("grid.elasticsearch.address", "localhost:9300");
        String elasticsearchClusterName = config.getOrDefault("grid.elasticsearch.clusterName", "");
        String elasticsearchWebIndexName= config.getOrDefault("grid.elasticsearch.webIndexName", "web");
        Path webMappingPath = Paths.get("conf/mappings/web.json");
        if (webMappingPath.toFile().exists()) try {
            index = new ElasticsearchClient(new String[]{elasticsearchAddress}, elasticsearchClusterName.length() == 0 ? null : elasticsearchClusterName);
            index.createIndexIfNotExists(elasticsearchWebIndexName, 1 /*shards*/, 1 /*replicas*/);
            String mapping = new String(Files.readAllBytes(webMappingPath));
            JSONObject mo = new JSONObject(new JSONTokener(mapping));
            mo = mo.getJSONObject("mappings").getJSONObject("_default_");
            index.setMapping("web", mo.toString());
            Data.logger.info("Connected elasticsearch at " + getHost(elasticsearchAddress));
        } catch (IOException | NoNodeAvailableException e) {
            index = null; // index not available
            Data.logger.info("Failed connecting elasticsearch at " + getHost(elasticsearchAddress) + ": " + e.getMessage(), e);
        } else {
            Data.logger.info("no web index mapping available, no connection to elasticsearch attempted");
        }
    }
    return index;
}

@Override
public void run() {
    while(true) {
        try {
            logger.debug("getting es cluster health.");
            ActionFuture<ClusterHealthResponse> healthFuture = transportClient.admin().cluster().health(Requests.clusterHealthRequest());
            ClusterHealthResponse healthResponse = healthFuture.get(5, TimeUnit.SECONDS);
            logger.debug("Get num of node:{}", healthResponse.getNumberOfNodes());
            logger.debug("Get cluster health:{} ", healthResponse.getStatus());
            isClusterOn.getAndSet(true);
        } catch(Throwable t) {
            if(t instanceof NoNodeAvailableException){//集群不可用
                logger.error("the cluster no node avaliable.");
                isClusterOn.getAndSet(false);
                  }else{
                isClusterOn.getAndSet(true);
                  }
        }
        try {
            Thread.sleep(1000);//FIXME
        } catch (InterruptedException ie) { 
            ie.printStackTrace(); 
        }
    }
}

@After
public void cleanup() throws IOException
{
  try {
    DeleteIndexResponse delete = store.client.admin().indices().delete(new DeleteIndexRequest(INDEX_NAME)).actionGet();
    if (!delete.isAcknowledged()) {
      logger.error("Index wasn't deleted");
    }

    store.disconnect();
  } catch (NoNodeAvailableException e) {
    //This indicates that elasticsearch is not running on a particular machine.
    //Silently ignore in this case.
  }

}

private static boolean testClusterRunning(boolean withSecurity) throws IOException {
        try {
            NodesInfoResponse response = client.admin().cluster().prepareNodesInfo().get();
            Version version = response.getNodes().get(0).getVersion();
            logger.info("Starting integration tests against an external cluster running elasticsearch [{}] with {}",
                    version, withSecurity ? "security" : "no security" );
            return withSecurity;
//        } catch (NoNodeAvailableException e) {
//            // If we have an exception here, let's ignore the test
//            logger.warn("Integration tests are skipped: [{}]", e.getMessage());
//            assumeThat("Integration tests are skipped", e.getMessage(), not(containsString("Connection refused")));
//            return withSecurity;
        } catch (NoNodeAvailableException e) {
            if (e.getMessage() == "401") {
                logger.debug("The cluster is secured. So we need to build a client with security", e);
                return true;
            } else {
                logger.error("Full error is", e);
                throw e;
            }
        }
    }

/**
 * A method to persist Liferay index to Elasticsearch server document.
 * 
 * @param esDocument
 *            the json document
 */
private void writeIndex(ElasticserachJSONDocument esDocument) {

    try {
        if (esDocument.isError()) {
            _log.warn("Coudln't store document in index. Error..." + esDocument.getErrorMessage());

        } else {
            Client client = _esConnector.getClient();
            IndexResponse response = client
                    .prepareIndex(ElasticsearchIndexerConstants.ELASTIC_SEARCH_LIFERAY_INDEX,
                            esDocument.getIndexType(), esDocument.getId()).setSource(esDocument.getJsonDocument())
                    .execute().actionGet();
            if (_log.isDebugEnabled()) {
                _log.debug("Document indexed successfully with Id:" + esDocument.getId() + " ,Type:"
                        + esDocument.getIndexType() + " ,Updated index version:" + response.getVersion());
            }
        }
    } catch (NoNodeAvailableException noNodeEx) {
        _log.error("No node available:" + noNodeEx.getDetailedMessage());
    }
}

/**
 * Delete index by query.
 * 
 * @param uid
 *            the uid
 */
private void deleteIndexByQuery(String uid) {

    try {
        /** Don't handle plugin deployment documents, skip them */
        if(!uid.endsWith(ElasticsearchIndexerConstants.WAR)){
            Client client = _esConnector.getClient();
            DeleteByQueryResponse response = client
                    .prepareDeleteByQuery(ElasticsearchIndexerConstants.ELASTIC_SEARCH_LIFERAY_INDEX)
                    .setQuery(QueryBuilders.queryString(ElasticsearchIndexerConstants.ELASTIC_SEARCH_QUERY_UID + uid))
                    .execute().actionGet();

            if (_log.isDebugEnabled()) {
                _log.debug("Document deleted successfully with Id:" + uid + " , Status:" + response.status());
            }
        }
    } catch (NoNodeAvailableException noNodeEx) {
        _log.error("No node available:" + noNodeEx.getDetailedMessage());
    } catch (IndexMissingException indexMissingEx) {
        _log.error("No index availabe:" + indexMissingEx.getDetailedMessage());
    }
}

@Override
public void onFailure(Throwable e) {
    if (ExceptionsHelper.unwrapCause(e) instanceof ConnectTransportException) {
        int n = ++this.n;
        if (n >= nodes.size()) {
            listener.onFailure(new NoNodeAvailableException("none of the configured nodes were available: "
                    + nodes, e));
        } else {
            try {
                logger.warn("retrying on anoher node (n={}, nodes={})", n, nodes.size());
                callback.doWithNode(nodes.get((index + n) % nodes.size()), this);
            } catch (final Throwable t) {
                listener.onFailure(t);
            }
        }
    } else {
        listener.onFailure(e);
    }
}

@Test
public void testBulkClient() throws IOException {
    final BulkTransportClient client = ClientBuilder.builder()
            .put(getSettings())
            .put(ClientBuilder.FLUSH_INTERVAL, TimeValue.timeValueSeconds(60))
            .setMetric(new LongAdderIngestMetric())
            .toBulkTransportClient();
    client.newIndex("test");
    if (client.hasThrowable()) {
        logger.error("error", client.getThrowable());
    }
    assertFalse(client.hasThrowable());
    try {
        client.deleteIndex("test")
          .newIndex("test")
          .deleteIndex("test");
    } catch (NoNodeAvailableException e) {
        logger.error("no node available");
    } finally {
        if (client.hasThrowable()) {
            logger.error("error", client.getThrowable());
        }
        assertFalse(client.hasThrowable());
        client.shutdown();
    }
}

@Test
public void testRandomDocsNodeClient() throws Exception {
    long numactions = NUM_ACTIONS;
    final BulkNodeClient client = ClientBuilder.builder()
            .put(ClientBuilder.MAX_ACTIONS_PER_REQUEST, MAX_ACTIONS)
            .put(ClientBuilder.FLUSH_INTERVAL, TimeValue.timeValueSeconds(60))
            .setMetric(new ElasticsearchIngestMetric())
            .toBulkNodeClient(client("1"));
    try {
        client.newIndex("test");
        for (int i = 0; i < NUM_ACTIONS; i++) {
            client.index("test", "test", null, "{ \"name\" : \"" + randomString(32) + "\"}");
        }
        client.flushIngest();
        client.waitForResponses(TimeValue.timeValueSeconds(30));
    } catch (NoNodeAvailableException e) {
        logger.warn("skipping, no node available");
    } finally {
        assertEquals(numactions, client.getMetric().getSucceeded().getCount());
        if (client.hasThrowable()) {
            logger.error("error", client.getThrowable());
        }
        assertFalse(client.hasThrowable());
        client.shutdown();
    }
}

/**
 * With retry
 */
public static long countRequest(final String index, final QueryBuilder query, final String... types) {
    CountResponse response = withRetry(new ElasticsearchWithRetry<CountResponse>() {
        public CountResponse call(Client client) throws NoNodeAvailableException {
            CountRequestBuilder request_builder = new CountRequestBuilder(client);
            request_builder.setIndices(index);
            request_builder.setTypes(types);
            request_builder.setQuery(query);
            return request_builder.execute().actionGet();
        }
    });
    if (response == null) {
        return 0;
    }
    return response.getCount();
}

private void execute() {
    Loggers.ElasticSearch.debug("Prepare to update database with numberOfActions: " + bulk_request_builder.numberOfActions());

    final BulkRequest bu_r = bulk_request_builder.request();

    BulkResponse bulkresponse = null;

    bulkresponse = Elasticsearch.withRetry(new ElasticsearchWithRetry<BulkResponse>() {
        public BulkResponse call(Client client) throws NoNodeAvailableException {
            return client.bulk(bu_r).actionGet();
        }
    });

    if (bulkresponse != null) {
        if (bulkresponse.hasFailures()) {
            Loggers.ElasticSearch.error("Errors during update database: " + bulkresponse.buildFailureMessage());
        }
    }

    bulk_request_builder.request().requests().clear();

    if (on_push_callback != null) {
        on_push_callback.run();
    }
}

/**
 * create a new index. This method must be called to ensure that an elasticsearch index is available and can be used.
 * @param indexName
 * @param shards
 * @param replicas
 * @throws NoNodeAvailableException in case that no elasticsearch server can be contacted.
 */
public void createIndexIfNotExists(String indexName, final int shards, final int replicas) throws NoNodeAvailableException {
    // create an index if not existent
    if (!this.elasticsearchClient.admin().indices().prepareExists(indexName).execute().actionGet().isExists()) {
        Settings.Builder settings = Settings.builder()
                .put("number_of_shards", shards)
                .put("number_of_replicas", replicas);
        this.elasticsearchClient.admin().indices().prepareCreate(indexName)
            .setSettings(settings)
            .setUpdateAllTypes(true)
            .execute().actionGet();
    } else {
        //LOGGER.debug("Index with name {} already exists", indexName);
    }
}

@Override
public void performOn(ShardRouting replica, ReplicaRequest request, ActionListener<ReplicationOperation.ReplicaResponse> listener) {
    String nodeId = replica.currentNodeId();
    final DiscoveryNode node = clusterService.state().nodes().get(nodeId);
    if (node == null) {
        listener.onFailure(new NoNodeAvailableException("unknown node [" + nodeId + "]"));
        return;
    }
    final ConcreteShardRequest<ReplicaRequest> concreteShardRequest =
            new ConcreteShardRequest<>(request, replica.allocationId().getId());
    sendReplicaRequest(concreteShardRequest, node, listener);
}

@Test
public void portDefaultsTo9300() throws IOException {
  try (LazyClient lazyClient = new LazyClient(ElasticsearchStorage.builder()
      .hosts(asList("localhost")))) {

    assertThat(((NativeClient) lazyClient.get()).client.transportAddresses())
        .extracting(TransportAddress::getPort)
        .containsOnly(9300);

  } catch (NoNodeAvailableException e) {
    throw new AssumptionViolatedException(e.getMessage());
  }
}

@Test
public void check_failsInsteadOfThrowing() {
  CheckResult result =
      ElasticsearchStorage.builder().cluster("1.1.1.1").build().check();

  assertThat(result.ok).isFalse();
  assertThat(result.exception)
      .isInstanceOf(NoNodeAvailableException.class);
}

@Override
public void run() {
    while(true) {
        try {
            logger.debug("getting es cluster health.");
            ActionFuture<ClusterHealthResponse> healthFuture = transportClient.admin().cluster().health(Requests.clusterHealthRequest());
            ClusterHealthResponse healthResponse = healthFuture.get(5, TimeUnit.SECONDS);

            if(healthResponse.getStatus() == ClusterHealthStatus.RED){//es集群处于不健康状态给提示
                logger.error("elasticsearch info num of node:{}", healthResponse.getNumberOfNodes());
                    logger.error("elasticsearch info cluster health:{} ", healthResponse.getStatus());
            }

            isClusterOn.set(true);
        } catch(Throwable t) {
            if(t instanceof NoNodeAvailableException){//集群不可用
                logger.error("the cluster no node avaliable.");
                    isClusterOn.set(false);
                  }else{
                    isClusterOn.set(true);
                  }
        }
        try {
            Thread.sleep(3000);//FIXME
        } catch (InterruptedException ie) { 
            ie.printStackTrace(); 
        }
    }
}

@Override
public Set<String> getIndices() {
    try {
        return client.admin().indices().stats(new IndicesStatsRequest()).actionGet()
                .getIndices().keySet();
    } catch (NoNodeAvailableException e) {
        throw new ODataRuntimeException("Elasticsearch has no node available.", e);
    }
}

@Test
public void testSingleDocHttpClient() throws Exception {
    try (HttpClient client = HttpClient.builder()
            .url(new URL("http://127.0.0.1:9200"))
            .build()) {
        CreateIndexRequestBuilder createIndexRequestBuilder =
                new CreateIndexRequestBuilder(client, CreateIndexAction.INSTANCE).setIndex("test");
        createIndexRequestBuilder.execute().actionGet();
        IndexRequestBuilder indexRequestBuilder =
                new IndexRequestBuilder(client, IndexAction.INSTANCE)
                        .setIndex("test")
                        .setType("type")
                        .setId("1")
                        .setSource(jsonBuilder().startObject().field("name", "Hello World").endObject());
        indexRequestBuilder.execute().actionGet();
        RefreshRequestBuilder refreshRequestBuilder =
                new RefreshRequestBuilder(client, RefreshAction.INSTANCE)
                        .setIndices("test");
        refreshRequestBuilder.execute().actionGet();
        SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client, SearchAction.INSTANCE)
                .setIndices("test")
                .setQuery(QueryBuilders.matchAllQuery()).setSize(0);
        assertTrue(searchRequestBuilder.execute().actionGet().getHits().getTotalHits() > 0);
    } catch (NoNodeAvailableException e) {
        logger.warn("skipping, no node available");
    }
}

@Test
public void testRandomDocs() throws Exception {
    final HttpBulkClient client = HttpBulkClient.builder()
            .url(new URL("http://127.0.0.1:9200"))
            .maxActionsPerRequest(MAX_ACTIONS)
            .flushIngestInterval(TimeValue.timeValueSeconds(60))
            .build();
    try {
        client.newIndex("test");
        for (int i = 0; i < NUM_ACTIONS; i++) {
            client.index("test", "test", null, "{ \"name\" : \"" + randomString(32) + "\"}");
        }
        client.flushIngest();
        client.waitForResponses(TimeValue.timeValueSeconds(30));
        if (client.hasException()) {
            logger.error("error", client.getException());
        }
        assertFalse(client.hasException());
        client.refreshIndex("test");
        SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client.client(), SearchAction.INSTANCE)
                .setIndices("test")
                .setQuery(QueryBuilders.matchAllQuery()).setSize(0);
        assertTrue(searchRequestBuilder.execute().actionGet().getHits().getTotalHits() > 0) ;
    } catch (NoNodeAvailableException e) {
        logger.warn("skipping, no node available");
    } finally {
        client.shutdown();
    }
}

@Test
public void testPercolate() throws IOException
{
  try {
    registerPercolateQueries();
    checkPercolateResponse();
  } catch (NoNodeAvailableException e) {
    //This indicates that elasticsearch is not running on a particular machine.
    //Silently ignore in this case.
  }
}

@Test
public void testTransportClientSSLFail() throws Exception {
    thrown.expect(NoNodeAvailableException.class);

    final Settings settings = Settings.builder().put("searchguard.ssl.transport.enabled", true)
            .put(SSLConfigConstants.SEARCHGUARD_SSL_HTTP_ENABLE_OPENSSL_IF_AVAILABLE, allowOpenSSL)
            .put(SSLConfigConstants.SEARCHGUARD_SSL_TRANSPORT_ENABLE_OPENSSL_IF_AVAILABLE, allowOpenSSL)
            .put(SSLConfigConstants.SEARCHGUARD_SSL_TRANSPORT_KEYSTORE_ALIAS, "node-0")
            .put("searchguard.ssl.transport.keystore_filepath", getAbsoluteFilePathFromClassPath("node-0-keystore.jks"))
            .put("searchguard.ssl.transport.truststore_filepath", getAbsoluteFilePathFromClassPath("truststore.jks"))
            .put("searchguard.ssl.transport.enforce_hostname_verification", false)
            .put("searchguard.ssl.transport.resolve_hostname", false).build();

    startES(settings);

    final Settings tcSettings = Settings.builder().put("cluster.name", clustername)
            .put("path.home", getAbsoluteFilePathFromClassPath("node-0-keystore.jks").getParent())
            .put("searchguard.ssl.transport.keystore_filepath", getAbsoluteFilePathFromClassPath("node-0-keystore.jks"))
            .put("searchguard.ssl.transport.truststore_filepath", getAbsoluteFilePathFromClassPath("truststore_fail.jks"))
            .put("searchguard.ssl.transport.enforce_hostname_verification", false)
            .put("searchguard.ssl.transport.resolve_hostname", false).build();

    try (TransportClient tc = new TransportClientImpl(tcSettings, asCollection(SearchGuardSSLPlugin.class))) {
        tc.addTransportAddress(new TransportAddress(new InetSocketAddress(nodeHost, nodePort)));
        Assert.assertEquals(3, tc.admin().cluster().nodesInfo(new NodesInfoRequest()).actionGet().getNodes().size());
    }
}

public void testTransportClientWrongAuthentication() throws Exception {
    NodesInfoResponse nodeInfos = client().admin().cluster().prepareNodesInfo().get();
    List<NodeInfo> nodes = nodeInfos.getNodes();
    assertTrue(nodes.size() > 0);
    TransportAddress publishAddress = randomFrom(nodes).getTransport().address().publishAddress();
    String clusterName = nodeInfos.getClusterName().value();

    Settings settings;
    if (randomBoolean()) {
        settings = Settings.builder()
                .put("cluster.name", clusterName)
                .put(ThreadContext.PREFIX + "." + CustomRealm.USER_HEADER, randomFrom(KNOWN_USERS) + randomAlphaOfLength(1))
                .put(ThreadContext.PREFIX + "." + CustomRealm.PW_HEADER, PASSWORD)
                .build();
    } else {
        settings = Settings.builder()
                .put("cluster.name", clusterName)
                .put(ThreadContext.PREFIX + "." + CustomRealm.USER_HEADER, randomFrom(KNOWN_USERS))
                .put(ThreadContext.PREFIX + "." + CustomRealm.PW_HEADER, randomAlphaOfLengthBetween(16, 32))
                .build();
    }

    try (TransportClient client = new PreBuiltXPackTransportClient(settings)) {
        client.addTransportAddress(publishAddress);
        client.admin().cluster().prepareHealth().execute().actionGet();
        fail("authentication failure should have resulted in a NoNodesAvailableException");
    } catch (NoNodeAvailableException e) {
        // expected
    }
}

@BeforeClass
public static void setTestBehavior() {
    try {
        NodesInfoResponse response = client.admin().cluster().prepareNodesInfo().get();
        for (NodeInfo nodeInfo : response.getNodes()) {
            Version version = nodeInfo.getVersion();
            if (version.id >= 6000000) {
                supportsMultipleTypes = false;
            }
        }
    } catch (NoNodeAvailableException e) {
        assumeNoException(e);
    }
}

@Before
public void cleanCluster() {
    try {
        client.admin().indices().prepareDelete("_all").get();
    } catch (NoNodeAvailableException e) {
        assumeNoException(e);
    }
}

@Test
public void testShieldAuthentication()
{
    logger.info("Testing Shield Authentication");
    ESShieldServer shieldServer = new ESShieldServer(12000);
    Thread listener = new Thread(shieldServer);
    listener.start();
    logger.debug("Shield Authentication Listener Started");

    ESSearch search = new ESSearch(null, null, ESSearch.ES_MODE_HITS, "localhost", SHIELD_LISTENER_PORT,
            SHIELD_USER, SHIELD_PASSWORD, clusterName);

    logger.debug("Shield Authenticated Search Starting");
    try {
        search.search(getQuery("test-hits.json"));
    } catch(NoNodeAvailableException nonodeEx) {
        logger.debug("Client finally gave up - checking  for authentication header");
        try {
            listener.join();
            Assert.assertTrue("Client Passed Authentication in Request",
                    shieldServer.authenticated(SHIELD_USER, SHIELD_PASSWORD));
        } catch (InterruptedException intEx) {
            logger.error("Failed to wait for server to join: " + intEx.getLocalizedMessage());
        }
    } catch (Exception ex) {
        logger.error(ex);
        logger.error("Failed to test Shield Authentication: " + ex.getLocalizedMessage());
    } finally {
        search.close();
    }

}

public Result indexDelete() {
    try {
        elasticsearchService.deleteIndex();
        flash("info", "index gelöscht");
    } catch (NoNodeAvailableException nna) {
        flash("error", nna.getMessage());
    } catch (IndexNotFoundException infe) {
        flash("error", infe.getMessage());
    }

    return ok(indexing.render());
}

protected TransportClient getClient() throws EsServerException {
    if (_client == null) {
        try {
            connect();
            _client.admin().cluster().prepareState().execute();
        } catch (NoNodeAvailableException e) {
            throw new EsServerException();
        } catch (Throwable t) {
            System.out.println(t);
        }
    }
    return _client;
}

@Override
public SearchTransportClient init(Settings settings) throws IOException {
    super.createClient(settings);
    Collection<InetSocketTransportAddress> addrs = findAddresses(settings);
    if (!connect(addrs, settings.getAsBoolean("autodiscover", false))) {
        throw new NoNodeAvailableException("no cluster nodes available, check settings "
                + settings.getAsMap());
    }
    return this;
}

@Test
public void testRandomDocs() throws Exception {
    long numactions = NUM_ACTIONS;
    final HttpBulkNodeClient client = ClientBuilder.builder()
            .setMetric(new LongAdderIngestMetric())
            .put("host", "127.0.0.1")
            .put("port", 9200)
            .put(ClientBuilder.MAX_ACTIONS_PER_REQUEST, MAX_ACTIONS)
            .put(ClientBuilder.FLUSH_INTERVAL, TimeValue.timeValueSeconds(60))
            .toHttpBulkNodeClient();
    try {
        client.newIndex("test");
        for (int i = 0; i < NUM_ACTIONS; i++) {
            client.index("test", "test", null, "{ \"name\" : \"" + randomString(32) + "\"}");
        }
        client.flushIngest();
        client.waitForResponses(TimeValue.timeValueSeconds(30));
    } catch (NoNodeAvailableException e) {
        logger.warn("skipping, no node available");
    } finally {
        assertEquals(numactions, client.getMetric().getSucceeded().getCount());
        if (client.hasThrowable()) {
            logger.error("error", client.getThrowable());
        }
        assertFalse(client.hasThrowable());
        client.shutdown();
    }
}

@Test
public void testUpdateReplicaLevel() throws Exception {

    int numberOfShards = 2;
    int replicaLevel = 3;

    // we need 3 nodes for replica level 3
    startNode("2");
    startNode("3");

    int shardsAfterReplica;

    Settings settings = Settings.settingsBuilder()
            .put("index.number_of_shards", numberOfShards)
            .put("index.number_of_replicas", 0)
            .build();
    final IngestTransportClient ingest = ClientBuilder.builder()
            .put(getSettings())
            .setMetric(new LongAdderIngestMetric())
            .toIngestTransportClient();
    try {
        ingest.newIndex("replicatest", settings, null);
        ingest.waitForCluster("GREEN", TimeValue.timeValueSeconds(30));
        for (int i = 0; i < 12345; i++) {
            ingest.index("replicatest", "replicatest", null, "{ \"name\" : \"" + randomString(32) + "\"}");
        }
        ingest.flushIngest();
        ingest.waitForResponses(TimeValue.timeValueSeconds(30));
        shardsAfterReplica = ingest.updateReplicaLevel("replicatest", replicaLevel);
        assertEquals(shardsAfterReplica, numberOfShards * (replicaLevel + 1));
    } catch (NoNodeAvailableException e) {
        logger.warn("skipping, no node available");
    } finally {
        ingest.shutdown();
        if (ingest.hasThrowable()) {
            logger.error("error", ingest.getThrowable());
        }
        assertFalse(ingest.hasThrowable());
    }
}

@Test
public void testDuplicateDocIDs() throws Exception {
    long numactions = NUM_ACTIONS;
    final BulkTransportClient client = ClientBuilder.builder()
            .put(getSettings())
            .put(ClientBuilder.MAX_ACTIONS_PER_REQUEST, MAX_ACTIONS)
            .setMetric(new LongAdderIngestMetric())
            .toBulkTransportClient();
    try {
        client.newIndex("test");
        for (int i = 0; i < NUM_ACTIONS; i++) {
            client.index("test", "test", randomString(1), "{ \"name\" : \"" + randomString(32) + "\"}");
        }
        client.flushIngest();
        client.waitForResponses(TimeValue.timeValueSeconds(30));
        client.refreshIndex("test");
        SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client.client(), SearchAction.INSTANCE)
                .setIndices("test")
                .setTypes("test")
                .setQuery(matchAllQuery());
        long hits = searchRequestBuilder.execute().actionGet().getHits().getTotalHits();
        logger.info("hits = {}", hits);
        assertTrue(hits < NUM_ACTIONS);
    } catch (NoNodeAvailableException e) {
        logger.warn("skipping, no node available");
    } finally {
        client.shutdown();
        assertEquals(numactions, client.getMetric().getSucceeded().getCount());
        if (client.hasThrowable()) {
            logger.error("error", client.getThrowable());
        }
        assertFalse(client.hasThrowable());
    }
}

@Test
public void testDuplicateDocIDs() throws Exception {
    long numactions = NUM_ACTIONS;
    final BulkNodeClient client = ClientBuilder.builder()
            .put(ClientBuilder.MAX_ACTIONS_PER_REQUEST, MAX_ACTIONS)
            .setMetric(new LongAdderIngestMetric())
            .toBulkNodeClient(client("1"));
    try {
        client.newIndex("test");
        for (int i = 0; i < NUM_ACTIONS; i++) {
            client.index("test", "test", randomString(1), "{ \"name\" : \"" + randomString(32) + "\"}");
        }
        client.flushIngest();
        client.waitForResponses(TimeValue.timeValueSeconds(30));
        client.refreshIndex("test");
        SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client.client(), SearchAction.INSTANCE)
                .setIndices("test")
                .setTypes("test")
                .setQuery(matchAllQuery());
        long hits = searchRequestBuilder.execute().actionGet().getHits().getTotalHits();
        logger.info("hits = {}", hits);
        assertTrue(hits < NUM_ACTIONS);
    } catch (NoNodeAvailableException e) {
        logger.warn("skipping, no node available");
    } finally {
        client.shutdown();
        assertEquals(numactions, client.getMetric().getSucceeded().getCount());
        if (client.hasThrowable()) {
            logger.error("error", client.getThrowable());
        }
        assertFalse(client.hasThrowable());
    }
}

/**
 * Protected to some IndexMissingException
 */
public static void deleteIndexRequest(final String index_name) throws ElasticsearchException {
    try {
        Loggers.ElasticSearch.info("Delete index " + index_name);
        Elasticsearch.withRetry(new ElasticsearchWithRetry<Void>() {
            public Void call(Client client) throws NoNodeAvailableException {
                client.admin().indices().delete(new DeleteIndexRequest(index_name)).actionGet();
                return null;
            }
        });
    } catch (IndexMissingException e) {
    }
}

public static boolean isIndexExists(final String index_name) {
    return Elasticsearch.withRetry(new ElasticsearchWithRetry<Boolean>() {
        public Boolean call(Client client) throws NoNodeAvailableException {
            return client.admin().indices().exists(new IndicesExistsRequest(index_name)).actionGet().isExists();
        }
    });
}

public static boolean createIndex(final String index_name) {
    return Elasticsearch.withRetry(new ElasticsearchWithRetry<Boolean>() {
        public Boolean call(Client client) throws NoNodeAvailableException {
            return client.admin().indices().prepareCreate(index_name).execute().actionGet().isAcknowledged();
        }
    });
}

public static boolean addMappingToIndex(final String index_name, final String type, final String json_mapping_source) {
    Loggers.ElasticSearch.info("Add mapping to index: " + index_name + ", type: " + type + ", mapping_source: " + json_mapping_source);
    return Elasticsearch.withRetry(new ElasticsearchWithRetry<Boolean>() {
        public Boolean call(Client client) throws NoNodeAvailableException {
            // Inspired by http://stackoverflow.com/questions/22071198/adding-mapping-to-a-type-from-java-how-do-i-do-it
            PutMappingRequestBuilder request = client.admin().indices().preparePutMapping(index_name);
            request.setType(type);
            request.setSource(json_mapping_source);
            return request.execute().actionGet().isAcknowledged();
        }
    });
}

static <T> T withRetry(ElasticsearchWithRetry<T> callable) throws NoNodeAvailableException {
    for (int pos_retry = 0; pos_retry < max_retry; pos_retry++) {
        try {
            return callable.call(getClient());
        } catch (NoNodeAvailableException e) {
            try {
                /**
                 * Wait before to retry, after the 2nd try.
                 */
                Thread.sleep(pos_retry * 100);
            } catch (InterruptedException e1) {
                Loggers.ElasticSearch.warn("Stop sleep", e1);
                return null;
            }
            if (pos_retry == (max_retry - 2)) {
                /**
                 * Before the last try, force refesh configuration.
                 */
                Elasticsearch.refeshconfiguration();
            } else if (pos_retry + 1 == max_retry) {
                /**
                 * The last try has failed, throw error.
                 */
                Loggers.ElasticSearch.error("The last (" + max_retry + ") try has failed, throw error", e);
                throw e;
            }
        }
    }
    return null;
}

/**
 * With retry
 */
public static GetResponse get(final GetRequest request) {
    return withRetry(new ElasticsearchWithRetry<GetResponse>() {
        public GetResponse call(Client client) throws NoNodeAvailableException {
            return client.get(request).actionGet();
        }
    });
}

/**
 * With retry
 */
public static IndexResponse index(final IndexRequest request) {
    return withRetry(new ElasticsearchWithRetry<IndexResponse>() {
        public IndexResponse call(Client client) throws NoNodeAvailableException {
            return client.index(request).actionGet();
        }
    });
}

public void testReplicaProxy() throws InterruptedException, ExecutionException {
    ReplicationOperation.Replicas proxy = action.newReplicasProxy();
    final String index = "test";
    final ShardId shardId = new ShardId(index, "_na_", 0);
    ClusterState state = stateWithActivePrimary(index, true, 1 + randomInt(3), randomInt(2));
    logger.info("using state: {}", state);
    setState(clusterService, state);

    // check that at unknown node fails
    PlainActionFuture<ReplicaResponse> listener = new PlainActionFuture<>();
    proxy.performOn(
        TestShardRouting.newShardRouting(shardId, "NOT THERE", false, randomFrom(ShardRoutingState.values())),
        new Request(), listener);
    assertTrue(listener.isDone());
    assertListenerThrows("non existent node should throw a NoNodeAvailableException", listener, NoNodeAvailableException.class);

    final IndexShardRoutingTable shardRoutings = state.routingTable().shardRoutingTable(shardId);
    final ShardRouting replica = randomFrom(shardRoutings.replicaShards().stream()
        .filter(ShardRouting::assignedToNode).collect(Collectors.toList()));
    listener = new PlainActionFuture<>();
    proxy.performOn(replica, new Request(), listener);
    assertFalse(listener.isDone());

    CapturingTransport.CapturedRequest[] captures = transport.getCapturedRequestsAndClear();
    assertThat(captures, arrayWithSize(1));
    if (randomBoolean()) {
        final TransportReplicationAction.ReplicaResponse response =
            new TransportReplicationAction.ReplicaResponse(randomAsciiOfLength(10), randomLong());
        transport.handleResponse(captures[0].requestId, response);
        assertTrue(listener.isDone());
        assertThat(listener.get(), equalTo(response));
    } else if (randomBoolean()) {
        transport.handleRemoteError(captures[0].requestId, new ElasticsearchException("simulated"));
        assertTrue(listener.isDone());
        assertListenerThrows("listener should reflect remote error", listener, ElasticsearchException.class);
    } else {
        transport.handleError(captures[0].requestId, new TransportException("simulated"));
        assertTrue(listener.isDone());
        assertListenerThrows("listener should reflect remote error", listener, TransportException.class);
    }

    AtomicReference<Object> failure = new AtomicReference<>();
    AtomicReference<Object> ignoredFailure = new AtomicReference<>();
    AtomicBoolean success = new AtomicBoolean();
    proxy.failShardIfNeeded(replica, randomIntBetween(1, 10), "test", new ElasticsearchException("simulated"),
            () -> success.set(true), failure::set, ignoredFailure::set
    );
    CapturingTransport.CapturedRequest[] shardFailedRequests = transport.getCapturedRequestsAndClear();
    // A replication action doesn't not fail the request
    assertEquals(0, shardFailedRequests.length);
}