ElasticCatClient implements Closeable { @SuppressFBWarnings("RCN_REDUNDANT_NULLCHECK_WOULD_HAVE_BEEN_A_NPE") public List<Shard> shards(String... indices) { Map<String, String> idToAddress = nodes().stream().collect(toMap(Node::getId, Node::getHttpAddress)); try { Map<String, String> params = new HashMap<>(); params.put("format", "json"); params.put("h", "id,index,shard,prirep,docs,state,ip,node"); Response res = withRetry( () -> client.performRequest("GET", "/_cat/shards/" + String.join(",", indices), params), retries ); try (InputStreamReader reader = new InputStreamReader(res.getEntity().getContent(), UTF_8)) { JsonArray array = Json.parse(reader).asArray(); List<Shard> shards = new ArrayList<>(array.size()); for (JsonValue value : array) { Optional<Shard> shard = convertToShard(value, idToAddress); shard.ifPresent(shards::add); } LOG.log(FINE, "Shards " + shards); return shards; } } catch (IOException e) { throw new JetException("Could not get ES shards", e); } } ElasticCatClient(RestClient client, int retries); Master master(); List<Node> nodes(); @SuppressFBWarnings("RCN_REDUNDANT_NULLCHECK_WOULD_HAVE_BEEN_A_NPE") List<Shard> shards(String... indices); @Override void close();  }
@Test public void shards() throws IOException { ElasticCatClient catClient = new ElasticCatClient(restClient, 5); Response nodesResponse = response("es2node_nodes.json"); Response shardsResponse = response("es2node_shards.json"); when(restClient.performRequest(any(), any(), any(Map.class))) .thenReturn(nodesResponse, shardsResponse); List<Shard> shards = catClient.shards("my-index"); assertThat(shards).extracting(Shard::getHttpAddress) .containsOnly("127.0.0.1:9200", "127.0.0.1:9201"); }
@Test public void shouldRetryOnShards() throws IOException { ElasticCatClient catClient = new ElasticCatClient(restClient, 2); Response nodesResponse = response("es2node_nodes.json"); Response shardsResponse = response("es2node_shards.json"); when(restClient.performRequest(any(), any(), any(Map.class))) .thenThrow(new IOException("Could not connect")) .thenReturn(nodesResponse, shardsResponse); List<Shard> shards = catClient.shards("my-index"); assertThat(shards).extracting(Shard::getHttpAddress) .containsOnly("127.0.0.1:9200", "127.0.0.1:9201"); }
@Test public void shouldFailAfterFiveAttemptsShards() throws IOException { ElasticCatClient catClient = new ElasticCatClient(restClient, 2); when(restClient.performRequest(any(), any(), any(Map.class))) .thenThrow(new IOException("Could not connect")); assertThatThrownBy(() -> catClient.shards("my-index")) .isInstanceOf(JetException.class) .hasRootCauseInstanceOf(IOException.class) .hasRootCauseMessage("Could not connect"); }