joerg/elx
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

new elx implementation for Elasticsearch 7.6.1

Browse source
This commit is contained in:
Jörg Prante 2020-05-10 17:40:58 +02:00
parent 3cc4881f19
commit a5c8dc8b5f
56 changed files with 546 additions and 661 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
build.gradle
View file

@ -7,7 +7,7 @@ plugins {
} }
if (JavaVersion.current() < JavaVersion.VERSION_11) { if (JavaVersion.current() < JavaVersion.VERSION_11) {
throw new GradleException("This build must be run with java 11") throw new GradleException("This build must be run with Java/OpenJDK 11+")
} }
subprojects { subprojects {
@ -50,16 +50,15 @@ subprojects {
test { test {
enabled = true enabled = true
useJUnitPlatform() useJUnitPlatform()
doFirst { // for Lucene to access jdk.internal.ref and jdk.internal.misc in Java 11+
jvmArgs = [ jvmArgs = [
'--add-exports=java.base/jdk.internal.ref=ALL-UNNAMED', '--add-exports=java.base/jdk.internal.ref=ALL-UNNAMED',
'--add-exports=java.base/jdk.internal.misc=ALL-UNNAMED', '--add-exports=java.base/jdk.internal.misc=ALL-UNNAMED',
'--add-opens=java.base/java.nio=ALL-UNNAMED' '--add-opens=java.base/java.nio=ALL-UNNAMED'
] ]
}
systemProperty 'java.util.logging.manager', 'org.apache.logging.log4j.jul.LogManager' systemProperty 'java.util.logging.manager', 'org.apache.logging.log4j.jul.LogManager'
systemProperty 'path.home', "${project.buildDir}/"
systemProperty 'jna.debug_load', 'true' systemProperty 'jna.debug_load', 'true'
systemProperty 'path.home', "${project.buildDir}/"
failFast = true failFast = true
testLogging { testLogging {
events 'PASSED', 'FAILED', 'SKIPPED' events 'PASSED', 'FAILED', 'SKIPPED'

2
elx-api/build.gradle
View file

@ -1,4 +1,4 @@
dependencies { dependencies {
compile "org.xbib:metrics-common:${project.property('xbib-metrics.version')}" compile "org.xbib:metrics-common:${project.property('xbib-metrics.version')}"
compile "org.xbib.elasticsearch:elasticsearch:${rootProject.property('elasticsearch-server.version')}" compile "org.elasticsearch:elasticsearch:${rootProject.property('elasticsearch.version')}"
} }

2
elx-api/src/main/java/org/xbib/elx/api/BulkProcessor.java
View file

@ -12,8 +12,6 @@ public interface BulkProcessor extends Closeable, Flushable {
BulkProcessor add(ActionRequest request); BulkProcessor add(ActionRequest request);
BulkProcessor add(ActionRequest request, Object payload);
boolean awaitFlush(long timeout, TimeUnit unit) throws InterruptedException; boolean awaitFlush(long timeout, TimeUnit unit) throws InterruptedException;
boolean awaitClose(long timeout, TimeUnit unit) throws InterruptedException; boolean awaitClose(long timeout, TimeUnit unit) throws InterruptedException;

17
elx-api/src/main/java/org/xbib/elx/api/ExtendedClient.java
View file

@ -6,6 +6,7 @@ import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.ElasticsearchClient; import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import java.io.Closeable; import java.io.Closeable;
import java.io.Flushable; import java.io.Flushable;
@ -126,8 +127,9 @@ public interface ExtendedClient extends Flushable, Closeable {
* @param id the id * @param id the id
* @param source the source * @param source the source
* @return this * @return this
* @throws IOException if update fails
*/ */
ExtendedClient update(String index, String id, BytesReference source); ExtendedClient update(String index, String id, BytesReference source) throws IOException;
/** /**
* Update document. Use with precaution! Does not work in all cases. * Update document. Use with precaution! Does not work in all cases.
@ -199,7 +201,18 @@ public interface ExtendedClient extends Flushable, Closeable {
* @return this * @return this
* @throws IOException if settings/mapping is invalid or index creation fails * @throws IOException if settings/mapping is invalid or index creation fails
*/ */
ExtendedClient newIndex(String index, Settings settings, Map<String, Object> mapping) throws IOException; ExtendedClient newIndex(String index, Settings settings, XContentBuilder mapping) throws IOException;
/**
* Create a new index.
*
* @param index index
* @param settings settings
* @param mapping mapping
* @return this
* @throws IOException if settings/mapping is invalid or index creation fails
*/
ExtendedClient newIndex(String index, Settings settings, Map<String, ?> mapping) throws IOException;
/** /**
* Create a new index. * Create a new index.

5
elx-common/build.gradle
View file

@ -1,5 +1,6 @@
dependencies{ dependencies{
compile project(':elx-api') compile project(':elx-api')
testCompile "org.xbib.elasticsearch:elasticsearch-analysis-common:${rootProject.property('elasticsearch-server.version')}" testCompile "org.elasticsearch.plugin:transport-netty4-client:${rootProject.property('elasticsearch.version')}"
testCompile "org.xbib.elasticsearch:transport-netty4:${rootProject.property('elasticsearch-server.version')}" testCompile "io.netty:netty-codec-http:${project.property('netty.version')}"
testCompile "io.netty:netty-transport:${project.property('netty.version')}"
} }

189
elx-common/src/main/java/org/xbib/elx/common/AbstractExtendedClient.java
View file

@ -5,6 +5,7 @@ import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import org.apache.logging.log4j.Level; import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.apache.lucene.search.TotalHits;
import org.elasticsearch.ElasticsearchTimeoutException; import org.elasticsearch.ElasticsearchTimeoutException;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthAction; import org.elasticsearch.action.admin.cluster.health.ClusterHealthAction;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest; import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
@ -14,7 +15,6 @@ import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse; import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesAction; import org.elasticsearch.action.admin.indices.alias.IndicesAliasesAction;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest; import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesResponse;
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesAction; import org.elasticsearch.action.admin.indices.alias.get.GetAliasesAction;
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest; import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest;
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesResponse; import org.elasticsearch.action.admin.indices.alias.get.GetAliasesResponse;
@ -23,7 +23,6 @@ import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse; import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexAction; import org.elasticsearch.action.admin.indices.delete.DeleteIndexAction;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest; import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.admin.indices.flush.FlushAction; import org.elasticsearch.action.admin.indices.flush.FlushAction;
import org.elasticsearch.action.admin.indices.flush.FlushRequest; import org.elasticsearch.action.admin.indices.flush.FlushRequest;
import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeAction; import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeAction;
@ -46,6 +45,7 @@ import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchAction; import org.elasticsearch.action.search.SearchAction;
import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest; import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.ElasticsearchClient; import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.client.transport.NoNodeAvailableException; import org.elasticsearch.client.transport.NoNodeAvailableException;
@ -53,7 +53,6 @@ import org.elasticsearch.cluster.health.ClusterHealthStatus;
import org.elasticsearch.cluster.metadata.AliasMetaData; import org.elasticsearch.cluster.metadata.AliasMetaData;
import org.elasticsearch.cluster.metadata.AliasOrIndex; import org.elasticsearch.cluster.metadata.AliasOrIndex;
import org.elasticsearch.cluster.metadata.MappingMetaData; import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.collect.ImmutableOpenMap; import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
@ -89,8 +88,8 @@ import java.time.LocalDate;
import java.time.ZoneId; import java.time.ZoneId;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections; import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap; import java.util.HashMap;
import java.util.LinkedHashMap; import java.util.LinkedHashMap;
import java.util.List; import java.util.List;
@ -113,22 +112,13 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
private static final Logger logger = LogManager.getLogger(AbstractExtendedClient.class.getName()); private static final Logger logger = LogManager.getLogger(AbstractExtendedClient.class.getName());
/**
* The one and only index type name used in the extended client.
* Notr that all Elasticsearch version < 6.2.0 do not allow a prepending "_".
*/
private static final String TYPE_NAME = "doc";
/**
* The Elasticsearch client.
*/
private ElasticsearchClient client; private ElasticsearchClient client;
private BulkMetric bulkMetric; private BulkMetric bulkMetric;
private BulkController bulkController; private BulkController bulkController;
private AtomicBoolean closed; private final AtomicBoolean closed;
private static final IndexShiftResult EMPTY_INDEX_SHIFT_RESULT = new IndexShiftResult() { private static final IndexShiftResult EMPTY_INDEX_SHIFT_RESULT = new IndexShiftResult() {
@Override @Override
@ -195,7 +185,7 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
@Override @Override
public AbstractExtendedClient init(Settings settings) throws IOException { public AbstractExtendedClient init(Settings settings) throws IOException {
logger.log(Level.INFO, "initializing with settings = " + settings.toDelimitedString(',')); logger.info("initializing with settings = " + settings.toDelimitedString(','));
if (client == null) { if (client == null) {
client = createClient(settings); client = createClient(settings);
} }
@ -303,7 +293,7 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
@Override @Override
public ExtendedClient newIndex(String index, Settings settings) throws IOException { public ExtendedClient newIndex(String index, Settings settings) throws IOException {
return newIndex(index, settings, (Map<String, Object>) null); return newIndex(index, settings, (Map<String, ?>) null);
} }
@Override @Override
@ -314,7 +304,7 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
} }
@Override @Override
public ExtendedClient newIndex(String index, Settings settings, Map<String, Object> mapping) throws IOException { public ExtendedClient newIndex(String index, Settings settings, XContentBuilder mapping) throws IOException {
ensureActive(); ensureActive();
if (index == null) { if (index == null) {
logger.warn("no index name given to create index"); logger.warn("no index name given to create index");
@ -325,13 +315,37 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
createIndexRequest.settings(settings); createIndexRequest.settings(settings);
} }
if (mapping != null) { if (mapping != null) {
createIndexRequest.mapping(TYPE_NAME, mapping); createIndexRequest.mapping("_doc", mapping);
} }
CreateIndexResponse createIndexResponse = client.execute(CreateIndexAction.INSTANCE, createIndexRequest).actionGet(); CreateIndexResponse createIndexResponse = client.execute(CreateIndexAction.INSTANCE, createIndexRequest).actionGet();
XContentBuilder builder = XContentFactory.jsonBuilder(); if (createIndexResponse.isAcknowledged()) {
logger.info("index {} created: {}", index, return this;
Strings.toString(createIndexResponse.toXContent(builder, ToXContent.EMPTY_PARAMS))); }
return this; throw new IllegalStateException("index creation not acknowledged");
}
@Override
public ExtendedClient newIndex(String index, Settings settings, Map<String, ?> mapping) throws IOException {
ensureActive();
if (index == null) {
logger.warn("no index name given to create index");
return this;
}
CreateIndexRequest createIndexRequest = new CreateIndexRequest().index(index);
if (settings != null) {
createIndexRequest.settings(settings);
}
if (mapping != null) {
if (mapping.size() != 1) {
throw new IllegalArgumentException("mapping invalid, just use 'doc' for mapping");
}
createIndexRequest.mapping("_doc", mapping);
}
CreateIndexResponse createIndexResponse = client.execute(CreateIndexAction.INSTANCE, createIndexRequest).actionGet();
if (createIndexResponse.isAcknowledged()) {
return this;
}
throw new IllegalStateException("index creation not acknowledged");
} }
@Override @Override
@ -387,13 +401,13 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
@Override @Override
public ExtendedClient index(String index, String id, boolean create, String source) { public ExtendedClient index(String index, String id, boolean create, String source) {
return index(new IndexRequest(index, TYPE_NAME, id).create(create) return index(new IndexRequest().index(index).id(id).create(create)
.source(source.getBytes(StandardCharsets.UTF_8), XContentType.JSON)); .source(source.getBytes(StandardCharsets.UTF_8), XContentType.JSON));
} }
@Override @Override
public ExtendedClient index(String index, String id, boolean create, BytesReference source) { public ExtendedClient index(String index, String id, boolean create, BytesReference source) {
return index(new IndexRequest(index, TYPE_NAME, id).create(create) return index(new IndexRequest().index(index).id(id).create(create)
.source(source, XContentType.JSON)); .source(source, XContentType.JSON));
} }
@ -406,7 +420,7 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
@Override @Override
public ExtendedClient delete(String index, String id) { public ExtendedClient delete(String index, String id) {
return delete(new DeleteRequest(index, TYPE_NAME, id)); return delete(new DeleteRequest().index(index).id(id));
} }
@Override @Override
@ -417,14 +431,14 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
} }
@Override @Override
public ExtendedClient update(String index, String id, BytesReference source) { public ExtendedClient update(String index, String id, BytesReference source) throws IOException {
return update(new UpdateRequest(index, TYPE_NAME, id) return update(new UpdateRequest().index(index).id(id)
.doc(source, XContentType.JSON)); .doc(source, XContentType.JSON));
} }
@Override @Override
public ExtendedClient update(String index, String id, String source) { public ExtendedClient update(String index, String id, String source) {
return update(new UpdateRequest(index, TYPE_NAME, id) return update(new UpdateRequest().index(index).id(id)
.doc(source.getBytes(StandardCharsets.UTF_8), XContentType.JSON)); .doc(source.getBytes(StandardCharsets.UTF_8), XContentType.JSON));
} }
@ -458,7 +472,7 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
ClusterHealthResponse healthResponse = ClusterHealthResponse healthResponse =
client.execute(ClusterHealthAction.INSTANCE, clusterHealthRequest).actionGet(); client.execute(ClusterHealthAction.INSTANCE, clusterHealthRequest).actionGet();
if (healthResponse != null && healthResponse.isTimedOut()) { if (healthResponse != null && healthResponse.isTimedOut()) {
logger.error("timeout waiting for recovery"); logger.warn("timeout waiting for recovery");
return false; return false;
} }
} }
@ -473,9 +487,7 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
ClusterHealthResponse healthResponse = client.execute(ClusterHealthAction.INSTANCE, ClusterHealthResponse healthResponse = client.execute(ClusterHealthAction.INSTANCE,
new ClusterHealthRequest().timeout(timeout).waitForStatus(status)).actionGet(); new ClusterHealthRequest().timeout(timeout).waitForStatus(status)).actionGet();
if (healthResponse != null && healthResponse.isTimedOut()) { if (healthResponse != null && healthResponse.isTimedOut()) {
if (logger.isErrorEnabled()) { logger.warn("timeout, cluster state is " + healthResponse.getStatus().name() + " and not " + status.name());
logger.error("timeout, cluster state is " + healthResponse.getStatus().name() + " and not " + status.name());
}
return false; return false;
} }
return true; return true;
@ -693,7 +705,7 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
} }
if (!indicesAliasesRequest.getAliasActions().isEmpty()) { if (!indicesAliasesRequest.getAliasActions().isEmpty()) {
logger.debug("indices alias request = {}", indicesAliasesRequest.getAliasActions().toString()); logger.debug("indices alias request = {}", indicesAliasesRequest.getAliasActions().toString());
IndicesAliasesResponse indicesAliasesResponse = AcknowledgedResponse indicesAliasesResponse =
client.execute(IndicesAliasesAction.INSTANCE, indicesAliasesRequest).actionGet(); client.execute(IndicesAliasesAction.INSTANCE, indicesAliasesRequest).actionGet();
logger.debug("response isAcknowledged = {} isFragment = {}", logger.debug("response isAcknowledged = {} isFragment = {}",
indicesAliasesResponse.isAcknowledged(), indicesAliasesResponse.isFragment()); indicesAliasesResponse.isAcknowledged(), indicesAliasesResponse.isFragment());
@ -719,7 +731,7 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
GetIndexRequestBuilder getIndexRequestBuilder = new GetIndexRequestBuilder(client, GetIndexAction.INSTANCE); GetIndexRequestBuilder getIndexRequestBuilder = new GetIndexRequestBuilder(client, GetIndexAction.INSTANCE);
GetIndexResponse getIndexResponse = getIndexRequestBuilder.execute().actionGet(); GetIndexResponse getIndexResponse = getIndexRequestBuilder.execute().actionGet();
Pattern pattern = Pattern.compile("^(.*?)(\\d+)$"); Pattern pattern = Pattern.compile("^(.*?)(\\d+)$");
logger.info("{} indices", getIndexResponse.getIndices().length); logger.info("pruneIndex: total of {} indices", getIndexResponse.getIndices().length);
List<String> candidateIndices = new ArrayList<>(); List<String> candidateIndices = new ArrayList<>();
for (String s : getIndexResponse.getIndices()) { for (String s : getIndexResponse.getIndices()) {
Matcher m = pattern.matcher(s); Matcher m = pattern.matcher(s);
@ -754,21 +766,30 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
String[] s = new String[indicesToDelete.size()]; String[] s = new String[indicesToDelete.size()];
DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest() DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest()
.indices(indicesToDelete.toArray(s)); .indices(indicesToDelete.toArray(s));
DeleteIndexResponse response = client.execute(DeleteIndexAction.INSTANCE, deleteIndexRequest).actionGet(); AcknowledgedResponse response = client.execute(DeleteIndexAction.INSTANCE, deleteIndexRequest).actionGet();
return new SuccessPruneResult(candidateIndices, indicesToDelete, response); if( response.isAcknowledged()) {
logger.log(Level.INFO, "deletion of {} acknowledged, waiting for GREEN", Arrays.asList(s));
waitForCluster("GREEN", 30L, TimeUnit.SECONDS);
return new SuccessPruneResult(candidateIndices, indicesToDelete, response);
} else {
logger.log(Level.WARN, "deletion of {} not acknowledged", Arrays.asList(s));
return new FailPruneResult(candidateIndices, indicesToDelete, response);
}
} }
@Override @Override
public Long mostRecentDocument(String index, String timestampfieldname) { public Long mostRecentDocument(String index, String timestampfieldname) {
ensureActive(); ensureActive();
SortBuilder<?> sort = SortBuilders.fieldSort(timestampfieldname).order(SortOrder.DESC); SortBuilder<?> sort = SortBuilders
SearchSourceBuilder builder = new SearchSourceBuilder(); .fieldSort(timestampfieldname)
builder.sort(sort); .order(SortOrder.DESC);
builder.storedField(timestampfieldname); SearchSourceBuilder builder = new SearchSourceBuilder()
builder.size(1); .sort(sort)
SearchRequest searchRequest = new SearchRequest(); .storedField(timestampfieldname)
searchRequest.indices(index); .size(1);
searchRequest.source(builder); SearchRequest searchRequest = new SearchRequest()
.indices(index)
.source(builder);
SearchResponse searchResponse = SearchResponse searchResponse =
client.execute(SearchAction.INSTANCE, searchRequest).actionGet(); client.execute(SearchAction.INSTANCE, searchRequest).actionGet();
if (searchResponse.getHits().getHits().length == 1) { if (searchResponse.getHits().getHits().length == 1) {
@ -912,17 +933,17 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
private void checkMapping(String index, String type, MappingMetaData mappingMetaData) { private void checkMapping(String index, String type, MappingMetaData mappingMetaData) {
try { try {
SearchSourceBuilder builder = new SearchSourceBuilder(); SearchSourceBuilder builder = new SearchSourceBuilder()
builder.query(QueryBuilders.matchAllQuery()); .query(QueryBuilders.matchAllQuery())
builder.size(0); .size(0)
SearchRequest searchRequest = new SearchRequest(); .trackTotalHits(true);
searchRequest.indices(index); SearchRequest searchRequest = new SearchRequest()
searchRequest.types(type); .indices(index)
searchRequest.source(builder); .source(builder);
SearchResponse searchResponse = SearchResponse searchResponse =
client.execute(SearchAction.INSTANCE, searchRequest).actionGet(); client.execute(SearchAction.INSTANCE, searchRequest).actionGet();
long total = searchResponse.getHits().getTotalHits(); TotalHits total = searchResponse.getHits().getTotalHits();
if (total > 0L) { if (total.value > 0L) {
Map<String, Long> fields = new TreeMap<>(); Map<String, Long> fields = new TreeMap<>();
Map<String, Object> root = mappingMetaData.getSourceAsMap(); Map<String, Object> root = mappingMetaData.getSourceAsMap();
checkMapping(index, type, "", "", root, fields); checkMapping(index, type, "", "", root, fields);
@ -930,9 +951,7 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
Map<String, Long> map = sortByValue(fields); Map<String, Long> map = sortByValue(fields);
map.forEach((key, value) -> { map.forEach((key, value) -> {
logger.info("{} {} {}", logger.info("{} {} {}",
key, key, value, (double) value * 100 / total.value);
value,
(double) value * 100 / total);
if (value == 0) { if (value == 0) {
empty.incrementAndGet(); empty.incrementAndGet();
} }
@ -976,23 +995,23 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
} else if ("type".equals(key)) { } else if ("type".equals(key)) {
QueryBuilder filterBuilder = QueryBuilders.existsQuery(path); QueryBuilder filterBuilder = QueryBuilders.existsQuery(path);
QueryBuilder queryBuilder = QueryBuilders.constantScoreQuery(filterBuilder); QueryBuilder queryBuilder = QueryBuilders.constantScoreQuery(filterBuilder);
SearchSourceBuilder builder = new SearchSourceBuilder(); SearchSourceBuilder builder = new SearchSourceBuilder()
builder.query(queryBuilder); .query(queryBuilder)
builder.size(0); .size(0)
SearchRequest searchRequest = new SearchRequest(); .trackTotalHits(true);
searchRequest.indices(index); SearchRequest searchRequest = new SearchRequest()
searchRequest.types(type); .indices(index)
searchRequest.source(builder); .source(builder);
SearchResponse searchResponse = SearchResponse searchResponse =
client.execute(SearchAction.INSTANCE, searchRequest).actionGet(); client.execute(SearchAction.INSTANCE, searchRequest).actionGet();
fields.put(path, searchResponse.getHits().getTotalHits()); fields.put(path, searchResponse.getHits().getTotalHits().value);
} }
} }
} }
private static <K, V extends Comparable<? super V>> Map<K, V> sortByValue(Map<K, V> map) { private static <K, V extends Comparable<? super V>> Map<K, V> sortByValue(Map<K, V> map) {
Map<K, V> result = new LinkedHashMap<>(); Map<K, V> result = new LinkedHashMap<>();
map.entrySet().stream().sorted(Comparator.comparing(Map.Entry::getValue)) map.entrySet().stream().sorted(Map.Entry.comparingByValue())
.forEachOrdered(e -> result.put(e.getKey(), e.getValue())); .forEachOrdered(e -> result.put(e.getKey(), e.getValue()));
return result; return result;
} }
@ -1046,10 +1065,46 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
List<String> indicesToDelete; List<String> indicesToDelete;
DeleteIndexResponse response; AcknowledgedResponse response;
SuccessPruneResult(List<String> candidateIndices, List<String> indicesToDelete, SuccessPruneResult(List<String> candidateIndices, List<String> indicesToDelete,
DeleteIndexResponse response) { AcknowledgedResponse response) {
this.candidateIndices = candidateIndices;
this.indicesToDelete = indicesToDelete;
this.response = response;
}
@Override
public IndexPruneResult.State getState() {
return IndexPruneResult.State.SUCCESS;
}
@Override
public List<String> getCandidateIndices() {
return candidateIndices;
}
@Override
public List<String> getDeletedIndices() {
return indicesToDelete;
}
@Override
public boolean isAcknowledged() {
return response.isAcknowledged();
}
}
private static class FailPruneResult implements IndexPruneResult {
List<String> candidateIndices;
List<String> indicesToDelete;
AcknowledgedResponse response;
FailPruneResult(List<String> candidateIndices, List<String> indicesToDelete,
AcknowledgedResponse response) {
this.candidateIndices = candidateIndices; this.candidateIndices = candidateIndices;
this.indicesToDelete = indicesToDelete; this.indicesToDelete = indicesToDelete;
this.response = response; this.response = response;

6
elx-common/src/main/java/org/xbib/elx/common/DefaultBulkController.java
View file

@ -125,7 +125,7 @@ public class DefaultBulkController implements BulkController {
} }
try { try {
if (bulkMetric != null) { if (bulkMetric != null) {
bulkMetric.getCurrentIngest().inc(indexRequest.index(), indexRequest.type(), indexRequest.id()); bulkMetric.getCurrentIngest().inc(indexRequest.index(), "_doc", indexRequest.id());
} }
bulkProcessor.add(indexRequest); bulkProcessor.add(indexRequest);
} catch (Exception e) { } catch (Exception e) {
@ -144,7 +144,7 @@ public class DefaultBulkController implements BulkController {
} }
try { try {
if (bulkMetric != null) { if (bulkMetric != null) {
bulkMetric.getCurrentIngest().inc(deleteRequest.index(), deleteRequest.type(), deleteRequest.id()); bulkMetric.getCurrentIngest().inc(deleteRequest.index(), "_doc", deleteRequest.id());
} }
bulkProcessor.add(deleteRequest); bulkProcessor.add(deleteRequest);
} catch (Exception e) { } catch (Exception e) {
@ -163,7 +163,7 @@ public class DefaultBulkController implements BulkController {
} }
try { try {
if (bulkMetric != null) { if (bulkMetric != null) {
bulkMetric.getCurrentIngest().inc(updateRequest.index(), updateRequest.type(), updateRequest.id()); bulkMetric.getCurrentIngest().inc(updateRequest.index(), "_doc", updateRequest.id());
} }
bulkProcessor.add(updateRequest); bulkProcessor.add(updateRequest);
} catch (Exception e) { } catch (Exception e) {

22
elx-common/src/main/java/org/xbib/elx/common/DefaultBulkProcessor.java
View file

@ -140,19 +140,7 @@ public class DefaultBulkProcessor implements BulkProcessor {
*/ */
@Override @Override
public DefaultBulkProcessor add(ActionRequest request) { public DefaultBulkProcessor add(ActionRequest request) {
return add(request, null); internalAdd(request);
}
/**
* Adds either a delete or an index request with a payload.
*
* @param request request
* @param payload payload
* @return his bulk processor
*/
@Override
public DefaultBulkProcessor add(ActionRequest request, Object payload) {
internalAdd(request, payload);
return this; return this;
} }
@ -186,14 +174,14 @@ public class DefaultBulkProcessor implements BulkProcessor {
} }
} }
private synchronized void internalAdd(ActionRequest request, Object payload) { private synchronized void internalAdd(ActionRequest request) {
ensureOpen(); ensureOpen();
if (request instanceof IndexRequest) { if (request instanceof IndexRequest) {
bulkRequest.add((IndexRequest) request, payload); bulkRequest.add((IndexRequest) request);
} else if (request instanceof DeleteRequest) { } else if (request instanceof DeleteRequest) {
bulkRequest.add((DeleteRequest) request, payload); bulkRequest.add((DeleteRequest) request);
} else if (request instanceof UpdateRequest) { } else if (request instanceof UpdateRequest) {
bulkRequest.add((UpdateRequest) request, payload); bulkRequest.add((UpdateRequest) request);
} else { } else {
throw new UnsupportedOperationException(); throw new UnsupportedOperationException();
} }

4
elx-common/src/test/java/org/xbib/elx/common/test/MockNode.java
View file

@ -6,10 +6,12 @@ import org.elasticsearch.node.Node;
import org.elasticsearch.plugins.Plugin; import org.elasticsearch.plugins.Plugin;
import java.util.List; import java.util.List;
import java.util.Map;
class MockNode extends Node { class MockNode extends Node {
MockNode(Settings settings, List<Class<? extends Plugin>> classpathPlugins) { MockNode(Settings settings, List<Class<? extends Plugin>> classpathPlugins) {
super(InternalSettingsPreparer.prepareEnvironment(settings, null), classpathPlugins); super(InternalSettingsPreparer.prepareEnvironment(settings, Map.of(), null, () -> "mock"),
classpathPlugins, false);
} }
} }

5
elx-common/src/test/java/org/xbib/elx/common/test/SearchTest.java
View file

@ -33,7 +33,7 @@ class SearchTest {
ElasticsearchClient client = helper.client("1"); ElasticsearchClient client = helper.client("1");
BulkRequestBuilder builder = new BulkRequestBuilder(client, BulkAction.INSTANCE); BulkRequestBuilder builder = new BulkRequestBuilder(client, BulkAction.INSTANCE);
for (int i = 0; i < 1000; i++) { for (int i = 0; i < 1000; i++) {
IndexRequest indexRequest = new IndexRequest("pages", "row") IndexRequest indexRequest = new IndexRequest().index("pages")
.source(XContentFactory.jsonBuilder() .source(XContentFactory.jsonBuilder()
.startObject() .startObject()
.field("user1", "joerg") .field("user1", "joerg")
@ -61,10 +61,9 @@ class SearchTest {
searchSource.size(10); searchSource.size(10);
SearchRequest searchRequest = new SearchRequest(); SearchRequest searchRequest = new SearchRequest();
searchRequest.indices("pages"); searchRequest.indices("pages");
searchRequest.types("row");
searchRequest.source(searchSource); searchRequest.source(searchSource);
SearchResponse searchResponse = client.execute(SearchAction.INSTANCE, searchRequest).actionGet(); SearchResponse searchResponse = client.execute(SearchAction.INSTANCE, searchRequest).actionGet();
assertTrue(searchResponse.getHits().getTotalHits() > 0); assertTrue(searchResponse.getHits().getTotalHits().value > 0);
} }
} }
} }

6
elx-common/src/test/java/org/xbib/elx/common/test/SimpleTest.java
View file

@ -41,13 +41,13 @@ class SimpleTest {
Settings indexSettings = Settings.builder() Settings indexSettings = Settings.builder()
.put("index.analysis.analyzer.default.filter.0", "lowercase") .put("index.analysis.analyzer.default.filter.0", "lowercase")
.put("index.analysis.analyzer.default.filter.1", "trim") .put("index.analysis.analyzer.default.filter.1", "trim")
.put("index.analysis.analyzer.default.tokenizer", "keyword") .put("index.analysis.analyzer.default.type", "keyword")
.build(); .build();
CreateIndexRequest createIndexRequest = new CreateIndexRequest(); CreateIndexRequest createIndexRequest = new CreateIndexRequest();
createIndexRequest.index("test").settings(indexSettings); createIndexRequest.index("test").settings(indexSettings);
helper.client("1").execute(CreateIndexAction.INSTANCE, createIndexRequest).actionGet(); helper.client("1").execute(CreateIndexAction.INSTANCE, createIndexRequest).actionGet();
IndexRequest indexRequest = new IndexRequest(); IndexRequest indexRequest = new IndexRequest();
indexRequest.index("test").type("test").id("1") indexRequest.index("test").id("1")
.source(XContentFactory.jsonBuilder().startObject().field("field", .source(XContentFactory.jsonBuilder().startObject().field("field",
"1%2fPJJP3JV2C24iDfEu9XpHBaYxXh%2fdHTbmchB35SDznXO2g8Vz4D7GTIvY54iMiX_149c95f02a8").endObject()); "1%2fPJJP3JV2C24iDfEu9XpHBaYxXh%2fdHTbmchB35SDznXO2g8Vz4D7GTIvY54iMiX_149c95f02a8").endObject());
helper.client("1").execute(IndexAction.INSTANCE, indexRequest).actionGet(); helper.client("1").execute(IndexAction.INSTANCE, indexRequest).actionGet();
@ -58,7 +58,7 @@ class SimpleTest {
builder.query(QueryBuilders.matchQuery("field", builder.query(QueryBuilders.matchQuery("field",
"1%2fPJJP3JV2C24iDfEu9XpHBaYxXh%2fdHTbmchB35SDznXO2g8Vz4D7GTIvY54iMiX_149c95f02a8")); "1%2fPJJP3JV2C24iDfEu9XpHBaYxXh%2fdHTbmchB35SDznXO2g8Vz4D7GTIvY54iMiX_149c95f02a8"));
SearchRequest searchRequest = new SearchRequest(); SearchRequest searchRequest = new SearchRequest();
searchRequest.indices("test").types("test"); searchRequest.indices("test");
searchRequest.source(builder); searchRequest.source(builder);
String doc = helper.client("1").execute(SearchAction.INSTANCE, searchRequest).actionGet() String doc = helper.client("1").execute(SearchAction.INSTANCE, searchRequest).actionGet()
.getHits().getAt(0).getSourceAsString(); .getHits().getAt(0).getSourceAsString();

11
elx-common/src/test/java/org/xbib/elx/common/test/TestExtension.java
View file

@ -12,7 +12,6 @@ import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
import org.elasticsearch.action.admin.cluster.state.ClusterStateAction; import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest; import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse; import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.analysis.common.CommonAnalysisPlugin;
import org.elasticsearch.client.ElasticsearchClient; import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.client.support.AbstractClient; import org.elasticsearch.client.support.AbstractClient;
import org.elasticsearch.cluster.health.ClusterHealthStatus; import org.elasticsearch.cluster.health.ClusterHealthStatus;
@ -22,7 +21,7 @@ import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.node.Node; import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeValidationException; import org.elasticsearch.node.NodeValidationException;
import org.elasticsearch.plugins.Plugin; import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.transport.netty4.Netty4Plugin; import org.elasticsearch.transport.Netty4Plugin;
import org.junit.jupiter.api.extension.AfterEachCallback; import org.junit.jupiter.api.extension.AfterEachCallback;
import org.junit.jupiter.api.extension.BeforeEachCallback; import org.junit.jupiter.api.extension.BeforeEachCallback;
import org.junit.jupiter.api.extension.ExtensionContext; import org.junit.jupiter.api.extension.ExtensionContext;
@ -37,7 +36,7 @@ import java.nio.file.Path;
import java.nio.file.Paths; import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor; import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes; import java.nio.file.attribute.BasicFileAttributes;
import java.util.Arrays; import java.util.Collections;
import java.util.HashMap; import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
@ -188,8 +187,8 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
return Settings.builder() return Settings.builder()
.put("cluster.name", getClusterName()) .put("cluster.name", getClusterName())
.put("path.home", getHome()) .put("path.home", getHome())
.put("discovery.zen.master_election.ignore_non_master_pings", "true") .put("cluster.initial_master_nodes", "1")
.put("transport.netty.epoll", "false") .put("discovery.seed_hosts", "127.0.0.1:9300")
.build(); .build();
} }
@ -215,7 +214,7 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
.put(getNodeSettings()) .put(getNodeSettings())
.put("node.name", id) .put("node.name", id)
.build(); .build();
List<Class<? extends Plugin>> plugins = Arrays.asList(CommonAnalysisPlugin.class, Netty4Plugin.class); List<Class<? extends Plugin>> plugins = Collections.singletonList(Netty4Plugin.class);
Node node = new MockNode(nodeSettings, plugins); Node node = new MockNode(nodeSettings, plugins);
AbstractClient client = (AbstractClient) node.client(); AbstractClient client = (AbstractClient) node.client();
nodes.put(id, node); nodes.put(id, node);

16
elx-common/src/test/java/org/xbib/elx/common/test/WildcardTest.java
View file

@ -45,20 +45,20 @@ class WildcardTest {
} }
private void index(ElasticsearchClient client, String id, String fieldValue) throws IOException { private void index(ElasticsearchClient client, String id, String fieldValue) throws IOException {
client.execute(IndexAction.INSTANCE, new IndexRequest("index", "type", id) client.execute(IndexAction.INSTANCE, new IndexRequest().index("index").id(id)
.source(XContentFactory.jsonBuilder().startObject().field("field", fieldValue).endObject())) .source(XContentFactory.jsonBuilder().startObject().field("field", fieldValue).endObject()))
.actionGet(); .actionGet();
client.execute(RefreshAction.INSTANCE, new RefreshRequest()).actionGet(); client.execute(RefreshAction.INSTANCE, new RefreshRequest()).actionGet();
} }
private long count(ElasticsearchClient client, QueryBuilder queryBuilder) { private long count(ElasticsearchClient client, QueryBuilder queryBuilder) {
SearchSourceBuilder builder = new SearchSourceBuilder(); SearchSourceBuilder builder = new SearchSourceBuilder()
builder.query(queryBuilder); .query(queryBuilder)
SearchRequest searchRequest = new SearchRequest(); .trackTotalHits(true);
searchRequest.indices("index"); SearchRequest searchRequest = new SearchRequest()
searchRequest.types("type"); .indices("index")
searchRequest.source(builder); .source(builder);
return client.execute(SearchAction.INSTANCE, searchRequest).actionGet().getHits().getTotalHits(); return client.execute(SearchAction.INSTANCE, searchRequest).actionGet().getHits().getTotalHits().value;
} }
private void validateCount(ElasticsearchClient client, QueryBuilder queryBuilder, long expectedHits) { private void validateCount(ElasticsearchClient client, QueryBuilder queryBuilder, long expectedHits) {

3
elx-http/build.gradle
View file

@ -2,7 +2,7 @@ import org.apache.tools.ant.taskdefs.condition.Os
dependencies{ dependencies{
compile project(':elx-common') compile project(':elx-common')
compile "org.xbib.elasticsearch:transport-netty4:${rootProject.property('elasticsearch-server.version')}" compile "org.elasticsearch.plugin:transport-netty4-client:${rootProject.property('elasticsearch.version')}"
compile "org.xbib:netty-http-client:${project.property('xbib-netty-http.version')}" compile "org.xbib:netty-http-client:${project.property('xbib-netty-http.version')}"
runtime "org.bouncycastle:bcpkix-jdk15on:${project.property('bouncycastle.version')}" runtime "org.bouncycastle:bcpkix-jdk15on:${project.property('bouncycastle.version')}"
if (Os.isFamily(Os.FAMILY_MAC)) { if (Os.isFamily(Os.FAMILY_MAC)) {
@ -10,5 +10,4 @@ dependencies{
} else if (Os.isFamily(Os.FAMILY_UNIX)) { } else if (Os.isFamily(Os.FAMILY_UNIX)) {
runtime "io.netty:netty-tcnative-boringssl-static:${project.property('tcnative.version')}" runtime "io.netty:netty-tcnative-boringssl-static:${project.property('tcnative.version')}"
} }
testCompile "org.xbib.elasticsearch:elasticsearch-analysis-common:${rootProject.property('elasticsearch-server.version')}"
} }

26
elx-http/src/main/java/org/xbib/elx/http/ExtendedHttpClient.java
View file

@ -3,13 +3,12 @@ package org.xbib.elx.http;
import org.apache.logging.log4j.Level; import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.Action;
import org.elasticsearch.action.ActionFuture; import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.ActionListener; import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.ActionRequest; import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.action.ActionRequestBuilder; import org.elasticsearch.action.ActionRequestBuilder;
import org.elasticsearch.action.ActionResponse; import org.elasticsearch.action.ActionResponse;
import org.elasticsearch.action.GenericAction; import org.elasticsearch.action.ActionType;
import org.elasticsearch.action.support.PlainActionFuture; import org.elasticsearch.action.support.PlainActionFuture;
import org.elasticsearch.client.ElasticsearchClient; import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
@ -43,7 +42,7 @@ public class ExtendedHttpClient extends AbstractExtendedClient implements Elasti
private final NamedXContentRegistry registry; private final NamedXContentRegistry registry;
@SuppressWarnings("rawtypes") @SuppressWarnings("rawtypes")
private final Map<GenericAction, HttpAction> actionMap; private final Map<ActionType, HttpAction> actionMap;
private String url; private String url;
@ -109,28 +108,19 @@ public class ExtendedHttpClient extends AbstractExtendedClient implements Elasti
} }
@Override @Override
public <Request extends ActionRequest, Response extends ActionResponse, public <Request extends ActionRequest, Response extends ActionResponse>
RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> ActionFuture<Response> ActionFuture<Response> execute(ActionType<Response> action, Request request) {
execute(Action<Request, Response, RequestBuilder> action, Request request) {
PlainActionFuture<Response> actionFuture = PlainActionFuture.newFuture(); PlainActionFuture<Response> actionFuture = PlainActionFuture.newFuture();
execute(action, request, actionFuture); execute(action, request, actionFuture);
return actionFuture; return actionFuture;
} }
@Override @Override
public <Request extends ActionRequest, Response extends ActionResponse, public <Request extends ActionRequest, Response extends ActionResponse>
RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> void void execute(ActionType<Response> action, Request request, ActionListener<Response> listener) {
execute(Action<Request, Response, RequestBuilder> action, Request request, ActionListener<Response> listener) {
doExecute(action, request, listener); doExecute(action, request, listener);
} }
@Override
public <Request extends ActionRequest, Response extends ActionResponse,
RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> RequestBuilder
prepareExecute(Action<Request, Response, RequestBuilder> action) {
return action.newRequestBuilder(this);
}
@Override @Override
public ThreadPool threadPool() { public ThreadPool threadPool() {
logger.log(Level.DEBUG, "returning null for threadPool() request"); logger.log(Level.DEBUG, "returning null for threadPool() request");
@ -138,8 +128,8 @@ public class ExtendedHttpClient extends AbstractExtendedClient implements Elasti
} }
@SuppressWarnings({"unchecked", "rawtypes"}) @SuppressWarnings({"unchecked", "rawtypes"})
private <R extends ActionRequest, T extends ActionResponse, B extends ActionRequestBuilder<R, T, B>> private <R extends ActionRequest, T extends ActionResponse, B extends ActionRequestBuilder<R, T>>
void doExecute(Action<R, T, B> action, R request, ActionListener<T> listener) { void doExecute(ActionType<T> action, R request, ActionListener<T> listener) {
HttpAction httpAction = actionMap.get(action); HttpAction httpAction = actionMap.get(action);
if (httpAction == null) { if (httpAction == null) {
throw new IllegalStateException("failed to find http action [" + action + "] to execute"); throw new IllegalStateException("failed to find http action [" + action + "] to execute");

17
elx-http/src/main/java/org/xbib/elx/http/HttpAction.java
View file

@ -12,7 +12,7 @@ import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.ActionRequest; import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.action.ActionRequestValidationException; import org.elasticsearch.action.ActionRequestValidationException;
import org.elasticsearch.action.ActionResponse; import org.elasticsearch.action.ActionResponse;
import org.elasticsearch.action.GenericAction; import org.elasticsearch.action.ActionType;
import org.elasticsearch.common.CheckedFunction; import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
@ -25,6 +25,7 @@ import org.elasticsearch.rest.RestStatus;
import org.xbib.net.URL; import org.xbib.net.URL;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.client.api.Transport; import org.xbib.netty.http.client.api.Transport;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
import java.nio.charset.StandardCharsets; import java.nio.charset.StandardCharsets;
@ -47,7 +48,7 @@ public abstract class HttpAction<R extends ActionRequest, T extends ActionRespon
this.settings = settings; this.settings = settings;
} }
public abstract GenericAction<R, T> getActionInstance(); public abstract ActionType<T> getActionInstance();
final void execute(HttpActionContext<R, T> httpActionContext, ActionListener<T> listener) throws IOException { final void execute(HttpActionContext<R, T> httpActionContext, ActionListener<T> listener) throws IOException {
try { try {
@ -71,7 +72,7 @@ public abstract class HttpAction<R extends ActionRequest, T extends ActionRespon
} else { } else {
ElasticsearchStatusException statusException = parseToError(httpActionContext); ElasticsearchStatusException statusException = parseToError(httpActionContext);
if (statusException.status().equals(RestStatus.NOT_FOUND)) { if (statusException.status().equals(RestStatus.NOT_FOUND)) {
listener.onResponse(emptyResponse()); listener.onResponse(parseToResponse(httpActionContext));
} else { } else {
listener.onFailure(statusException); listener.onFailure(statusException);
} }
@ -158,7 +159,7 @@ public abstract class HttpAction<R extends ActionRequest, T extends ActionRespon
.createParser(httpActionContext.getExtendedHttpClient().getRegistry(), .createParser(httpActionContext.getExtendedHttpClient().getRegistry(),
DeprecationHandler.THROW_UNSUPPORTED_OPERATION, DeprecationHandler.THROW_UNSUPPORTED_OPERATION,
httpActionContext.getHttpResponse().getBody().toString(StandardCharsets.UTF_8))) { httpActionContext.getHttpResponse().getBody().toString(StandardCharsets.UTF_8))) {
return entityParser().apply(parser); return entityParser(httpActionContext.getHttpResponse()).apply(parser);
} catch (IOException e) { } catch (IOException e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
return null; return null;
@ -166,12 +167,13 @@ public abstract class HttpAction<R extends ActionRequest, T extends ActionRespon
} }
protected ElasticsearchStatusException parseToError(HttpActionContext<R, T> httpActionContext) { protected ElasticsearchStatusException parseToError(HttpActionContext<R, T> httpActionContext) {
// we assume a non-empty, valid JSON response body. If there is none, this method must be overriden.
try (XContentParser parser = XContentFactory.xContent(XContentType.JSON) try (XContentParser parser = XContentFactory.xContent(XContentType.JSON)
.createParser(httpActionContext.getExtendedHttpClient().getRegistry(), .createParser(httpActionContext.getExtendedHttpClient().getRegistry(),
DeprecationHandler.THROW_UNSUPPORTED_OPERATION, DeprecationHandler.THROW_UNSUPPORTED_OPERATION,
httpActionContext.getHttpResponse().getBody().toString(StandardCharsets.UTF_8))) { httpActionContext.getHttpResponse().getBody().toString(StandardCharsets.UTF_8))) {
return errorParser().apply(parser); return errorParser().apply(parser);
} catch (IOException e) { } catch (Exception e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
return new ElasticsearchStatusException(e.getMessage(), RestStatus.INTERNAL_SERVER_ERROR, e); return new ElasticsearchStatusException(e.getMessage(), RestStatus.INTERNAL_SERVER_ERROR, e);
} }
@ -183,8 +185,5 @@ public abstract class HttpAction<R extends ActionRequest, T extends ActionRespon
protected abstract Request.Builder createHttpRequest(String baseUrl, R request) throws IOException; protected abstract Request.Builder createHttpRequest(String baseUrl, R request) throws IOException;
protected abstract CheckedFunction<XContentParser, T, IOException> entityParser(); protected abstract CheckedFunction<XContentParser, T, IOException> entityParser(HttpResponse httpResponse);
protected abstract T emptyResponse();
} }

8
elx-http/src/main/java/org/xbib/elx/http/action/admin/cluster/health/HttpClusterHealthAction.java
View file

@ -7,6 +7,7 @@ import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
@ -23,12 +24,7 @@ public class HttpClusterHealthAction extends HttpAction<ClusterHealthRequest, Cl
} }
@Override @Override
protected CheckedFunction<XContentParser, ClusterHealthResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, ClusterHealthResponse, IOException> entityParser(HttpResponse httpResponse) {
return HttpClusterHealthResponse::fromXContent; return HttpClusterHealthResponse::fromXContent;
} }
@Override
protected ClusterHealthResponse emptyResponse() {
return new HttpClusterHealthResponse();
}
} }

68
elx-http/src/main/java/org/xbib/elx/http/action/admin/cluster/health/HttpClusterHealthResponse.java
View file

@ -6,36 +6,13 @@ import org.elasticsearch.cluster.health.ClusterIndexHealth;
import org.elasticsearch.cluster.health.ClusterStateHealth; import org.elasticsearch.cluster.health.ClusterStateHealth;
import org.elasticsearch.common.io.stream.BytesStreamOutput; import org.elasticsearch.common.io.stream.BytesStreamOutput;
import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.rest.RestStatus; import org.elasticsearch.rest.RestStatus;
import java.io.IOException; import java.io.IOException;
import java.util.Collections;
import java.util.Map; import java.util.Map;
import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
public class HttpClusterHealthResponse extends ClusterHealthResponse { public class HttpClusterHealthResponse extends ClusterHealthResponse {
private static final String CLUSTER_NAME = "cluster_name";
private static final String STATUS = "status";
private static final String TIMED_OUT = "timed_out";
private static final String NUMBER_OF_NODES = "number_of_nodes";
private static final String NUMBER_OF_DATA_NODES = "number_of_data_nodes";
private static final String NUMBER_OF_PENDING_TASKS = "number_of_pending_tasks";
private static final String NUMBER_OF_IN_FLIGHT_FETCH = "number_of_in_flight_fetch";
private static final String DELAYED_UNASSIGNED_SHARDS = "delayed_unassigned_shards";
private static final String TASK_MAX_WAIT_TIME_IN_QUEUE = "task_max_waiting_in_queue";
private static final String TASK_MAX_WAIT_TIME_IN_QUEUE_IN_MILLIS = "task_max_waiting_in_queue_millis";
private static final String ACTIVE_SHARDS_PERCENT_AS_NUMBER = "active_shards_percent_as_number";
private static final String ACTIVE_SHARDS_PERCENT = "active_shards_percent";
private static final String ACTIVE_PRIMARY_SHARDS = "active_primary_shards";
private static final String ACTIVE_SHARDS = "active_shards";
private static final String RELOCATING_SHARDS = "relocating_shards";
private static final String INITIALIZING_SHARDS = "initializing_shards";
private static final String UNASSIGNED_SHARDS = "unassigned_shards";
private static final String INDICES = "indices";
private String clusterName; private String clusterName;
private ClusterStateHealth clusterStateHealth; private ClusterStateHealth clusterStateHealth;
@ -93,11 +70,6 @@ public class HttpClusterHealthResponse extends ClusterHealthResponse {
return clusterName; return clusterName;
} }
@Override
public ClusterStateHealth getClusterStateHealth() {
return clusterStateHealth;
}
@Override @Override
public boolean isTimedOut() { public boolean isTimedOut() {
return this.timedOut; return this.timedOut;
@ -172,44 +144,4 @@ public class HttpClusterHealthResponse extends ClusterHealthResponse {
public RestStatus status() { public RestStatus status() {
return isTimedOut() ? RestStatus.REQUEST_TIMEOUT : RestStatus.OK; return isTimedOut() ? RestStatus.REQUEST_TIMEOUT : RestStatus.OK;
} }
public static HttpClusterHealthResponse fromXContent(XContentParser parser) throws IOException {
XContentParser.Token token = parser.nextToken();
ensureExpectedToken(XContentParser.Token.START_OBJECT, token, parser::getTokenLocation);
Map<String, Object> map = parser.map();
String clusterName = (String) map.get(CLUSTER_NAME);
ClusterHealthStatus status = ClusterHealthStatus.fromString((String) map.get(STATUS));
Boolean timedOut = (Boolean) map.get(TIMED_OUT);
Integer numberOfNodes = (Integer) map.get(NUMBER_OF_NODES);
Integer numberOfDataNodes = (Integer) map.get(NUMBER_OF_DATA_NODES);
Integer activePrimaryShards = (Integer) map.get(ACTIVE_PRIMARY_SHARDS);
Integer activeShards = (Integer) map.get(ACTIVE_SHARDS);
Integer relocatingShards = (Integer) map.get(RELOCATING_SHARDS);
Integer initializingShards = (Integer) map.get(INITIALIZING_SHARDS);
Integer unassignedShards = (Integer) map.get(UNASSIGNED_SHARDS);
Integer delayedUnassignedShards = (Integer) map.get(DELAYED_UNASSIGNED_SHARDS);
Integer numberOfPendingTasks = (Integer) map.get(NUMBER_OF_PENDING_TASKS);
Integer numberOfInFlightFetch = (Integer) map.get(NUMBER_OF_IN_FLIGHT_FETCH);
Integer taskMaxWaitingInQueueMillis = (Integer) map.get(TASK_MAX_WAIT_TIME_IN_QUEUE_IN_MILLIS);
Double activeShardsPercentAsNumber = (Double) map.get(ACTIVE_SHARDS_PERCENT_AS_NUMBER);
HttpClusterHealthResponse clusterHealthResponse = new HttpClusterHealthResponse();
clusterHealthResponse.init(clusterName,
status,
timedOut,
numberOfNodes,
numberOfDataNodes,
Collections.emptyMap(),
activePrimaryShards,
activeShards,
relocatingShards,
initializingShards,
unassignedShards,
delayedUnassignedShards,
numberOfPendingTasks,
numberOfInFlightFetch,
TimeValue.timeValueMillis(taskMaxWaitingInQueueMillis),
activeShardsPercentAsNumber
);
return clusterHealthResponse;
}
} }

103
elx-http/src/main/java/org/xbib/elx/http/action/admin/cluster/node/info/HttpNodesInfoAction.java
View file

@ -1,39 +1,17 @@
package org.xbib.elx.http.action.admin.cluster.node.info; package org.xbib.elx.http.action.admin.cluster.node.info;
import org.elasticsearch.Build;
import org.elasticsearch.Version;
import org.elasticsearch.action.FailedNodeException;
import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoAction; import org.elasticsearch.action.admin.cluster.node.info.NodesInfoAction;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoRequest; import org.elasticsearch.action.admin.cluster.node.info.NodesInfoRequest;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse; import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
import org.elasticsearch.action.admin.cluster.node.info.PluginsAndModules;
import org.elasticsearch.cluster.ClusterName;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.CheckedFunction; import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.http.HttpInfo;
import org.elasticsearch.ingest.IngestInfo;
import org.elasticsearch.monitor.jvm.JvmInfo;
import org.elasticsearch.monitor.os.OsInfo;
import org.elasticsearch.monitor.process.ProcessInfo;
import org.elasticsearch.threadpool.ThreadPoolInfo;
import org.elasticsearch.transport.TransportInfo;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.elx.http.HttpActionContext;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
import java.io.UncheckedIOException;
import java.net.InetAddress;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedList; import java.util.LinkedList;
import java.util.List; import java.util.List;
import java.util.Map;
import java.util.Set;
/** /**
* *
@ -95,84 +73,7 @@ public class HttpNodesInfoAction extends HttpAction<NodesInfoRequest, NodesInfoR
} }
@Override @Override
protected CheckedFunction<XContentParser, NodesInfoResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, NodesInfoResponse, IOException> entityParser(HttpResponse httpResponse) {
throw new UnsupportedOperationException(); throw new UnsupportedOperationException();
} }
@Override
protected NodesInfoResponse emptyResponse() {
return new NodesInfoResponse();
}
@SuppressWarnings("unchecked")
protected NodesInfoResponse createResponse(HttpActionContext<NodesInfoRequest, NodesInfoResponse> httpContext) {
// BROKEN
Map<String, Object> map = null;
//String string = (String)map.get("cluster_name");
ClusterName clusterName = new ClusterName("");
List<NodeInfo> nodeInfoList = new LinkedList<>();
//map = (Map<String, Object>)map.get("nodes");
for (Map.Entry<String, Object> entry : map.entrySet()) {
String nodeId = entry.getKey();
String ephemeralId = null;
Map<String,Object> map2 = (Map<String, Object>) entry.getValue();
String nodeName = (String) map2.get("name");
String hostName = (String) map2.get("host");
String hostAddress = (String) map2.get("ip");
// <host>[/<ip>][:<port>]
String transportAddressString = (String) map2.get("transport_address");
int pos = transportAddressString.indexOf(':');
String host = pos > 0 ? transportAddressString.substring(0, pos) : transportAddressString;
int port = Integer.parseInt(pos > 0 ? transportAddressString.substring(pos + 1) : "0");
pos = host.indexOf('/');
host = pos > 0 ? host.substring(0, pos) : host;
try {
InetAddress[] inetAddresses = InetAddress.getAllByName(host);
TransportAddress transportAddress = new TransportAddress(inetAddresses[0], port);
Build build = new Build(Build.Flavor.OSS, Build.Type.TAR,
(String) map2.get("build"),
(String) map2.get("date"),
(Boolean) map2.get("snapshot"));
Map<String, String> attributes = Collections.emptyMap();
Set<DiscoveryNode.Role> roles = new HashSet<>();
Version version = Version.fromString((String) map2.get("version"));
DiscoveryNode discoveryNode = new DiscoveryNode(nodeName, nodeId, ephemeralId, hostName, hostAddress,
transportAddress,
attributes, roles, version);
/*Map<String, String> settingsMap = map2.containsKey("settings") ?
XContentHelper.
SettingsLoader.Helper.loadNestedFromMap((Map<String, Object>) map2.get("settings")) :
Collections.emptyMap();
Settings settings = Settings.builder()
.put(settingsMap)
.build();*/
OsInfo os = null;
ProcessInfo processInfo = null;
JvmInfo jvmInfo = null;
ThreadPoolInfo threadPoolInfo = null;
TransportInfo transportInfo = null;
HttpInfo httpInfo = null;
PluginsAndModules pluginsAndModules = null;
IngestInfo ingestInfo = null;
ByteSizeValue totalIndexingBuffer = null;
NodeInfo nodeInfo = new NodeInfo(version,
build,
discoveryNode,
//serviceAttributes,
//settings,
null,
os, processInfo, jvmInfo, threadPoolInfo, transportInfo, httpInfo, pluginsAndModules,
ingestInfo,
totalIndexingBuffer);
nodeInfoList.add(nodeInfo);
} catch (IOException e) {
throw new UncheckedIOException(e);
}
}
List<FailedNodeException> failures = null;
return new NodesInfoResponse(clusterName, nodeInfoList, failures);
}
} }

8
elx-http/src/main/java/org/xbib/elx/http/action/admin/cluster/settings/HttpClusterUpdateSettingsAction.java
View file

@ -10,6 +10,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
import java.io.UncheckedIOException; import java.io.UncheckedIOException;
@ -40,12 +41,7 @@ public class HttpClusterUpdateSettingsAction extends HttpAction<ClusterUpdateSet
} }
@Override @Override
protected CheckedFunction<XContentParser, ClusterUpdateSettingsResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, ClusterUpdateSettingsResponse, IOException> entityParser(HttpResponse httpResponse) {
return ClusterUpdateSettingsResponse::fromXContent; return ClusterUpdateSettingsResponse::fromXContent;
} }
@Override
protected ClusterUpdateSettingsResponse emptyResponse() {
return new ClusterUpdateSettingsResponse();
}
} }

11
elx-http/src/main/java/org/xbib/elx/http/action/admin/cluster/state/HttpClusterStateAction.java
View file

@ -1,7 +1,6 @@
package org.xbib.elx.http.action.admin.cluster.state; package org.xbib.elx.http.action.admin.cluster.state;
import com.carrotsearch.hppc.LongArrayList; import com.carrotsearch.hppc.LongArrayList;
import org.apache.logging.log4j.Level;
import org.elasticsearch.action.admin.cluster.state.ClusterStateAction; import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest; import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse; import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
@ -20,6 +19,7 @@ import org.elasticsearch.common.xcontent.NamedObjectNotFoundException;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
import java.util.ArrayList; import java.util.ArrayList;
@ -73,15 +73,10 @@ public class HttpClusterStateAction extends HttpAction<ClusterStateRequest, Clus
} }
@Override @Override
protected CheckedFunction<XContentParser, ClusterStateResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, ClusterStateResponse, IOException> entityParser(HttpResponse httpResponse) {
return this::fromXContent; return this::fromXContent;
} }
@Override
protected ClusterStateResponse emptyResponse() {
return new ClusterStateResponse();
}
private ClusterStateResponse fromXContent(XContentParser parser) throws IOException { private ClusterStateResponse fromXContent(XContentParser parser) throws IOException {
XContentParser.Token token = parser.nextToken(); XContentParser.Token token = parser.nextToken();
ensureExpectedToken(XContentParser.Token.START_OBJECT, token, parser::getTokenLocation); ensureExpectedToken(XContentParser.Token.START_OBJECT, token, parser::getTokenLocation);
@ -109,7 +104,7 @@ public class HttpClusterStateAction extends HttpAction<ClusterStateRequest, Clus
} }
ClusterState clusterState = builder.build(); ClusterState clusterState = builder.build();
return new ClusterStateResponse(clusterName, clusterState, length); return new ClusterStateResponse(clusterName, clusterState, true);
} }
private MetaData metadataFromXContent(XContentParser parser) throws IOException { private MetaData metadataFromXContent(XContentParser parser) throws IOException {

14
elx-http/src/main/java/org/xbib/elx/http/action/admin/indices/alias/HttpIndicesAliasesAction.java
View file

@ -3,7 +3,7 @@ package org.xbib.elx.http.action.admin.indices.alias;
import org.apache.logging.log4j.Level; import org.apache.logging.log4j.Level;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesAction; import org.elasticsearch.action.admin.indices.alias.IndicesAliasesAction;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest; import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesResponse; import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.common.CheckedFunction; import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.Strings; import org.elasticsearch.common.Strings;
import org.elasticsearch.common.xcontent.ToXContent; import org.elasticsearch.common.xcontent.ToXContent;
@ -12,10 +12,11 @@ import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.json.JsonXContent; import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
public class HttpIndicesAliasesAction extends HttpAction<IndicesAliasesRequest, IndicesAliasesResponse> { public class HttpIndicesAliasesAction extends HttpAction<IndicesAliasesRequest, AcknowledgedResponse> {
@Override @Override
public IndicesAliasesAction getActionInstance() { public IndicesAliasesAction getActionInstance() {
@ -37,12 +38,7 @@ public class HttpIndicesAliasesAction extends HttpAction<IndicesAliasesRequest,
} }
@Override @Override
protected CheckedFunction<XContentParser, IndicesAliasesResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, AcknowledgedResponse, IOException> entityParser(HttpResponse httpResponse) {
return IndicesAliasesResponse::fromXContent; return AcknowledgedResponse::fromXContent;
}
@Override
protected IndicesAliasesResponse emptyResponse() {
return new IndicesAliasesResponse();
} }
} }

13
elx-http/src/main/java/org/xbib/elx/http/action/admin/indices/alias/get/HttpGetAliasAction.java
View file

@ -1,6 +1,6 @@
package org.xbib.elx.http.action.admin.indices.alias.get; package org.xbib.elx.http.action.admin.indices.alias.get;
import org.elasticsearch.action.GenericAction; import org.elasticsearch.action.ActionType;
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesAction; import org.elasticsearch.action.admin.indices.alias.get.GetAliasesAction;
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest; import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest;
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesResponse; import org.elasticsearch.action.admin.indices.alias.get.GetAliasesResponse;
@ -10,6 +10,7 @@ import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
import java.util.ArrayList; import java.util.ArrayList;
@ -29,21 +30,15 @@ public class HttpGetAliasAction extends HttpAction<GetAliasesRequest, GetAliases
} }
@Override @Override
public GenericAction<GetAliasesRequest, GetAliasesResponse> getActionInstance() { public ActionType<GetAliasesResponse> getActionInstance() {
return GetAliasesAction.INSTANCE; return GetAliasesAction.INSTANCE;
} }
@Override @Override
protected CheckedFunction<XContentParser, GetAliasesResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, GetAliasesResponse, IOException> entityParser(HttpResponse httpResponse) {
return this::fromXContent; return this::fromXContent;
} }
@Override
protected GetAliasesResponse emptyResponse() {
ImmutableOpenMap.Builder<String, List<AliasMetaData>> aliasesBuilder = ImmutableOpenMap.builder();
return new GetAliasesResponse(aliasesBuilder.build());
}
private GetAliasesResponse fromXContent(XContentParser parser) throws IOException { private GetAliasesResponse fromXContent(XContentParser parser) throws IOException {
if (parser.currentToken() == null) { if (parser.currentToken() == null) {
parser.nextToken(); parser.nextToken();

47
elx-http/src/main/java/org/xbib/elx/http/action/admin/indices/create/HttpCreateIndexAction.java
View file

@ -1,21 +1,32 @@
package org.xbib.elx.http.action.admin.indices.create; package org.xbib.elx.http.action.admin.indices.create;
import org.elasticsearch.action.admin.indices.alias.Alias;
import org.elasticsearch.action.admin.indices.create.CreateIndexAction; import org.elasticsearch.action.admin.indices.create.CreateIndexAction;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest; import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse; import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.common.CheckedFunction; import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.ParseField;
import org.elasticsearch.common.bytes.BytesArray;
import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.xcontent.ToXContent; import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder; import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory; import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
import java.util.Map;
public class HttpCreateIndexAction extends HttpAction<CreateIndexRequest, CreateIndexResponse> { public class HttpCreateIndexAction extends HttpAction<CreateIndexRequest, CreateIndexResponse> {
public static final ParseField MAPPINGS = new ParseField("mappings");
public static final ParseField SETTINGS = new ParseField("settings");
public static final ParseField ALIASES = new ParseField("aliases");
@Override @Override
public CreateIndexAction getActionInstance() { public CreateIndexAction getActionInstance() {
return CreateIndexAction.INSTANCE; return CreateIndexAction.INSTANCE;
@ -23,18 +34,40 @@ public class HttpCreateIndexAction extends HttpAction<CreateIndexRequest, Create
@Override @Override
protected Request.Builder createHttpRequest(String url, CreateIndexRequest createIndexRequest) throws IOException { protected Request.Builder createHttpRequest(String url, CreateIndexRequest createIndexRequest) throws IOException {
XContentBuilder builder = XContentFactory.jsonBuilder(); XContentBuilder builder = toXContent(createIndexRequest, XContentFactory.jsonBuilder());
builder = createIndexRequest.toXContent(builder, ToXContent.EMPTY_PARAMS); return newPutRequest(url, "/" + createIndexRequest.index() + "?include_type_name=true",
return newPutRequest(url, "/" + createIndexRequest.index(), BytesReference.bytes(builder)); BytesReference.bytes(builder));
} }
@Override @Override
protected CheckedFunction<XContentParser, CreateIndexResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, CreateIndexResponse, IOException> entityParser(HttpResponse httpResponse) {
return CreateIndexResponse::fromXContent; return CreateIndexResponse::fromXContent;
} }
@Override // fixed version from CreateIndexRequest - use only one mapping
protected CreateIndexResponse emptyResponse() { private XContentBuilder toXContent(CreateIndexRequest createIndexRequest,
return new CreateIndexResponse(); XContentBuilder builder) throws IOException {
builder.startObject();
builder.startObject(SETTINGS.getPreferredName());
createIndexRequest.settings().toXContent(builder, ToXContent.EMPTY_PARAMS);
builder.endObject();
// there is only an empty or a single entry for mappings
if (createIndexRequest.mappings().isEmpty()) {
// ES wants a mappings element with an empty map
builder.startObject(MAPPINGS.getPreferredName());
builder.endObject();
} else {
Map<String, ?> mappingAsMap = createIndexRequest.mappings();
String mappingString = mappingAsMap.values().iterator().next().toString();
builder.field(MAPPINGS.getPreferredName());
builder.map(XContentHelper.convertToMap(new BytesArray(mappingString), false, XContentType.JSON).v2());
}
builder.startObject(ALIASES.getPreferredName());
for (Alias alias : createIndexRequest.aliases()) {
alias.toXContent(builder, ToXContent.EMPTY_PARAMS);
}
builder.endObject();
builder.endObject();
return builder;
} }
} }

14
elx-http/src/main/java/org/xbib/elx/http/action/admin/indices/delete/HttpDeleteIndexAction.java
View file

@ -2,15 +2,16 @@ package org.xbib.elx.http.action.admin.indices.delete;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexAction; import org.elasticsearch.action.admin.indices.delete.DeleteIndexAction;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest; import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse; import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.common.CheckedFunction; import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
public class HttpDeleteIndexAction extends HttpAction<DeleteIndexRequest, DeleteIndexResponse> { public class HttpDeleteIndexAction extends HttpAction<DeleteIndexRequest, AcknowledgedResponse> {
@Override @Override
public DeleteIndexAction getActionInstance() { public DeleteIndexAction getActionInstance() {
@ -23,12 +24,7 @@ public class HttpDeleteIndexAction extends HttpAction<DeleteIndexRequest, Delete
} }
@Override @Override
protected CheckedFunction<XContentParser, DeleteIndexResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, AcknowledgedResponse, IOException> entityParser(HttpResponse httpResponse) {
return DeleteIndexResponse::fromXContent; return AcknowledgedResponse::fromXContent;
}
@Override
protected DeleteIndexResponse emptyResponse() {
return new DeleteIndexResponse();
} }
} }

30
elx-http/src/main/java/org/xbib/elx/http/action/admin/indices/exists/indices/HttpIndicesExistsAction.java
View file

@ -10,6 +10,7 @@ import org.elasticsearch.rest.RestStatus;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.elx.http.HttpActionContext; import org.xbib.elx.http.HttpActionContext;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
@ -26,22 +27,27 @@ public class HttpIndicesExistsAction extends HttpAction<IndicesExistsRequest, In
return newHeadRequest(url, index); return newHeadRequest(url, index);
} }
@Override /**
protected CheckedFunction<XContentParser, IndicesExistsResponse, IOException> entityParser() { * Override for non-body logic.
return this::fromXContent; * @param httpActionContext the HTTP action context
} * @return the ELasticsearch sttatus exception
*/
@Override
protected IndicesExistsResponse emptyResponse() {
return new IndicesExistsResponse(false); // used for 404 Not found
}
@Override @Override
protected ElasticsearchStatusException parseToError(HttpActionContext<IndicesExistsRequest, IndicesExistsResponse> httpActionContext) { protected ElasticsearchStatusException parseToError(HttpActionContext<IndicesExistsRequest, IndicesExistsResponse> httpActionContext) {
return new ElasticsearchStatusException("not found", RestStatus.NOT_FOUND); return new ElasticsearchStatusException("not found", RestStatus.NOT_FOUND);
} }
private IndicesExistsResponse fromXContent(XContentParser parser) throws IOException { @Override
return new IndicesExistsResponse(true); // used for 200 OK protected CheckedFunction<XContentParser, IndicesExistsResponse, IOException> entityParser(HttpResponse httpResponse) {
return httpResponse.getStatus().getCode() == 200 ? this::found : this::notfound;
} }
private IndicesExistsResponse found(XContentParser parser) {
return new IndicesExistsResponse(true);
}
private IndicesExistsResponse notfound(XContentParser parser) {
return new IndicesExistsResponse(false);
}
} }

38
elx-http/src/main/java/org/elasticsearch/action/admin/indices/mapping/get/HttpGetMappingsAction.java → elx-http/src/main/java/org/xbib/elx/http/action/admin/indices/mapping/get/HttpGetMappingsAction.java
View file

@ -1,5 +1,8 @@
package org.elasticsearch.action.admin.indices.mapping.get; package org.xbib.elx.http.action.admin.indices.mapping.get;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsAction;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse;
import org.elasticsearch.cluster.metadata.MappingMetaData; import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.elasticsearch.common.CheckedFunction; import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.ParseField; import org.elasticsearch.common.ParseField;
@ -7,12 +10,10 @@ import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
import java.util.Map; import java.util.Map;
import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
public class HttpGetMappingsAction extends HttpAction<GetMappingsRequest, GetMappingsResponse> { public class HttpGetMappingsAction extends HttpAction<GetMappingsRequest, GetMappingsResponse> {
private static final ParseField MAPPINGS = new ParseField("mappings"); private static final ParseField MAPPINGS = new ParseField("mappings");
@ -29,31 +30,24 @@ public class HttpGetMappingsAction extends HttpAction<GetMappingsRequest, GetMap
} }
@Override @Override
protected CheckedFunction<XContentParser, GetMappingsResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, GetMappingsResponse, IOException> entityParser(HttpResponse httpResponse) {
return this::fromXContent; return this::fromXContent;
} }
@Override // fixed version from GetMappingsRequest - use only one mapping per index with type "_doc"
protected GetMappingsResponse emptyResponse() {
return new GetMappingsResponse();
}
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
private GetMappingsResponse fromXContent(XContentParser parser) throws IOException { private GetMappingsResponse fromXContent(XContentParser parser) throws IOException {
XContentParser.Token token = parser.nextToken(); if (parser.currentToken() == null) {
ensureExpectedToken(XContentParser.Token.START_OBJECT, token, parser::getTokenLocation); parser.nextToken();
Map<String, Object> parts = parser.map(); }
Map<String, Object> map = parser.map();
ImmutableOpenMap.Builder<String, ImmutableOpenMap<String, MappingMetaData>> builder = new ImmutableOpenMap.Builder<>(); ImmutableOpenMap.Builder<String, ImmutableOpenMap<String, MappingMetaData>> builder = new ImmutableOpenMap.Builder<>();
for (Map.Entry<String, Object> entry : parts.entrySet()) { for (Map.Entry<String, Object> entry : map.entrySet()) {
String indexName = entry.getKey(); final String indexName = entry.getKey();
Map<String, Object> mapping = (Map<String, Object>) ((Map) entry.getValue()).get(MAPPINGS.getPreferredName()); final Map<String, Object> mapping = (Map<String, Object>) ((Map<String, Object>) entry.getValue()).get(MAPPINGS.getPreferredName());
ImmutableOpenMap.Builder<String, MappingMetaData> typeBuilder = new ImmutableOpenMap.Builder<>(); ImmutableOpenMap.Builder<String, MappingMetaData> typeBuilder = new ImmutableOpenMap.Builder<>();
for (Map.Entry<String, Object> typeEntry : mapping.entrySet()) { MappingMetaData mmd = new MappingMetaData("_doc", mapping);
String typeName = typeEntry.getKey(); typeBuilder.put("_doc", mmd);
Map<String, Object> fieldMappings = (Map<String, Object>) typeEntry.getValue();
MappingMetaData mmd = new MappingMetaData(typeName, fieldMappings);
typeBuilder.put(typeName, mmd);
}
builder.put(indexName, typeBuilder.build()); builder.put(indexName, typeBuilder.build());
} }
return new GetMappingsResponse(builder.build()); return new GetMappingsResponse(builder.build());

8
elx-http/src/main/java/org/xbib/elx/http/action/admin/indices/refresh/HttpRefreshIndexAction.java
View file

@ -7,6 +7,7 @@ import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
@ -24,12 +25,7 @@ public class HttpRefreshIndexAction extends HttpAction<RefreshRequest, RefreshRe
} }
@Override @Override
protected CheckedFunction<XContentParser, RefreshResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, RefreshResponse, IOException> entityParser(HttpResponse httpResponse) {
return RefreshResponse::fromXContent; return RefreshResponse::fromXContent;
} }
@Override
protected RefreshResponse emptyResponse() {
return new RefreshResponse();
}
} }

65
elx-http/src/main/java/org/xbib/elx/http/action/admin/indices/settings/get/HttpGetSettingsAction.java
View file

@ -4,16 +4,12 @@ import org.elasticsearch.action.admin.indices.settings.get.GetSettingsAction;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsRequest; import org.elasticsearch.action.admin.indices.settings.get.GetSettingsRequest;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse; import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse;
import org.elasticsearch.common.CheckedFunction; import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentParserUtils;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
public class HttpGetSettingsAction extends HttpAction<GetSettingsRequest, GetSettingsResponse> { public class HttpGetSettingsAction extends HttpAction<GetSettingsRequest, GetSettingsResponse> {
@ -30,62 +26,7 @@ public class HttpGetSettingsAction extends HttpAction<GetSettingsRequest, GetSet
} }
@Override @Override
protected CheckedFunction<XContentParser, GetSettingsResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, GetSettingsResponse, IOException> entityParser(HttpResponse httpResponse) {
return this::fromXContent; return GetSettingsResponse::fromXContent;
}
@Override
protected GetSettingsResponse emptyResponse() {
ImmutableOpenMap<String, Settings> settingsMap = ImmutableOpenMap.<String, Settings>builder().build();
return new GetSettingsResponse(settingsMap);
}
private GetSettingsResponse fromXContent(XContentParser parser) throws IOException {
Map<String, Settings> indexToSettings = new HashMap<>();
Map<String, Settings> indexToDefaultSettings = new HashMap<>();
if (parser.currentToken() == null) {
parser.nextToken();
}
XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.currentToken(), parser::getTokenLocation);
parser.nextToken();
while (!parser.isClosed()) {
if (parser.currentToken() == XContentParser.Token.START_OBJECT) {
parseIndexEntry(parser, indexToSettings, indexToDefaultSettings);
} else if (parser.currentToken() == XContentParser.Token.START_ARRAY) {
parser.skipChildren();
} else {
parser.nextToken();
}
}
ImmutableOpenMap<String, Settings> settingsMap = ImmutableOpenMap.<String, Settings>builder().putAll(indexToSettings).build();
return new GetSettingsResponse(settingsMap);
}
private static void parseIndexEntry(XContentParser parser, Map<String, Settings> indexToSettings,
Map<String, Settings> indexToDefaultSettings) throws IOException {
String indexName = parser.currentName();
parser.nextToken();
while (!parser.isClosed() && parser.currentToken() != XContentParser.Token.END_OBJECT) {
parseSettingsField(parser, indexName, indexToSettings, indexToDefaultSettings);
}
}
private static void parseSettingsField(XContentParser parser, String currentIndexName, Map<String, Settings> indexToSettings,
Map<String, Settings> indexToDefaultSettings) throws IOException {
if (parser.currentToken() == XContentParser.Token.START_OBJECT) {
switch (parser.currentName()) {
case "settings":
indexToSettings.put(currentIndexName, Settings.fromXContent(parser));
break;
case "defaults":
indexToDefaultSettings.put(currentIndexName, Settings.fromXContent(parser));
break;
default:
parser.skipChildren();
}
} else if (parser.currentToken() == XContentParser.Token.START_ARRAY) {
parser.skipChildren();
}
parser.nextToken();
} }
} }

14
elx-http/src/main/java/org/xbib/elx/http/action/admin/indices/settings/put/HttpUpdateSettingsAction.java
View file

@ -2,7 +2,7 @@ package org.xbib.elx.http.action.admin.indices.settings.put;
import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsAction; import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsAction;
import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest; import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest;
import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsResponse; import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.common.CheckedFunction; import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.xcontent.ToXContent; import org.elasticsearch.common.xcontent.ToXContent;
@ -11,11 +11,12 @@ import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
import java.io.UncheckedIOException; import java.io.UncheckedIOException;
public class HttpUpdateSettingsAction extends HttpAction<UpdateSettingsRequest, UpdateSettingsResponse> { public class HttpUpdateSettingsAction extends HttpAction<UpdateSettingsRequest, AcknowledgedResponse> {
@Override @Override
public UpdateSettingsAction getActionInstance() { public UpdateSettingsAction getActionInstance() {
@ -37,12 +38,7 @@ public class HttpUpdateSettingsAction extends HttpAction<UpdateSettingsRequest,
} }
@Override @Override
protected CheckedFunction<XContentParser, UpdateSettingsResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, AcknowledgedResponse, IOException> entityParser(HttpResponse httpResponse) {
return UpdateSettingsResponse::fromXContent; return AcknowledgedResponse::fromXContent;
}
@Override
protected UpdateSettingsResponse emptyResponse() {
return new UpdateSettingsResponse();
} }
} }

22
elx-http/src/main/java/org/xbib/elx/http/action/bulk/HttpBulkAction.java
View file

@ -14,6 +14,7 @@ import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType; import org.elasticsearch.common.xcontent.XContentType;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
@ -32,16 +33,13 @@ public class HttpBulkAction extends HttpAction<BulkRequest, BulkResponse> {
IndexRequest indexRequest = (IndexRequest) actionRequest; IndexRequest indexRequest = (IndexRequest) actionRequest;
bulkContent.append("{\"").append(indexRequest.opType().getLowercase()).append("\":{"); bulkContent.append("{\"").append(indexRequest.opType().getLowercase()).append("\":{");
bulkContent.append("\"_index\":\"").append(indexRequest.index()).append("\""); bulkContent.append("\"_index\":\"").append(indexRequest.index()).append("\"");
bulkContent.append(",\"_type\":\"").append(indexRequest.type()).append("\""); bulkContent.append(",\"_type\":\"").append("_doc").append("\"");
if (indexRequest.id() != null) { if (indexRequest.id() != null) {
bulkContent.append(",\"_id\":\"").append(indexRequest.id()).append("\""); bulkContent.append(",\"_id\":\"").append(indexRequest.id()).append("\"");
} }
if (indexRequest.routing() != null) { if (indexRequest.routing() != null) {
bulkContent.append(",\"_routing\":\"").append(indexRequest.routing()).append("\""); bulkContent.append(",\"_routing\":\"").append(indexRequest.routing()).append("\"");
} }
if (indexRequest.parent() != null) {
bulkContent.append(",\"_parent\":\"").append(indexRequest.parent()).append("\"");
}
if (indexRequest.version() > 0) { if (indexRequest.version() > 0) {
bulkContent.append(",\"_version\":\"").append(indexRequest.version()).append("\""); bulkContent.append(",\"_version\":\"").append(indexRequest.version()).append("\"");
if (indexRequest.versionType() != null) { if (indexRequest.versionType() != null) {
@ -55,14 +53,11 @@ public class HttpBulkAction extends HttpAction<BulkRequest, BulkResponse> {
UpdateRequest updateRequest = (UpdateRequest) actionRequest; UpdateRequest updateRequest = (UpdateRequest) actionRequest;
bulkContent.append("{\"update\":{"); bulkContent.append("{\"update\":{");
bulkContent.append("\"_index\":\"").append(updateRequest.index()).append("\""); bulkContent.append("\"_index\":\"").append(updateRequest.index()).append("\"");
bulkContent.append(",\"_type\":\"").append(updateRequest.type()).append("\""); bulkContent.append(",\"_type\":\"").append("_doc").append("\"");
bulkContent.append(",\"_id\":\"").append(updateRequest.id()).append("\""); bulkContent.append(",\"_id\":\"").append(updateRequest.id()).append("\"");
if (updateRequest.routing() != null) { if (updateRequest.routing() != null) {
bulkContent.append(",\"_routing\":\"").append(updateRequest.routing()).append("\""); bulkContent.append(",\"_routing\":\"").append(updateRequest.routing()).append("\"");
} }
if (updateRequest.parent() != null) {
bulkContent.append(",\"_parent\":\"").append(updateRequest.parent()).append("\"");
}
if (updateRequest.version() > 0) { if (updateRequest.version() > 0) {
bulkContent.append(",\"_version\":\"").append(updateRequest.version()).append("\""); bulkContent.append(",\"_version\":\"").append(updateRequest.version()).append("\"");
if (updateRequest.versionType() != null) { if (updateRequest.versionType() != null) {
@ -80,7 +75,7 @@ public class HttpBulkAction extends HttpAction<BulkRequest, BulkResponse> {
DeleteRequest deleteRequest = (DeleteRequest) actionRequest; DeleteRequest deleteRequest = (DeleteRequest) actionRequest;
bulkContent.append("{\"delete\":{"); bulkContent.append("{\"delete\":{");
bulkContent.append("\"_index\":\"").append(deleteRequest.index()).append("\""); bulkContent.append("\"_index\":\"").append(deleteRequest.index()).append("\"");
bulkContent.append(",\"_type\":\"").append(deleteRequest.type()).append("\""); bulkContent.append(",\"_type\":\"").append("_doc").append("\"");
bulkContent.append(",\"_id\":\"").append(deleteRequest.id()).append("\""); bulkContent.append(",\"_id\":\"").append(deleteRequest.id()).append("\"");
if (deleteRequest.routing() != null) { if (deleteRequest.routing() != null) {
bulkContent.append(",\"_routing\":\"").append(deleteRequest.routing()).append("\""); // _routing bulkContent.append(",\"_routing\":\"").append(deleteRequest.routing()).append("\""); // _routing
@ -92,14 +87,7 @@ public class HttpBulkAction extends HttpAction<BulkRequest, BulkResponse> {
} }
@Override @Override
protected CheckedFunction<XContentParser, BulkResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, BulkResponse, IOException> entityParser(HttpResponse httpResponse) {
return BulkResponse::fromXContent; return BulkResponse::fromXContent;
} }
@Override
protected BulkResponse emptyResponse() {
BulkItemResponse[] responses = null;
long took = 0L;
return new BulkResponse(responses, took);
}
} }

14
elx-http/src/main/java/org/xbib/elx/http/action/get/HttpExistsAction.java
View file

@ -1,6 +1,6 @@
package org.xbib.elx.http.action.get; package org.xbib.elx.http.action.get;
import org.elasticsearch.action.GenericAction; import org.elasticsearch.action.ActionType;
import org.elasticsearch.action.get.GetAction; import org.elasticsearch.action.get.GetAction;
import org.elasticsearch.action.get.GetRequest; import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse; import org.elasticsearch.action.get.GetResponse;
@ -8,28 +8,24 @@ import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
public class HttpExistsAction extends HttpAction<GetRequest, GetResponse> { public class HttpExistsAction extends HttpAction<GetRequest, GetResponse> {
@Override @Override
public GenericAction<GetRequest, GetResponse> getActionInstance() { public ActionType<GetResponse> getActionInstance() {
return GetAction.INSTANCE; return GetAction.INSTANCE;
} }
@Override @Override
protected Request.Builder createHttpRequest(String url, GetRequest request) { protected Request.Builder createHttpRequest(String url, GetRequest request) {
return newHeadRequest(url, "/" + request.index() + "/" + request.type() + "/" + request.id()); return newHeadRequest(url, "/" + request.index() + "/_doc/" + request.id());
} }
@Override @Override
protected CheckedFunction<XContentParser, GetResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, GetResponse, IOException> entityParser(HttpResponse httpResponse) {
return GetResponse::fromXContent; return GetResponse::fromXContent;
} }
@Override
protected GetResponse emptyResponse() {
return new GetResponse();
}
} }

14
elx-http/src/main/java/org/xbib/elx/http/action/get/HttpGetAction.java
View file

@ -1,6 +1,6 @@
package org.xbib.elx.http.action.get; package org.xbib.elx.http.action.get;
import org.elasticsearch.action.GenericAction; import org.elasticsearch.action.ActionType;
import org.elasticsearch.action.get.GetAction; import org.elasticsearch.action.get.GetAction;
import org.elasticsearch.action.get.GetRequest; import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse; import org.elasticsearch.action.get.GetResponse;
@ -8,28 +8,24 @@ import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
public class HttpGetAction extends HttpAction<GetRequest, GetResponse> { public class HttpGetAction extends HttpAction<GetRequest, GetResponse> {
@Override @Override
public GenericAction<GetRequest, GetResponse> getActionInstance() { public ActionType<GetResponse> getActionInstance() {
return GetAction.INSTANCE; return GetAction.INSTANCE;
} }
@Override @Override
protected Request.Builder createHttpRequest(String url, GetRequest request) { protected Request.Builder createHttpRequest(String url, GetRequest request) {
return newGetRequest(url, "/" + request.index() + "/" + request.type() + "/" + request.id()); return newGetRequest(url, "/" + request.index() + "/_doc/" + "/" + request.id());
} }
@Override @Override
protected CheckedFunction<XContentParser, GetResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, GetResponse, IOException> entityParser(HttpResponse httpResponse) {
return GetResponse::fromXContent; return GetResponse::fromXContent;
} }
@Override
protected GetResponse emptyResponse() {
return new GetResponse();
}
} }

14
elx-http/src/main/java/org/xbib/elx/http/action/index/HttpIndexAction.java
View file

@ -1,6 +1,6 @@
package org.xbib.elx.http.action.index; package org.xbib.elx.http.action.index;
import org.elasticsearch.action.GenericAction; import org.elasticsearch.action.ActionType;
import org.elasticsearch.action.index.IndexAction; import org.elasticsearch.action.index.IndexAction;
import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.action.index.IndexResponse;
@ -8,29 +8,25 @@ import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
public class HttpIndexAction extends HttpAction<IndexRequest, IndexResponse> { public class HttpIndexAction extends HttpAction<IndexRequest, IndexResponse> {
@Override @Override
public GenericAction<IndexRequest, IndexResponse> getActionInstance() { public ActionType<IndexResponse> getActionInstance() {
return IndexAction.INSTANCE; return IndexAction.INSTANCE;
} }
@Override @Override
protected Request.Builder createHttpRequest(String url, IndexRequest request) { protected Request.Builder createHttpRequest(String url, IndexRequest request) {
return newPutRequest(url, "/" + request.index() + "/" + request.type() + "/" + request.id(), return newPutRequest(url, "/" + request.index() + "/_doc/" + request.id(),
request.source()); request.source());
} }
@Override @Override
protected CheckedFunction<XContentParser, IndexResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, IndexResponse, IOException> entityParser(HttpResponse httpResponse) {
return IndexResponse::fromXContent; return IndexResponse::fromXContent;
} }
@Override
protected IndexResponse emptyResponse() {
return new IndexResponse();
}
} }

12
elx-http/src/main/java/org/xbib/elx/http/action/main/HttpMainAction.java
View file

@ -1,6 +1,6 @@
package org.xbib.elx.http.action.main; package org.xbib.elx.http.action.main;
import org.elasticsearch.action.GenericAction; import org.elasticsearch.action.ActionType;
import org.elasticsearch.action.main.MainAction; import org.elasticsearch.action.main.MainAction;
import org.elasticsearch.action.main.MainRequest; import org.elasticsearch.action.main.MainRequest;
import org.elasticsearch.action.main.MainResponse; import org.elasticsearch.action.main.MainResponse;
@ -8,13 +8,14 @@ import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
public class HttpMainAction extends HttpAction<MainRequest, MainResponse> { public class HttpMainAction extends HttpAction<MainRequest, MainResponse> {
@Override @Override
public GenericAction<MainRequest, MainResponse> getActionInstance() { public ActionType<MainResponse> getActionInstance() {
return MainAction.INSTANCE; return MainAction.INSTANCE;
} }
@ -24,12 +25,7 @@ public class HttpMainAction extends HttpAction<MainRequest, MainResponse> {
} }
@Override @Override
protected CheckedFunction<XContentParser, MainResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, MainResponse, IOException> entityParser(HttpResponse httpResponse) {
return MainResponse::fromXContent; return MainResponse::fromXContent;
} }
@Override
protected MainResponse emptyResponse() {
return new MainResponse();
}
} }

8
elx-http/src/main/java/org/xbib/elx/http/action/search/HttpSearchAction.java
View file

@ -7,6 +7,7 @@ import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
@ -25,12 +26,7 @@ public class HttpSearchAction extends HttpAction<SearchRequest, SearchResponse>
} }
@Override @Override
protected CheckedFunction<XContentParser, SearchResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, SearchResponse, IOException> entityParser(HttpResponse httpResponse) {
return SearchResponse::fromXContent; return SearchResponse::fromXContent;
} }
@Override
protected SearchResponse emptyResponse() {
return new SearchResponse();
}
} }

14
elx-http/src/main/java/org/xbib/elx/http/action/update/HttpUpdateAction.java
View file

@ -1,6 +1,6 @@
package org.xbib.elx.http.action.update; package org.xbib.elx.http.action.update;
import org.elasticsearch.action.GenericAction; import org.elasticsearch.action.ActionType;
import org.elasticsearch.action.update.UpdateAction; import org.elasticsearch.action.update.UpdateAction;
import org.elasticsearch.action.update.UpdateRequest; import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse; import org.elasticsearch.action.update.UpdateResponse;
@ -12,13 +12,14 @@ import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType; import org.elasticsearch.common.xcontent.XContentType;
import org.xbib.elx.http.HttpAction; import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request; import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException; import java.io.IOException;
public class HttpUpdateAction extends HttpAction<UpdateRequest, UpdateResponse> { public class HttpUpdateAction extends HttpAction<UpdateRequest, UpdateResponse> {
@Override @Override
public GenericAction<UpdateRequest, UpdateResponse> getActionInstance() { public ActionType<UpdateResponse> getActionInstance() {
return UpdateAction.INSTANCE; return UpdateAction.INSTANCE;
} }
@ -47,7 +48,7 @@ public class HttpUpdateAction extends HttpAction<UpdateRequest, UpdateResponse>
} }
BytesReference source = XContentHelper.toXContent(updateRequest, xContentType, false); BytesReference source = XContentHelper.toXContent(updateRequest, xContentType, false);
return newPostRequest(url, return newPostRequest(url,
"/" + updateRequest.index() + "/" + updateRequest.type() + "/" + updateRequest.id() + "/_update", "/" + updateRequest.index() + "/_doc/" + updateRequest.id() + "/_update",
source); source);
} catch (IOException e) { } catch (IOException e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
@ -56,12 +57,7 @@ public class HttpUpdateAction extends HttpAction<UpdateRequest, UpdateResponse>
} }
@Override @Override
protected CheckedFunction<XContentParser, UpdateResponse, IOException> entityParser() { protected CheckedFunction<XContentParser, UpdateResponse, IOException> entityParser(HttpResponse httpResponse) {
return UpdateResponse::fromXContent; return UpdateResponse::fromXContent;
} }
@Override
protected UpdateResponse emptyResponse() {
return new UpdateResponse();
}
} }

2
elx-http/src/main/resources/META-INF/services/org.xbib.elx.http.HttpAction
View file

@ -1,4 +1,4 @@
org.elasticsearch.action.admin.indices.mapping.get.HttpGetMappingsAction org.xbib.elx.http.action.admin.indices.mapping.get.HttpGetMappingsAction
org.xbib.elx.http.action.admin.cluster.health.HttpClusterHealthAction org.xbib.elx.http.action.admin.cluster.health.HttpClusterHealthAction
org.xbib.elx.http.action.admin.cluster.node.info.HttpNodesInfoAction org.xbib.elx.http.action.admin.cluster.node.info.HttpNodesInfoAction
org.xbib.elx.http.action.admin.cluster.settings.HttpClusterUpdateSettingsAction org.xbib.elx.http.action.admin.cluster.settings.HttpClusterUpdateSettingsAction

68
elx-http/src/test/java/org/xbib/elx/http/test/ClientTest.java
View file

@ -1,19 +1,25 @@
package org.xbib.elx.http.test; package org.xbib.elx.http.test;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsAction; import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsAction;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest; import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse; import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsAction;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsRequest;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse;
import org.elasticsearch.action.search.SearchAction; import org.elasticsearch.action.search.SearchAction;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchRequestBuilder; import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.NoNodeAvailableException; import org.elasticsearch.client.transport.NoNodeAvailableException;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder; import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.json.JsonXContent; import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.api.extension.ExtendWith;
import org.xbib.elx.common.ClientBuilder; import org.xbib.elx.common.ClientBuilder;
@ -21,6 +27,7 @@ import org.xbib.elx.common.Parameters;
import org.xbib.elx.http.ExtendedHttpClient; import org.xbib.elx.http.ExtendedHttpClient;
import org.xbib.elx.http.ExtendedHttpClientProvider; import org.xbib.elx.http.ExtendedHttpClientProvider;
import java.util.Map;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
@ -82,28 +89,50 @@ class ClientTest {
} }
@Test @Test
void testMapping() throws Exception { void testNewIndexWithSettings() throws Exception {
final ExtendedHttpClient client = ClientBuilder.builder() final ExtendedHttpClient client = ClientBuilder.builder()
.provider(ExtendedHttpClientProvider.class) .provider(ExtendedHttpClientProvider.class)
.put(helper.getHttpSettings()) .put(helper.getHttpSettings())
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(5)) .put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(5))
.build(); .build();
Settings settings = Settings.builder().put("index.number_of_shards", "1").build();
client.newIndex("test", settings);
GetSettingsRequest getSettingsRequest = new GetSettingsRequest()
.indices("test");
GetSettingsResponse getSettingsResponse =
client.getClient().execute(GetSettingsAction.INSTANCE, getSettingsRequest).actionGet();
logger.log(Level.INFO, "settings=" + getSettingsResponse.getSetting("test", "index.number_of_shards"));
assertEquals("1", getSettingsResponse.getSetting("test", "index.number_of_shards"));
client.close();
}
@Test
void testNewIndexWithSettingsAndMappings() throws Exception {
final ExtendedHttpClient client = ClientBuilder.builder()
.provider(ExtendedHttpClientProvider.class)
.put(helper.getHttpSettings())
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(5))
.build();
Settings settings = Settings.builder().put("index.number_of_shards", "1").build();
XContentBuilder builder = JsonXContent.contentBuilder() XContentBuilder builder = JsonXContent.contentBuilder()
.startObject() .startObject()
.startObject("doc")
.startObject("properties") .startObject("properties")
.startObject("location") .startObject("pos")
.field("type", "geo_point") .field("type", "geo_point")
.endObject() .endObject()
.endObject() .endObject()
.endObject()
.endObject(); .endObject();
client.newIndex("test", Settings.EMPTY, Strings.toString(builder)); client.newIndex("test", settings, builder);
GetMappingsRequest getMappingsRequest = new GetMappingsRequest().indices("test"); GetSettingsRequest getSettingsRequest = new GetSettingsRequest()
.indices("test");
GetSettingsResponse getSettingsResponse =
client.getClient().execute(GetSettingsAction.INSTANCE, getSettingsRequest).actionGet();
assertEquals("1", getSettingsResponse.getSetting("test", "index.number_of_shards"));
GetMappingsRequest getMappingsRequest = new GetMappingsRequest()
.indices("test");
GetMappingsResponse getMappingsResponse = GetMappingsResponse getMappingsResponse =
client.getClient().execute(GetMappingsAction.INSTANCE, getMappingsRequest).actionGet(); client.getClient().execute(GetMappingsAction.INSTANCE, getMappingsRequest).actionGet();
logger.info("mappings={}", getMappingsResponse.getMappings()); assertTrue(getMappingsResponse.getMappings().get("test").containsKey("_doc"));
assertTrue(getMappingsResponse.getMappings().get("test").containsKey("doc"));
client.close(); client.close();
} }
@ -136,8 +165,7 @@ class ClientTest {
client.refreshIndex("test"); client.refreshIndex("test");
SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client.getClient(), SearchAction.INSTANCE) SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client.getClient(), SearchAction.INSTANCE)
.setQuery(QueryBuilders.matchAllQuery()).setSize(0); .setQuery(QueryBuilders.matchAllQuery()).setSize(0);
assertEquals(numactions, assertEquals(numactions, searchRequestBuilder.execute().actionGet().getHits().getTotalHits().value);
searchRequestBuilder.execute().actionGet().getHits().getTotalHits());
client.close(); client.close();
} }
} }
@ -146,7 +174,7 @@ class ClientTest {
void testThreadedRandomDocs() throws Exception { void testThreadedRandomDocs() throws Exception {
int maxthreads = Runtime.getRuntime().availableProcessors(); int maxthreads = Runtime.getRuntime().availableProcessors();
Long maxActionsPerRequest = MAX_ACTIONS_PER_REQUEST; Long maxActionsPerRequest = MAX_ACTIONS_PER_REQUEST;
final Long actions = ACTIONS; final long actions = ACTIONS;
logger.info("maxthreads={} maxactions={} maxloop={}", maxthreads, maxActionsPerRequest, actions); logger.info("maxthreads={} maxactions={} maxloop={}", maxthreads, maxActionsPerRequest, actions);
final ExtendedHttpClient client = ClientBuilder.builder() final ExtendedHttpClient client = ClientBuilder.builder()
.provider(ExtendedHttpClientProvider.class) .provider(ExtendedHttpClientProvider.class)
@ -196,11 +224,19 @@ class ClientTest {
logger.error("error", client.getBulkController().getLastBulkError()); logger.error("error", client.getBulkController().getLastBulkError());
} }
assertNull(client.getBulkController().getLastBulkError()); assertNull(client.getBulkController().getLastBulkError());
assertEquals(maxthreads * actions, client.getBulkMetric().getSucceeded().getCount());
logger.log(Level.INFO, "refreshing index test");
client.refreshIndex("test"); client.refreshIndex("test");
SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client.getClient(), SearchAction.INSTANCE) SearchSourceBuilder builder = new SearchSourceBuilder()
.setQuery(QueryBuilders.matchAllQuery()).setSize(0); .query(QueryBuilders.matchAllQuery())
assertEquals(maxthreads * actions, .size(0)
searchRequestBuilder.execute().actionGet().getHits().getTotalHits()); .trackTotalHits(true);
SearchRequest searchRequest = new SearchRequest()
.indices("test")
.source(builder);
SearchResponse searchResponse = client.getClient().execute(SearchAction.INSTANCE, searchRequest).actionGet();
assertEquals(maxthreads * actions, searchResponse.getHits().getTotalHits().value);
client.close();
client.close(); client.close();
} }
} }

23
elx-http/src/test/java/org/xbib/elx/http/test/DuplicateIDTest.java
View file

@ -3,8 +3,11 @@ package org.xbib.elx.http.test;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.search.SearchAction; import org.elasticsearch.action.search.SearchAction;
import org.elasticsearch.action.search.SearchRequestBuilder; import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.NoNodeAvailableException; import org.elasticsearch.client.transport.NoNodeAvailableException;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.api.extension.ExtendWith;
import org.xbib.elx.common.ClientBuilder; import org.xbib.elx.common.ClientBuilder;
@ -14,7 +17,6 @@ import org.xbib.elx.http.ExtendedHttpClientProvider;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.assertTrue;
@ -26,7 +28,7 @@ class DuplicateIDTest {
private static final Long MAX_ACTIONS_PER_REQUEST = 10L; private static final Long MAX_ACTIONS_PER_REQUEST = 10L;
private static final Long ACTIONS = 50L; private static final Long ACTIONS = 100L;
private final TestExtension.Helper helper; private final TestExtension.Helper helper;
@ -51,11 +53,16 @@ class DuplicateIDTest {
client.flush(); client.flush();
client.waitForResponses(30L, TimeUnit.SECONDS); client.waitForResponses(30L, TimeUnit.SECONDS);
client.refreshIndex("test"); client.refreshIndex("test");
SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client.getClient(), SearchAction.INSTANCE) SearchSourceBuilder builder = new SearchSourceBuilder()
.setIndices("test") .query(QueryBuilders.matchAllQuery())
.setTypes("test") .size(0)
.setQuery(matchAllQuery()); .trackTotalHits(true);
long hits = searchRequestBuilder.execute().actionGet().getHits().getTotalHits(); SearchRequest searchRequest = new SearchRequest()
.indices("test")
.source(builder);
SearchResponse searchResponse =
helper.client("1").execute(SearchAction.INSTANCE, searchRequest).actionGet();
long hits = searchResponse.getHits().getTotalHits().value;
logger.info("hits = {}", hits); logger.info("hits = {}", hits);
assertTrue(hits < ACTIONS); assertTrue(hits < ACTIONS);
} catch (NoNodeAvailableException e) { } catch (NoNodeAvailableException e) {

2
elx-http/src/test/java/org/xbib/elx/http/test/IndexPruneTest.java
View file

@ -66,9 +66,9 @@ class IndexPruneTest {
indicesExistsRequest.indices(index); indicesExistsRequest.indices(index);
IndicesExistsResponse indicesExistsResponse = IndicesExistsResponse indicesExistsResponse =
client.getClient().execute(IndicesExistsAction.INSTANCE, indicesExistsRequest).actionGet(); client.getClient().execute(IndicesExistsAction.INSTANCE, indicesExistsRequest).actionGet();
logger.info("indices exists response for {} is {}", index, indicesExistsResponse.isExists());
list.add(indicesExistsResponse.isExists()); list.add(indicesExistsResponse.isExists());
} }
logger.info(list);
assertFalse(list.get(0)); assertFalse(list.get(0));
assertFalse(list.get(1)); assertFalse(list.get(1));
assertTrue(list.get(2)); assertTrue(list.get(2));

4
elx-http/src/test/java/org/xbib/elx/http/test/MockNode.java
View file

@ -6,10 +6,12 @@ import org.elasticsearch.node.Node;
import org.elasticsearch.plugins.Plugin; import org.elasticsearch.plugins.Plugin;
import java.util.List; import java.util.List;
import java.util.Map;
public class MockNode extends Node { public class MockNode extends Node {
public MockNode(Settings settings, List<Class<? extends Plugin>> classpathPlugins) { public MockNode(Settings settings, List<Class<? extends Plugin>> classpathPlugins) {
super(InternalSettingsPreparer.prepareEnvironment(settings, null), classpathPlugins); super(InternalSettingsPreparer.prepareEnvironment(settings, Map.of(), null, () -> "mock"),
classpathPlugins, false);
} }
} }

11
elx-http/src/test/java/org/xbib/elx/http/test/TestExtension.java
View file

@ -12,7 +12,6 @@ import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
import org.elasticsearch.action.admin.cluster.state.ClusterStateAction; import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest; import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse; import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.analysis.common.CommonAnalysisPlugin;
import org.elasticsearch.client.ElasticsearchClient; import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.client.support.AbstractClient; import org.elasticsearch.client.support.AbstractClient;
import org.elasticsearch.cluster.health.ClusterHealthStatus; import org.elasticsearch.cluster.health.ClusterHealthStatus;
@ -22,7 +21,7 @@ import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.node.Node; import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeValidationException; import org.elasticsearch.node.NodeValidationException;
import org.elasticsearch.plugins.Plugin; import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.transport.netty4.Netty4Plugin; import org.elasticsearch.transport.Netty4Plugin;
import org.junit.jupiter.api.extension.AfterEachCallback; import org.junit.jupiter.api.extension.AfterEachCallback;
import org.junit.jupiter.api.extension.BeforeEachCallback; import org.junit.jupiter.api.extension.BeforeEachCallback;
import org.junit.jupiter.api.extension.ExtensionContext; import org.junit.jupiter.api.extension.ExtensionContext;
@ -37,7 +36,7 @@ import java.nio.file.Path;
import java.nio.file.Paths; import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor; import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes; import java.nio.file.attribute.BasicFileAttributes;
import java.util.Arrays; import java.util.Collections;
import java.util.HashMap; import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
@ -192,8 +191,8 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
return Settings.builder() return Settings.builder()
.put("cluster.name", getClusterName()) .put("cluster.name", getClusterName())
.put("path.home", getHome()) .put("path.home", getHome())
.put("discovery.zen.master_election.ignore_non_master_pings", "true") .put("cluster.initial_master_nodes", "1")
.put("transport.netty.epoll", "false") .put("discovery.seed_hosts", "127.0.0.1:9300")
.build(); .build();
} }
@ -228,7 +227,7 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
.put(getNodeSettings()) .put(getNodeSettings())
.put("node.name", id) .put("node.name", id)
.build(); .build();
List<Class<? extends Plugin>> plugins = Arrays.asList(CommonAnalysisPlugin.class, Netty4Plugin.class); List<Class<? extends Plugin>> plugins = Collections.singletonList(Netty4Plugin.class);
Node node = new MockNode(nodeSettings, plugins); Node node = new MockNode(nodeSettings, plugins);
AbstractClient client = (AbstractClient) node.client(); AbstractClient client = (AbstractClient) node.client();
nodes.put(id, node); nodes.put(id, node);

5
elx-node/build.gradle
View file

@ -1,5 +1,6 @@
dependencies { dependencies {
compile project(':elx-common') compile project(':elx-common')
compile "org.xbib.elasticsearch:transport-netty4:${rootProject.property('elasticsearch-server.version')}" compile "org.elasticsearch.plugin:transport-netty4-client:${project.property('elasticsearch.version')}"
testCompile "org.xbib.elasticsearch:elasticsearch-analysis-common:${rootProject.property('elasticsearch-server.version')}" compile "io.netty:netty-codec-http:${project.property('netty.version')}"
compile "io.netty:netty-transport:${project.property('netty.version')}"
} }

2
elx-node/src/main/java/org/xbib/elx/node/ExtendedNodeClient.java
View file

@ -63,7 +63,7 @@ public class ExtendedNodeClient extends AbstractExtendedClient {
private static class BulkNode extends Node { private static class BulkNode extends Node {
BulkNode(Environment env, Collection<Class<? extends Plugin>> classpathPlugins) { BulkNode(Environment env, Collection<Class<? extends Plugin>> classpathPlugins) {
super(env, classpathPlugins); super(env, classpathPlugins, false);
} }
} }
} }

61
elx-node/src/test/java/org/xbib/elx/node/test/ClientTest.java
View file

@ -1,19 +1,25 @@
package org.xbib.elx.node.test; package org.xbib.elx.node.test;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsAction; import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsAction;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest; import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse; import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsAction;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsRequest;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse;
import org.elasticsearch.action.search.SearchAction; import org.elasticsearch.action.search.SearchAction;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchRequestBuilder; import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.NoNodeAvailableException; import org.elasticsearch.client.transport.NoNodeAvailableException;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder; import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.json.JsonXContent; import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.api.extension.ExtendWith;
import org.xbib.elx.common.ClientBuilder; import org.xbib.elx.common.ClientBuilder;
@ -80,27 +86,50 @@ class ClientTest {
} }
@Test @Test
void testMapping() throws Exception { void testNewIndexWithSettings() throws Exception {
final ExtendedNodeClient client = ClientBuilder.builder(helper.client("1")) final ExtendedNodeClient client = ClientBuilder.builder(helper.client("1"))
.provider(ExtendedNodeClientProvider.class) .provider(ExtendedNodeClientProvider.class)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(5)) .put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(5))
.build(); .build();
Settings settings = Settings.builder().put("index.number_of_shards", "1").build();
client.newIndex("test", settings);
GetSettingsRequest getSettingsRequest = new GetSettingsRequest()
.indices("test");
GetSettingsResponse getSettingsResponse =
client.getClient().execute(GetSettingsAction.INSTANCE, getSettingsRequest).actionGet();
logger.log(Level.INFO, "settings=" + getSettingsResponse.getSetting("test", "index.number_of_shards"));
assertEquals("1", getSettingsResponse.getSetting("test", "index.number_of_shards"));
client.close();
}
@Test
void testNewIndexWithSettingsAndMappings() throws Exception {
final ExtendedNodeClient client = ClientBuilder.builder(helper.client("1"))
.provider(ExtendedNodeClientProvider.class)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(5))
.build();
Settings settings = Settings.builder().put("index.number_of_shards", "1").build();
XContentBuilder builder = JsonXContent.contentBuilder() XContentBuilder builder = JsonXContent.contentBuilder()
.startObject() .startObject()
.startObject("doc")
.startObject("properties") .startObject("properties")
.startObject("location") .startObject("location")
.field("type", "geo_point") .field("type", "geo_point")
.endObject() .endObject()
.endObject() .endObject()
.endObject()
.endObject(); .endObject();
client.newIndex("test", Settings.EMPTY, Strings.toString(builder)); client.newIndex("test", settings, builder);
GetMappingsRequest getMappingsRequest = new GetMappingsRequest().indices("test"); GetSettingsRequest getSettingsRequest = new GetSettingsRequest()
.indices("test");
GetSettingsResponse getSettingsResponse =
client.getClient().execute(GetSettingsAction.INSTANCE, getSettingsRequest).actionGet();
logger.log(Level.INFO, "settings=" + getSettingsResponse.getSetting("test", "index.number_of_shards"));
assertEquals("1", getSettingsResponse.getSetting("test", "index.number_of_shards"));
GetMappingsRequest getMappingsRequest = new GetMappingsRequest()
.indices("test");
GetMappingsResponse getMappingsResponse = GetMappingsResponse getMappingsResponse =
client.getClient().execute(GetMappingsAction.INSTANCE, getMappingsRequest).actionGet(); client.getClient().execute(GetMappingsAction.INSTANCE, getMappingsRequest).actionGet();
logger.info("mappings={}", getMappingsResponse.getMappings()); logger.info("mappings={}", getMappingsResponse.getMappings());
assertTrue(getMappingsResponse.getMappings().get("test").containsKey("doc")); assertTrue(getMappingsResponse.getMappings().get("test").containsKey("_doc"));
client.close(); client.close();
} }
@ -131,8 +160,7 @@ class ClientTest {
client.refreshIndex("test"); client.refreshIndex("test");
SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client.getClient(), SearchAction.INSTANCE) SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client.getClient(), SearchAction.INSTANCE)
.setQuery(QueryBuilders.matchAllQuery()).setSize(0); .setQuery(QueryBuilders.matchAllQuery()).setSize(0);
assertEquals(numactions, assertEquals(numactions, searchRequestBuilder.execute().actionGet().getHits().getTotalHits().value);
searchRequestBuilder.execute().actionGet().getHits().getTotalHits());
client.close(); client.close();
} }
} }
@ -189,11 +217,18 @@ class ClientTest {
logger.error("error", client.getBulkController().getLastBulkError()); logger.error("error", client.getBulkController().getLastBulkError());
} }
assertNull(client.getBulkController().getLastBulkError()); assertNull(client.getBulkController().getLastBulkError());
assertEquals(maxthreads * actions, client.getBulkMetric().getSucceeded().getCount());
logger.log(Level.INFO, "refreshing index test");
client.refreshIndex("test"); client.refreshIndex("test");
SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client.getClient(), SearchAction.INSTANCE) SearchSourceBuilder builder = new SearchSourceBuilder()
.setQuery(QueryBuilders.matchAllQuery()).setSize(0); .query(QueryBuilders.matchAllQuery())
assertEquals(maxthreads * actions, .size(0)
searchRequestBuilder.execute().actionGet().getHits().getTotalHits()); .trackTotalHits(true);
SearchRequest searchRequest = new SearchRequest()
.indices("test")
.source(builder);
SearchResponse searchResponse = client.getClient().execute(SearchAction.INSTANCE, searchRequest).actionGet();
assertEquals(maxthreads * actions, searchResponse.getHits().getTotalHits().value);
client.close(); client.close();
} }
} }

20
elx-node/src/test/java/org/xbib/elx/node/test/DuplicateIDTest.java
View file

@ -4,6 +4,7 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.search.SearchAction; import org.elasticsearch.action.search.SearchAction;
import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.NoNodeAvailableException; import org.elasticsearch.client.transport.NoNodeAvailableException;
import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder; import org.elasticsearch.search.builder.SearchSourceBuilder;
@ -27,7 +28,7 @@ class DuplicateIDTest {
private static final Long MAX_ACTIONS_PER_REQUEST = 10L; private static final Long MAX_ACTIONS_PER_REQUEST = 10L;
private static final Long ACTIONS = 50L; private static final Long ACTIONS = 100L;
private final TestExtension.Helper helper; private final TestExtension.Helper helper;
@ -51,13 +52,16 @@ class DuplicateIDTest {
client.flush(); client.flush();
client.waitForResponses(30L, TimeUnit.SECONDS); client.waitForResponses(30L, TimeUnit.SECONDS);
client.refreshIndex("test"); client.refreshIndex("test");
SearchSourceBuilder builder = new SearchSourceBuilder(); SearchSourceBuilder builder = new SearchSourceBuilder()
builder.query(QueryBuilders.matchAllQuery()); .query(QueryBuilders.matchAllQuery())
SearchRequest searchRequest = new SearchRequest(); .size(0)
searchRequest.indices("test"); .trackTotalHits(true);
searchRequest.types("test"); SearchRequest searchRequest = new SearchRequest()
searchRequest.source(builder); .indices("test")
long hits = helper.client("1").execute(SearchAction.INSTANCE, searchRequest).actionGet().getHits().getTotalHits(); .source(builder);
SearchResponse searchResponse =
helper.client("1").execute(SearchAction.INSTANCE, searchRequest).actionGet();
long hits = searchResponse.getHits().getTotalHits().value;
logger.info("hits = {}", hits); logger.info("hits = {}", hits);
assertTrue(hits < ACTIONS); assertTrue(hits < ACTIONS);
} catch (NoNodeAvailableException e) { } catch (NoNodeAvailableException e) {

4
elx-node/src/test/java/org/xbib/elx/node/test/MockNode.java
View file

@ -6,10 +6,12 @@ import org.elasticsearch.node.Node;
import org.elasticsearch.plugins.Plugin; import org.elasticsearch.plugins.Plugin;
import java.util.List; import java.util.List;
import java.util.Map;
public class MockNode extends Node { public class MockNode extends Node {
public MockNode(Settings settings, List<Class<? extends Plugin>> classpathPlugins) { public MockNode(Settings settings, List<Class<? extends Plugin>> classpathPlugins) {
super(InternalSettingsPreparer.prepareEnvironment(settings, null), classpathPlugins); super(InternalSettingsPreparer.prepareEnvironment(settings, Map.of(), null, () -> "mock"),
classpathPlugins, false);
} }
} }

11
elx-node/src/test/java/org/xbib/elx/node/test/TestExtension.java
View file

@ -12,7 +12,6 @@ import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
import org.elasticsearch.action.admin.cluster.state.ClusterStateAction; import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest; import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse; import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.analysis.common.CommonAnalysisPlugin;
import org.elasticsearch.client.ElasticsearchClient; import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.client.support.AbstractClient; import org.elasticsearch.client.support.AbstractClient;
import org.elasticsearch.cluster.health.ClusterHealthStatus; import org.elasticsearch.cluster.health.ClusterHealthStatus;
@ -22,7 +21,7 @@ import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.node.Node; import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeValidationException; import org.elasticsearch.node.NodeValidationException;
import org.elasticsearch.plugins.Plugin; import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.transport.netty4.Netty4Plugin; import org.elasticsearch.transport.Netty4Plugin;
import org.junit.jupiter.api.extension.AfterEachCallback; import org.junit.jupiter.api.extension.AfterEachCallback;
import org.junit.jupiter.api.extension.BeforeEachCallback; import org.junit.jupiter.api.extension.BeforeEachCallback;
import org.junit.jupiter.api.extension.ExtensionContext; import org.junit.jupiter.api.extension.ExtensionContext;
@ -37,7 +36,7 @@ import java.nio.file.Path;
import java.nio.file.Paths; import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor; import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes; import java.nio.file.attribute.BasicFileAttributes;
import java.util.Arrays; import java.util.Collections;
import java.util.HashMap; import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
@ -192,8 +191,8 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
return Settings.builder() return Settings.builder()
.put("cluster.name", getClusterName()) .put("cluster.name", getClusterName())
.put("path.home", getHome()) .put("path.home", getHome())
.put("discovery.zen.master_election.ignore_non_master_pings", "true") .put("cluster.initial_master_nodes", "1")
.put("transport.netty.epoll", "false") .put("discovery.seed_hosts", "127.0.0.1:9300")
.build(); .build();
} }
@ -219,7 +218,7 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
.put(getNodeSettings()) .put(getNodeSettings())
.put("node.name", id) .put("node.name", id)
.build(); .build();
List<Class<? extends Plugin>> plugins = Arrays.asList(CommonAnalysisPlugin.class, Netty4Plugin.class); List<Class<? extends Plugin>> plugins = Collections.singletonList(Netty4Plugin.class);
Node node = new MockNode(nodeSettings, plugins); Node node = new MockNode(nodeSettings, plugins);
AbstractClient client = (AbstractClient) node.client(); AbstractClient client = (AbstractClient) node.client();
nodes.put(id, node); nodes.put(id, node);

5
elx-transport/build.gradle
View file

@ -1,5 +1,6 @@
dependencies { dependencies {
compile project(':elx-common') compile project(':elx-common')
compile "org.xbib.elasticsearch:transport-netty4:${rootProject.property('elasticsearch-server.version')}" compile "org.elasticsearch.plugin:transport-netty4-client:${rootProject.property('elasticsearch.version')}"
testCompile "org.xbib.elasticsearch:elasticsearch-analysis-common:${rootProject.property('elasticsearch-server.version')}" compile "io.netty:netty-codec-http:${project.property('netty.version')}"
compile "io.netty:netty-transport:${project.property('netty.version')}"
} }

2
elx-transport/src/main/java/org/xbib/elx/transport/ExtendedTransportClient.java
View file

@ -22,7 +22,7 @@ import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder; import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory; import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.plugins.Plugin; import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.transport.netty4.Netty4Plugin; import org.elasticsearch.transport.Netty4Plugin;
import org.xbib.elx.common.AbstractExtendedClient; import org.xbib.elx.common.AbstractExtendedClient;
import org.xbib.elx.common.util.NetworkUtils; import org.xbib.elx.common.util.NetworkUtils;

51
elx-transport/src/test/java/org/xbib/elx/transport/test/ClientTest.java
View file

@ -6,11 +6,13 @@ import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsAction; import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsAction;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest; import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse; import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsAction;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsRequest;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse;
import org.elasticsearch.action.search.SearchAction; import org.elasticsearch.action.search.SearchAction;
import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.NoNodeAvailableException; import org.elasticsearch.client.transport.NoNodeAvailableException;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder; import org.elasticsearch.common.xcontent.XContentBuilder;
@ -85,28 +87,45 @@ class ClientTest {
} }
@Test @Test
void testMapping() throws Exception { void testNewIndexWithSettings() throws Exception {
final ExtendedTransportClient client = ClientBuilder.builder() final ExtendedTransportClient client = ClientBuilder.builder()
.provider(ExtendedTransportClientProvider.class) .provider(ExtendedTransportClientProvider.class)
.put(helper.getTransportSettings()) .put(helper.getTransportSettings())
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(5)) .put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(5))
.build(); .build();
Settings settings = Settings.builder().put("index.number_of_shards", "1").build();
client.newIndex("test", settings);
GetSettingsRequest getSettingsRequest = new GetSettingsRequest()
.indices("test");
GetSettingsResponse getSettingsResponse =
client.getClient().execute(GetSettingsAction.INSTANCE, getSettingsRequest).actionGet();
logger.log(Level.INFO, "settings=" + getSettingsResponse.getSetting("test", "index.number_of_shards"));
assertEquals("1", getSettingsResponse.getSetting("test", "index.number_of_shards"));
client.close();
}
@Test
void testNewIndexWithSettingsAndMapping() throws Exception {
final ExtendedTransportClient client = ClientBuilder.builder()
.provider(ExtendedTransportClientProvider.class)
.put(helper.getTransportSettings())
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(5))
.build();
Settings settings = Settings.builder().put("index.number_of_shards", "1").build();
XContentBuilder builder = JsonXContent.contentBuilder() XContentBuilder builder = JsonXContent.contentBuilder()
.startObject() .startObject()
.startObject("doc")
.startObject("properties") .startObject("properties")
.startObject("location") .startObject("location")
.field("type", "geo_point") .field("type", "geo_point")
.endObject() .endObject()
.endObject() .endObject()
.endObject()
.endObject(); .endObject();
client.newIndex("test", Settings.EMPTY, Strings.toString(builder)); client.newIndex("test", settings, builder);
GetMappingsRequest getMappingsRequest = new GetMappingsRequest().indices("test"); GetMappingsRequest getMappingsRequest = new GetMappingsRequest().indices("test");
GetMappingsResponse getMappingsResponse = GetMappingsResponse getMappingsResponse =
client.getClient().execute(GetMappingsAction.INSTANCE, getMappingsRequest).actionGet(); client.getClient().execute(GetMappingsAction.INSTANCE, getMappingsRequest).actionGet();
logger.info("mappings={}", getMappingsResponse.getMappings()); logger.info("mappings={}", getMappingsResponse.getMappings());
assertTrue(getMappingsResponse.getMappings().get("test").containsKey("doc")); assertTrue(getMappingsResponse.getMappings().get("test").containsKey("_doc"));
client.close(); client.close();
} }
@ -146,7 +165,7 @@ class ClientTest {
searchRequest.source(builder); searchRequest.source(builder);
SearchResponse searchResponse = client.getClient().execute(SearchAction.INSTANCE, searchRequest).actionGet(); SearchResponse searchResponse = client.getClient().execute(SearchAction.INSTANCE, searchRequest).actionGet();
logger.log(Level.INFO, searchResponse.toString()); logger.log(Level.INFO, searchResponse.toString());
assertEquals(numactions, searchResponse.getHits().getTotalHits()); assertEquals(numactions, searchResponse.getHits().getTotalHits().value);
client.close(); client.close();
} }
} }
@ -207,16 +226,18 @@ class ClientTest {
logger.error("error", client.getBulkController().getLastBulkError()); logger.error("error", client.getBulkController().getLastBulkError());
} }
assertNull(client.getBulkController().getLastBulkError()); assertNull(client.getBulkController().getLastBulkError());
assertEquals(maxthreads * actions, client.getBulkMetric().getSucceeded().getCount());
logger.log(Level.INFO, "refreshing index test");
client.refreshIndex("test"); client.refreshIndex("test");
SearchSourceBuilder builder = new SearchSourceBuilder(); SearchSourceBuilder builder = new SearchSourceBuilder()
builder.query(QueryBuilders.matchAllQuery()); .query(QueryBuilders.matchAllQuery())
builder.size(0); .size(0)
SearchRequest searchRequest = new SearchRequest(); .trackTotalHits(true);
searchRequest.indices("test"); SearchRequest searchRequest = new SearchRequest()
searchRequest.source(builder); .indices("test")
.source(builder);
SearchResponse searchResponse = client.getClient().execute(SearchAction.INSTANCE, searchRequest).actionGet(); SearchResponse searchResponse = client.getClient().execute(SearchAction.INSTANCE, searchRequest).actionGet();
logger.log(Level.INFO, searchResponse.toString()); assertEquals(maxthreads * actions, searchResponse.getHits().getTotalHits().value);
assertEquals(maxthreads * actions, searchResponse.getHits().getTotalHits());
client.close(); client.close();
} }
} }

18
elx-transport/src/test/java/org/xbib/elx/transport/test/DuplicateIDTest.java
View file

@ -28,7 +28,7 @@ class DuplicateIDTest {
private static final Long MAX_ACTIONS_PER_REQUEST = 10L; private static final Long MAX_ACTIONS_PER_REQUEST = 10L;
private static final Long ACTIONS = 5L; private static final Long ACTIONS = 100L;
private final TestExtension.Helper helper; private final TestExtension.Helper helper;
@ -53,16 +53,16 @@ class DuplicateIDTest {
client.flush(); client.flush();
client.waitForResponses(30L, TimeUnit.SECONDS); client.waitForResponses(30L, TimeUnit.SECONDS);
client.refreshIndex("test_dup"); client.refreshIndex("test_dup");
SearchSourceBuilder builder = new SearchSourceBuilder(); SearchSourceBuilder builder = new SearchSourceBuilder()
builder.query(QueryBuilders.matchAllQuery()); .query(QueryBuilders.matchAllQuery())
builder.size(0); .size(0)
SearchRequest searchRequest = new SearchRequest(); .trackTotalHits(true);
searchRequest.indices("test_dup"); SearchRequest searchRequest = new SearchRequest()
searchRequest.types("test_dup"); .indices("test_dup")
searchRequest.source(builder); .source(builder);
SearchResponse searchResponse = SearchResponse searchResponse =
helper.client("1").execute(SearchAction.INSTANCE, searchRequest).actionGet(); helper.client("1").execute(SearchAction.INSTANCE, searchRequest).actionGet();
long hits = searchResponse.getHits().getTotalHits(); long hits = searchResponse.getHits().getTotalHits().value;
logger.info("hits = {}", hits); logger.info("hits = {}", hits);
assertTrue(hits < ACTIONS); assertTrue(hits < ACTIONS);
} catch (NoNodeAvailableException e) { } catch (NoNodeAvailableException e) {

4
elx-transport/src/test/java/org/xbib/elx/transport/test/MockNode.java
View file

@ -6,10 +6,12 @@ import org.elasticsearch.node.Node;
import org.elasticsearch.plugins.Plugin; import org.elasticsearch.plugins.Plugin;
import java.util.List; import java.util.List;
import java.util.Map;
public class MockNode extends Node { public class MockNode extends Node {
public MockNode(Settings settings, List<Class<? extends Plugin>> classpathPlugins) { public MockNode(Settings settings, List<Class<? extends Plugin>> classpathPlugins) {
super(InternalSettingsPreparer.prepareEnvironment(settings, null), classpathPlugins); super(InternalSettingsPreparer.prepareEnvironment(settings, Map.of(), null, () -> "mock"),
classpathPlugins, false);
} }
} }

10
elx-transport/src/test/java/org/xbib/elx/transport/test/TestExtension.java
View file

@ -12,7 +12,6 @@ import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
import org.elasticsearch.action.admin.cluster.state.ClusterStateAction; import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest; import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse; import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.analysis.common.CommonAnalysisPlugin;
import org.elasticsearch.client.ElasticsearchClient; import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.client.support.AbstractClient; import org.elasticsearch.client.support.AbstractClient;
import org.elasticsearch.cluster.health.ClusterHealthStatus; import org.elasticsearch.cluster.health.ClusterHealthStatus;
@ -22,7 +21,7 @@ import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.node.Node; import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeValidationException; import org.elasticsearch.node.NodeValidationException;
import org.elasticsearch.plugins.Plugin; import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.transport.netty4.Netty4Plugin; import org.elasticsearch.transport.Netty4Plugin;
import org.junit.jupiter.api.extension.AfterEachCallback; import org.junit.jupiter.api.extension.AfterEachCallback;
import org.junit.jupiter.api.extension.BeforeEachCallback; import org.junit.jupiter.api.extension.BeforeEachCallback;
import org.junit.jupiter.api.extension.ExtensionContext; import org.junit.jupiter.api.extension.ExtensionContext;
@ -37,7 +36,7 @@ import java.nio.file.Path;
import java.nio.file.Paths; import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor; import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes; import java.nio.file.attribute.BasicFileAttributes;
import java.util.Arrays; import java.util.Collections;
import java.util.HashMap; import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
@ -192,7 +191,8 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
return Settings.builder() return Settings.builder()
.put("cluster.name", getClusterName()) .put("cluster.name", getClusterName())
.put("path.home", getHome()) .put("path.home", getHome())
.put("discovery.zen.master_election.ignore_non_master_pings", "true") .put("cluster.initial_master_nodes", "1")
.put("discovery.seed_hosts", "127.0.0.1:9300")
.build(); .build();
} }
@ -227,7 +227,7 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
.put(getNodeSettings()) .put(getNodeSettings())
.put("node.name", id) .put("node.name", id)
.build(); .build();
List<Class<? extends Plugin>> plugins = Arrays.asList(CommonAnalysisPlugin.class, Netty4Plugin.class); List<Class<? extends Plugin>> plugins = Collections.singletonList(Netty4Plugin.class);
Node node = new MockNode(nodeSettings, plugins); Node node = new MockNode(nodeSettings, plugins);
AbstractClient client = (AbstractClient) node.client(); AbstractClient client = (AbstractClient) node.client();
nodes.put(id, node); nodes.put(id, node);

15
gradle.properties
View file

@ -1,19 +1,16 @@
group = org.xbib group = org.xbib
name = elx name = elx
version = 6.3.2.5 version = 7.6.1.0
profile = default
release = 0
elasticsearch-server.version = 6.3.2.4 elasticsearch.version = 7.6.1
log4j.version = 2.12.1 netty.version = 4.1.49.Final
xbib-netty-http.version = 4.1.49.0
tcnative.version = 2.0.29.Final tcnative.version = 2.0.29.Final
tcnative-legacy-macosx.version = 2.0.26.Final tcnative-legacy-macosx.version = 2.0.26.Final
bouncycastle.version = 1.64 bouncycastle.version = 1.64
xbib-metrics.version = 2.0.0 xbib-metrics.version = 2.0.0
xbib-netty-http.version = 4.1.49.0
# test # test
junit.version = 5.4.2 log4j.version = 2.13.1
junit.version = 5.6.2
asciidoclet.version = 1.6.0.0 asciidoclet.version = 1.6.0.0
org.gradle.warning.mode = all