This repository has been archived by the owner on May 18, 2020. It is now read-only.
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Support for Elasticsearch 2.x and 5.x (#9)
* Support for Elasticsearch 2.x and 5.x Besides: - got rid of elasticsearch dependency - few changes in API - plugins are now installed using elasticsearch-plugin/plugin tool - cluster name and tcp transport port are not set by default - configurable startup timeout & higher timeout in tests - possibility to set ES_JAVA_OPTS variable (that let me fix build problem on Travis) - updated README - added methods returning port numbers used by elasticsearch
- Loading branch information
Showing
42 changed files
with
966 additions
and
485 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,5 +1,5 @@ | ||
| --- | ||
| language: java | ||
| script: ./gradlew clean check | ||
| script: ./gradlew clean check --info | ||
| jdk: | ||
| - oraclejdk8 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| dependencies { | ||
| compile group: 'org.slf4j', name: 'slf4j-api', version: '1.7.21' | ||
| compile group: 'commons-io', name: 'commons-io', version: '2.5' | ||
| compile group: 'net.lingala.zip4j', name: 'zip4j', version: '1.3.2' | ||
| compile group: 'org.apache.commons', name: 'commons-lang3', version: '3.4' | ||
| compile group: 'org.apache.httpcomponents', name: 'httpclient', version: '4.5.2' | ||
| compile group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: '2.6.2' | ||
| compile group: 'com.fasterxml.jackson.dataformat', name: 'jackson-dataformat-yaml', version: '2.6.2' | ||
|
|
||
| testCompile group: 'org.codehaus.groovy', name: 'groovy', version: '2.4.6' | ||
| testCompile group: 'org.spockframework', name: 'spock-core', version: '1.0-groovy-2.4' | ||
| testCompile group: 'ch.qos.logback', name: 'logback-classic', version: '1.1.7' | ||
| testCompile group: 'org.skyscreamer', name: 'jsonassert', version: '1.3.0' | ||
| } |
File renamed without changes.
33 changes: 21 additions & 12 deletions
33
...lasticsearch/ElasticDownloadUrlUtils.java → ...lasticsearch/ElasticDownloadUrlUtils.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
157 changes: 157 additions & 0 deletions
157
core/src/main/java/pl/allegro/tech/embeddedelasticsearch/ElasticRestClient.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,157 @@ | ||
| package pl.allegro.tech.embeddedelasticsearch; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Collection; | ||
| import java.util.List; | ||
| import java.util.Optional; | ||
| import java.util.stream.Stream; | ||
| import java.util.stream.StreamSupport; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import org.apache.commons.io.IOUtils; | ||
| import org.apache.http.client.methods.CloseableHttpResponse; | ||
| import org.apache.http.client.methods.HttpDelete; | ||
| import org.apache.http.client.methods.HttpGet; | ||
| import org.apache.http.client.methods.HttpHead; | ||
| import org.apache.http.client.methods.HttpPost; | ||
| import org.apache.http.client.methods.HttpPut; | ||
| import org.apache.http.entity.StringEntity; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import static java.nio.charset.StandardCharsets.UTF_8; | ||
| import static java.util.stream.Collectors.joining; | ||
| import static java.util.stream.Collectors.toList; | ||
| import static org.apache.http.entity.ContentType.APPLICATION_JSON; | ||
| import static pl.allegro.tech.embeddedelasticsearch.HttpStatusCodes.OK; | ||
|
|
||
| class ElasticRestClient { | ||
|
|
||
| private static final Logger logger = LoggerFactory.getLogger(ElasticRestClient.class); | ||
|
|
||
| private int elasticsearchHttpPort; | ||
| private final HttpClient httpClient; | ||
| private final IndicesDescription indicesDescription; | ||
|
|
||
| ElasticRestClient(int elasticsearchHttpPort, HttpClient httpClient, IndicesDescription indicesDescription) { | ||
| this.elasticsearchHttpPort = elasticsearchHttpPort; | ||
| this.httpClient = httpClient; | ||
| this.indicesDescription = indicesDescription; | ||
| } | ||
|
|
||
| void createIndices() { | ||
| indicesDescription.getIndicesNames().forEach(this::createIndex); | ||
| } | ||
|
|
||
| void createIndex(String indexName) { | ||
| if (!indexExists(indexName)) { | ||
| HttpPut request = new HttpPut(url("/" + indexName)); | ||
| request.setEntity(new StringEntity(indicesDescription.getIndexSettings(indexName).toJson().toString(), APPLICATION_JSON)); | ||
| CloseableHttpResponse response = httpClient.execute(request); | ||
| if (response.getStatusLine().getStatusCode() != 200) { | ||
| String responseBody = readBodySafely(response); | ||
| throw new RuntimeException("Call to elasticsearch resulted in error:\n" + responseBody); | ||
| } | ||
| waitForClusterYellow(); | ||
| } | ||
| } | ||
|
|
||
| private boolean indexExists(String indexName) { | ||
| HttpHead request = new HttpHead(url("/" + indexName)); | ||
| CloseableHttpResponse response = httpClient.execute(request); | ||
| return response.getStatusLine().getStatusCode() == OK; | ||
| } | ||
|
|
||
| private void waitForClusterYellow() { | ||
| HttpGet request = new HttpGet(url("/_cluster/health?wait_for_status=yellow&timeout=60s")); | ||
| CloseableHttpResponse response = httpClient.execute(request); | ||
| assertOk(response, "Cluster does not reached yellow status in specified timeout"); | ||
| } | ||
|
|
||
| void deleteIndices() { | ||
| indicesDescription.getIndicesNames().forEach(this::deleteIndex); | ||
| } | ||
|
|
||
| void deleteIndex(String indexName) { | ||
| if (indexExists(indexName)) { | ||
| HttpDelete request = new HttpDelete(url("/" + indexName)); | ||
| assertOk(httpClient.execute(request), "Delete request resulted in error"); | ||
| waitForClusterYellow(); | ||
| } else { | ||
| logger.warn("Index: {} does not exists so cannot be removed", indexName); | ||
| } | ||
| } | ||
|
|
||
| void indexWithIds(String indexName, String indexType, Collection<DocumentWithId> idJsonMap) { | ||
| String bulkRequestBody = idJsonMap.stream() | ||
| .flatMap(json -> Stream.of(indexMetadataJsonWithId(json.getId()), json.getDocument())) | ||
| .map((jsonNodes) -> jsonNodes.replace('\n', ' ').replace('\r', ' ')) | ||
| .collect(joining("\n")) + "\n"; | ||
|
|
||
| HttpPost request = new HttpPost(url("/" + indexName + "/" + indexType + "/_bulk")); | ||
| request.setEntity(new StringEntity(bulkRequestBody, UTF_8)); | ||
| CloseableHttpResponse response = httpClient.execute(request); | ||
| assertOk(response, "Request finished with error"); | ||
| refresh(); | ||
| } | ||
|
|
||
| private String indexMetadataJsonWithId(String id) { | ||
| return "{ \"index\": { \"_id\": \"" + id + "\"} }"; | ||
| } | ||
|
|
||
| void refresh() { | ||
| HttpPost request = new HttpPost(url("/_refresh")); | ||
| httpClient.execute(request); | ||
| } | ||
|
|
||
| private String url(String path) { | ||
| return "http://localhost:" + elasticsearchHttpPort + path; | ||
| } | ||
|
|
||
| private void assertOk(CloseableHttpResponse response, String message) { | ||
| if (response.getStatusLine().getStatusCode() != OK) { | ||
| throw new IllegalStateException(message + "\nResponse body:\n" + readBodySafely(response)); | ||
| } | ||
| } | ||
|
|
||
| private String readBodySafely(CloseableHttpResponse response) { | ||
| try { | ||
| return IOUtils.toString(response.getEntity().getContent(), UTF_8); | ||
| } catch (IOException e) { | ||
| logger.error("Error during reading response body", e); | ||
| return ""; | ||
| } | ||
| } | ||
|
|
||
| List<String> fetchAllDocuments(String... indices) { | ||
| if (indices.length == 0) { | ||
| return searchForDocuments(Optional.empty()).collect(toList()); | ||
| } else { | ||
| return Stream.of(indices) | ||
| .flatMap((index) -> searchForDocuments(Optional.of(index))) | ||
| .collect(toList()); | ||
| } | ||
| } | ||
|
|
||
| private Stream<String> searchForDocuments(Optional<String> indexMaybe) { | ||
| String searchCommand = indexMaybe | ||
| .map(index -> "/" + index + "/_search") | ||
| .orElse("/_search"); | ||
| HttpGet request = new HttpGet(url(searchCommand)); | ||
|
|
||
| CloseableHttpResponse response = httpClient.execute(request); | ||
| assertOk(response, "Error during search (" + searchCommand + ")"); | ||
| String body = readBodySafely(response); | ||
|
|
||
| ObjectMapper objectMapper = new ObjectMapper(); | ||
| try { | ||
| JsonNode jsonNode = objectMapper.readTree(body); | ||
| return StreamSupport.stream(jsonNode.get("hits").get("hits").spliterator(), false) | ||
| .map(hitNode -> hitNode.get("_source")) | ||
| .map(JsonNode::toString); | ||
| } catch (IOException e) { | ||
| throw new RuntimeException(e); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.