some named modules introduced, API update to conform with 7.6 branch

This commit is contained in:
Jörg Prante 2020-06-09 17:18:08 +02:00
parent cfff32556f
commit 3c7d1e61b1
91 changed files with 1001 additions and 1275 deletions

View file

@ -1,7 +1,6 @@
plugins {
id "de.marcphilipp.nexus-publish" version "0.4.0"
id "io.codearte.nexus-staging" version "0.21.0"
id "org.xbib.gradle.plugin.asciidoctor" version "1.6.0.1"
id "io.codearte.nexus-staging" version "0.21.1"
}
wrapper {

View file

@ -10,7 +10,7 @@ import java.util.concurrent.TimeUnit;
/**
* Interface for extended managing and indexing methods of an Elasticsearch client.
*/
public interface AdminClient extends NativeClient {
public interface AdminClient extends BasicClient {
/**
* Build index definition from settings.
@ -22,9 +22,11 @@ public interface AdminClient extends NativeClient {
*/
IndexDefinition buildIndexDefinitionFromSettings(String index, Settings settings) throws IOException;
Map<String, ?> getMapping(String index);
Map<String, ?> getMapping(String index) throws IOException;
Map<String, ?> getMapping(String index, String type);
Map<String, ?> getMapping(String index, String type) throws IOException;
void checkMapping(String index);
/**
* Delete an index.

View file

@ -6,7 +6,7 @@ import java.io.Closeable;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
public interface NativeClient extends Closeable {
public interface BasicClient extends Closeable {
/**
* Set an Elasticsearch client to extend from it. May be null for TransportClient.

View file

@ -11,13 +11,7 @@ import java.io.IOException;
import java.util.Map;
import java.util.concurrent.TimeUnit;
public interface BulkClient extends NativeClient, Flushable {
/**
* Get bulk metric.
* @return the bulk metric
*/
BulkMetric getBulkMetric();
public interface BulkClient extends BasicClient, Flushable {
/**
* Get buulk control.
@ -68,7 +62,7 @@ public interface BulkClient extends NativeClient, Flushable {
* @param mapping mapping
* @throws IOException if settings/mapping is invalid or index creation fails
*/
void newIndex(String index, Settings settings, Map<String, ?> mapping) throws IOException;
void newIndex(String index, Settings settings, Map<String, Object> mapping) throws IOException;
/**
* Add index request. Each request will be added to a queue for bulking requests.

View file

@ -14,6 +14,10 @@ public interface BulkController extends Closeable, Flushable {
void init(Settings settings);
void inactivate();
BulkMetric getBulkMetric();
Throwable getLastBulkError();
void startBulkMode(IndexDefinition indexDefinition) throws IOException;
@ -21,13 +25,13 @@ public interface BulkController extends Closeable, Flushable {
void startBulkMode(String indexName, long startRefreshIntervalInSeconds,
long stopRefreshIntervalInSeconds) throws IOException;
void index(IndexRequest indexRequest);
void bulkIndex(IndexRequest indexRequest);
void delete(DeleteRequest deleteRequest);
void bulkDelete(DeleteRequest deleteRequest);
void update(UpdateRequest updateRequest);
void bulkUpdate(UpdateRequest updateRequest);
boolean waitForResponses(long timeout, TimeUnit timeUnit);
boolean waitForBulkResponses(long timeout, TimeUnit timeUnit);
void stopBulkMode(IndexDefinition indexDefinition) throws IOException;

View file

@ -0,0 +1,43 @@
package org.xbib.elx.api;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
public interface BulkListener {
/**
* Callback before the bulk is executed.
*
* @param executionId execution ID
* @param request request
*/
void beforeBulk(long executionId, BulkRequest request);
/**
* Callback after a successful execution of bulk request.
*
* @param executionId execution ID
* @param request request
* @param response response
*/
void afterBulk(long executionId, BulkRequest request, BulkResponse response);
/**
* Callback after a failed execution of bulk request.
*
* Note that in case an instance of <code>InterruptedException</code> is passed, which means that request
* processing has been
* cancelled externally, the thread's interruption status has been restored prior to calling this method.
*
* @param executionId execution ID
* @param request request
* @param failure failure
*/
void afterBulk(long executionId, BulkRequest request, Throwable failure);
/**
* Get the last bulk error
* @return the lst bulk error
*/
Throwable getLastBulkError();
}

View file

@ -10,6 +10,8 @@ public interface BulkMetric extends Closeable {
void init(Settings settings);
void markTotalIngest(long n);
Metered getTotalIngest();
Count getTotalIngestSizeInBytes();

View file

@ -1,9 +1,6 @@
package org.xbib.elx.api;
import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import java.io.Closeable;
import java.io.Flushable;
import java.util.concurrent.TimeUnit;
@ -12,53 +9,9 @@ public interface BulkProcessor extends Closeable, Flushable {
BulkProcessor add(ActionRequest request);
BulkProcessor add(ActionRequest request, Object payload);
boolean awaitFlush(long timeout, TimeUnit unit) throws InterruptedException;
boolean awaitClose(long timeout, TimeUnit unit) throws InterruptedException;
interface BulkRequestHandler {
void execute(BulkRequest bulkRequest, long executionId);
boolean close(long timeout, TimeUnit unit) throws InterruptedException;
}
/**
* A listener for the execution.
*/
interface Listener {
/**
* Callback before the bulk is executed.
*
* @param executionId execution ID
* @param request request
*/
void beforeBulk(long executionId, BulkRequest request);
/**
* Callback after a successful execution of bulk request.
*
* @param executionId execution ID
* @param request request
* @param response response
*/
void afterBulk(long executionId, BulkRequest request, BulkResponse response);
/**
* Callback after a failed execution of bulk request.
*
* Note that in case an instance of <code>InterruptedException</code> is passed, which means that request
* processing has been
* cancelled externally, the thread's interruption status has been restored prior to calling this method.
*
* @param executionId execution ID
* @param request request
* @param failure failure
*/
void afterBulk(long executionId, BulkRequest request, Throwable failure);
}
BulkListener getBulkListener();
}

View file

@ -0,0 +1,11 @@
package org.xbib.elx.api;
import org.elasticsearch.action.bulk.BulkRequest;
import java.util.concurrent.TimeUnit;
public interface BulkRequestHandler {
void execute(BulkRequest bulkRequest, long executionId);
boolean close(long timeout, TimeUnit unit) throws InterruptedException;
}

View file

@ -12,7 +12,9 @@ import java.util.Optional;
import java.util.function.Consumer;
import java.util.stream.Stream;
public interface SearchClient extends NativeClient {
public interface SearchClient extends BasicClient {
SearchMetric getSearchMetric();
Optional<GetResponse> get(Consumer<GetRequestBuilder> getRequestBuilder);

View file

@ -0,0 +1,29 @@
package org.xbib.elx.api;
import org.elasticsearch.common.settings.Settings;
import org.xbib.metrics.api.Count;
import org.xbib.metrics.api.Metered;
import java.io.Closeable;
public interface SearchMetric extends Closeable {
void init(Settings settings);
void markTotalQueries(long n);
Metered getTotalQueries();
Count getCurrentQueries();
Count getQueries();
Count getSucceededQueries();
Count getEmptyQueries();
long elapsed();
void start();
void stop();
}

View file

@ -1,4 +1,7 @@
module org.xbib.elx.common {
uses org.xbib.elx.api.AdminClientProvider;
uses org.xbib.elx.api.BulkClientProvider;
uses org.xbib.elx.api.SearchClientProvider;
exports org.xbib.elx.common;
exports org.xbib.elx.common.io;
exports org.xbib.elx.common.util;

View file

@ -28,8 +28,11 @@ 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.admin.indices.settings.put.UpdateSettingsAction;
import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest;
import org.elasticsearch.action.search.SearchAction;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.cluster.metadata.AliasMetaData;
import org.elasticsearch.cluster.metadata.AliasOrIndex;
@ -62,7 +65,6 @@ import java.io.IOException;
import java.io.InputStream;
import java.net.MalformedURLException;
import java.net.URL;
import java.nio.charset.MalformedInputException;
import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.time.ZoneId;
@ -87,10 +89,16 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
public abstract class AbstractAdminClient extends AbstractNativeClient implements AdminClient {
public abstract class AbstractAdminClient extends AbstractBasicClient implements AdminClient {
private static final Logger logger = LogManager.getLogger(AbstractAdminClient.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";
private static final IndexShiftResult EMPTY_INDEX_SHIFT_RESULT = new IndexShiftResult() {
@Override
public List<String> getMovedAliases() {
@ -147,13 +155,12 @@ public abstract class AbstractAdminClient extends AbstractNativeClient implement
@Override
public AdminClient deleteIndex(String index) {
ensureClientIsPresent();
if (index == null) {
logger.warn("no index name given to delete index");
return this;
}
ensureClientIsPresent();
DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest()
.indices(index);
DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest().indices(index);
client.execute(DeleteIndexAction.INSTANCE, deleteIndexRequest).actionGet();
waitForCluster("YELLOW", 30L, TimeUnit.SECONDS);
waitForShards(30L, TimeUnit.SECONDS);
@ -196,7 +203,6 @@ public abstract class AbstractAdminClient extends AbstractNativeClient implement
return replica;
}
@Override
public String resolveMostRecentIndex(String alias) {
ensureClientIsPresent();
@ -295,7 +301,7 @@ public abstract class AbstractAdminClient extends AbstractNativeClient implement
String alias = entry.getKey();
String filter = entry.getValue();
indicesAliasesRequest.addAliasAction(IndicesAliasesRequest.AliasActions.remove()
.indices(oldIndex).alias(alias));
.index(oldIndex).alias(alias));
if (filter != null) {
indicesAliasesRequest.addAliasAction(IndicesAliasesRequest.AliasActions.add()
.index(fullIndexName).alias(alias).filter(filter));
@ -321,7 +327,7 @@ public abstract class AbstractAdminClient extends AbstractNativeClient implement
} else {
String filter = oldAliasMap.get(additionalAlias);
indicesAliasesRequest.addAliasAction(IndicesAliasesRequest.AliasActions.remove()
.indices(oldIndex).alias(additionalAlias));
.index(oldIndex).alias(additionalAlias));
if (filter != null) {
indicesAliasesRequest.addAliasAction(IndicesAliasesRequest.AliasActions.add()
.index(fullIndexName).alias(additionalAlias).filter(filter));
@ -337,8 +343,8 @@ public abstract class AbstractAdminClient extends AbstractNativeClient implement
logger.debug("indices alias request = {}", indicesAliasesRequest.getAliasActions().toString());
IndicesAliasesResponse indicesAliasesResponse =
client.execute(IndicesAliasesAction.INSTANCE, indicesAliasesRequest).actionGet();
logger.debug("response isAcknowledged = {} isFragment = {}",
indicesAliasesResponse.isAcknowledged(), indicesAliasesResponse.isFragment());
logger.debug("response isAcknowledged = {}",
indicesAliasesResponse.isAcknowledged());
}
return new SuccessIndexShiftResult(moveAliases, newAliases);
}
@ -358,12 +364,11 @@ public abstract class AbstractAdminClient extends AbstractNativeClient implement
return EMPTY_INDEX_PRUNE_RESULT;
}
ensureClientIsPresent();
GetIndexRequestBuilder getIndexRequestBuilder =
new GetIndexRequestBuilder(client, GetIndexAction.INSTANCE);
GetIndexRequestBuilder getIndexRequestBuilder = new GetIndexRequestBuilder(client, GetIndexAction.INSTANCE);
GetIndexResponse getIndexResponse = getIndexRequestBuilder.execute().actionGet();
Pattern pattern = Pattern.compile("^(.*?)(\\d+)$");
logger.info("found {} indices", getIndexResponse.getIndices().length);
Set<String> candidateIndices = new TreeSet<>();
List<String> candidateIndices = new ArrayList<>();
for (String s : getIndexResponse.getIndices()) {
Matcher m = pattern.matcher(s);
if (m.matches() && index.equals(m.group(1)) && !s.equals(fullIndexName)) {
@ -456,7 +461,7 @@ public abstract class AbstractAdminClient extends AbstractNativeClient implement
@Override
public IndexDefinition buildIndexDefinitionFromSettings(String index, Settings settings)
throws IOException {
boolean isEnabled = settings.getAsBoolean("enabled", !(client instanceof MockAdminClient));
boolean isEnabled = settings.getAsBoolean("enabled", false);
String indexName = settings.get("name", index);
String fullIndexName;
String dateTimePattern = settings.get("dateTimePattern");
@ -488,6 +493,35 @@ public abstract class AbstractAdminClient extends AbstractNativeClient implement
.setStopRefreshInterval(settings.getAsLong("bulk.stoprefreshinterval", -1L));
}
@Override
public void updateIndexSetting(String index, String key, Object value, long timeout, TimeUnit timeUnit) throws IOException {
ensureClientIsPresent();
if (index == null) {
throw new IOException("no index name given");
}
Settings.Builder updateSettingsBuilder = Settings.builder();
updateSettingsBuilder.put(key, value.toString());
UpdateSettingsRequest updateSettingsRequest = new UpdateSettingsRequest(index)
.settings(updateSettingsBuilder).timeout(toTimeValue(timeout, timeUnit));
client.execute(UpdateSettingsAction.INSTANCE, updateSettingsRequest).actionGet();
}
@Override
public void checkMapping(String index) {
ensureClientIsPresent();
GetMappingsRequest getMappingsRequest = new GetMappingsRequest().indices(index);
GetMappingsResponse getMappingsResponse = client.execute(GetMappingsAction.INSTANCE, getMappingsRequest).actionGet();
ImmutableOpenMap<String, ImmutableOpenMap<String, MappingMetaData>> map = getMappingsResponse.getMappings();
map.keys().forEach((Consumer<ObjectCursor<String>>) stringObjectCursor -> {
ImmutableOpenMap<String, MappingMetaData> mappings = map.get(stringObjectCursor.value);
for (ObjectObjectCursor<String, MappingMetaData> cursor : mappings) {
String mappingName = cursor.key;
MappingMetaData mappingMetaData = cursor.value;
checkMapping(index, mappingName, mappingMetaData);
}
});
}
private static String findSettingsFrom(String string) throws IOException {
if (string == null) {
return null;
@ -514,7 +548,7 @@ public abstract class AbstractAdminClient extends AbstractNativeClient implement
try (InputStream inputStream = url.openStream()) {
if (string.endsWith(".json")) {
Map<String, ?> mappings = JsonXContent.jsonXContent.createParser(NamedXContentRegistry.EMPTY,
DeprecationHandler.THROW_UNSUPPORTED_OPERATION, inputStream).mapOrdered();
DeprecationHandler.THROW_UNSUPPORTED_OPERATION, inputStream).mapOrdered();
XContentBuilder builder = JsonXContent.contentBuilder();
builder.startObject().map(mappings).endObject();
return Strings.toString(builder);
@ -528,7 +562,7 @@ public abstract class AbstractAdminClient extends AbstractNativeClient implement
}
}
return string;
} catch (MalformedInputException e) {
} catch (MalformedURLException e) {
return string;
}
}
@ -549,37 +583,22 @@ public abstract class AbstractAdminClient extends AbstractNativeClient implement
return result;
}
public void checkMapping(String index) {
ensureClientIsPresent();
GetMappingsRequest getMappingsRequest = new GetMappingsRequest().indices(index);
GetMappingsResponse getMappingsResponse = client.execute(GetMappingsAction.INSTANCE, getMappingsRequest).actionGet();
ImmutableOpenMap<String, ImmutableOpenMap<String, MappingMetaData>> map = getMappingsResponse.getMappings();
map.keys().forEach((Consumer<ObjectCursor<String>>) stringObjectCursor -> {
ImmutableOpenMap<String, MappingMetaData> mappings = map.get(stringObjectCursor.value);
for (ObjectObjectCursor<String, MappingMetaData> cursor : mappings) {
String mappingName = cursor.key;
MappingMetaData mappingMetaData = cursor.value;
checkMapping(index, mappingName, mappingMetaData);
}
});
}
private void checkMapping(String index, String type, MappingMetaData mappingMetaData) {
try {
SearchSourceBuilder builder = new SearchSourceBuilder();
builder.query(QueryBuilders.matchAllQuery());
builder.size(0);
SearchRequest searchRequest = new SearchRequest();
searchRequest.indices(index);
searchRequest.types(type);
searchRequest.source(builder);
SearchResponse searchResponse =
client.execute(SearchAction.INSTANCE, searchRequest).actionGet();
SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client, SearchAction.INSTANCE)
.setIndices(index)
.setQuery(QueryBuilders.matchAllQuery())
.setSize(0)
.setTrackTotalHits(true);
SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
long total = searchResponse.getHits().getTotalHits();
if (total > 0L) {
Map<String, Long> fields = new TreeMap<>();
Map<String, Object> root = mappingMetaData.getSourceAsMap();
checkMapping(index, type, "", "", root, fields);
checkMapping(index, "", "", root, fields);
AtomicInteger empty = new AtomicInteger();
Map<String, Long> map = sortByValue(fields);
map.forEach((key, value) -> {
@ -600,7 +619,7 @@ public abstract class AbstractAdminClient extends AbstractNativeClient implement
}
@SuppressWarnings("unchecked")
private void checkMapping(String index, String type,
private void checkMapping(String index,
String pathDef, String fieldName, Map<String, Object> map,
Map<String, Long> fields) {
String path = pathDef;
@ -625,7 +644,7 @@ public abstract class AbstractAdminClient extends AbstractNativeClient implement
String fieldType = o instanceof String ? o.toString() : null;
// do not recurse into our custom field mapper
if (!"standardnumber".equals(fieldType) && !"ref".equals(fieldType)) {
checkMapping(index, type, path, key, child, fields);
checkMapping(index, path, key, child, fields);
}
} else if ("type".equals(key)) {
QueryBuilder filterBuilder = QueryBuilders.existsQuery(path);
@ -633,12 +652,12 @@ public abstract class AbstractAdminClient extends AbstractNativeClient implement
SearchSourceBuilder builder = new SearchSourceBuilder();
builder.query(queryBuilder);
builder.size(0);
SearchRequest searchRequest = new SearchRequest();
searchRequest.indices(index);
searchRequest.types(type);
searchRequest.source(builder);
SearchResponse searchResponse =
client.execute(SearchAction.INSTANCE, searchRequest).actionGet();
SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client, SearchAction.INSTANCE)
.setIndices(index)
.setQuery(queryBuilder)
.setSize(0)
.setTrackTotalHits(true);
SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
fields.put(path, searchResponse.getHits().getTotalHits());
}
}

View file

@ -23,31 +23,28 @@ import org.elasticsearch.cluster.health.ClusterHealthStatus;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.xbib.elx.api.NativeClient;
import org.xbib.elx.api.BasicClient;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
public abstract class AbstractNativeClient implements NativeClient {
public abstract class AbstractBasicClient implements BasicClient {
private static final Logger logger = LogManager.getLogger(AbstractNativeClient.class.getName());
/**
* The one and only index type name used in the extended client.
* NOTE: all Elasticsearch version less than 6.2.0 forbid a prepending "_".
*/
protected static final String TYPE_NAME = "doc";
private static final Logger logger = LogManager.getLogger(AbstractBasicClient.class.getName());
protected ElasticsearchClient client;
protected Settings settings;
protected final AtomicBoolean closed;
protected AbstractNativeClient() {
protected AbstractBasicClient() {
closed = new AtomicBoolean(false);
}
@Override
public void setClient(ElasticsearchClient client) {
logger.log(Level.INFO, "setting client = " + client);
this.client = client;
}
@ -56,15 +53,12 @@ public abstract class AbstractNativeClient implements NativeClient {
return client;
}
protected abstract ElasticsearchClient createClient(Settings settings) throws IOException;
protected abstract void closeClient() throws IOException;
@Override
public void init(Settings settings) throws IOException {
if (client == null) {
if (closed.compareAndSet(false, true)) {
logger.log(Level.INFO, "initializing with settings = " + settings.toDelimitedString(','));
client = createClient(settings);
this.settings = settings;
setClient(createClient(settings));
}
}
@ -91,7 +85,6 @@ public abstract class AbstractNativeClient implements NativeClient {
@Override
public void waitForCluster(String statusString, long maxWaitTime, TimeUnit timeUnit) {
ensureClientIsPresent();
logger.info("waiting for cluster status " + statusString);
ClusterHealthStatus status = ClusterHealthStatus.fromString(statusString);
TimeValue timeout = toTimeValue(maxWaitTime, timeUnit);
ClusterHealthRequest clusterHealthRequest = new ClusterHealthRequest()
@ -167,10 +160,14 @@ public abstract class AbstractNativeClient implements NativeClient {
public void close() throws IOException {
ensureClientIsPresent();
if (closed.compareAndSet(false, true)) {
closeClient();
closeClient(settings);
}
}
protected abstract ElasticsearchClient createClient(Settings settings) throws IOException;
protected abstract void closeClient(Settings settings) throws IOException;
protected void updateIndexSetting(String index, String key, Object value, long timeout, TimeUnit timeUnit) throws IOException {
ensureClientIsPresent();
if (index == null) {
@ -215,5 +212,4 @@ public abstract class AbstractNativeClient implements NativeClient {
throw new IllegalArgumentException("unknown time unit: " + timeUnit);
}
}
}

View file

@ -14,52 +14,41 @@ import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.bytes.BytesArray;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.DeprecationHandler;
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.xbib.elx.api.BulkClient;
import org.xbib.elx.api.BulkController;
import org.xbib.elx.api.BulkMetric;
import org.xbib.elx.api.IndexDefinition;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
public abstract class AbstractBulkClient extends AbstractNativeClient implements BulkClient {
public abstract class AbstractBulkClient extends AbstractBasicClient implements BulkClient {
private static final Logger logger = LogManager.getLogger(AbstractBulkClient.class.getName());
private BulkMetric bulkMetric;
private BulkController bulkController;
private final AtomicBoolean closed = new AtomicBoolean(true);
@Override
public void init(Settings settings) throws IOException {
super.init(settings);
if (bulkMetric == null) {
bulkMetric = new DefaultBulkMetric();
logger.log(Level.INFO, "initializing bulk metric with settings = " + settings.toDelimitedString(','));
bulkMetric.init(settings);
}
if (bulkController == null) {
bulkController = new DefaultBulkController(this, bulkMetric);
if (closed.compareAndSet(true, false)) {
super.init(settings);
bulkController = new DefaultBulkController(this);
logger.log(Level.INFO, "initializing bulk controller with settings = " + settings.toDelimitedString(','));
bulkController.init(settings);
}
}
@Override
public BulkMetric getBulkMetric() {
return bulkMetric;
}
@Override
public BulkController getBulkController() {
return bulkController;
@ -74,17 +63,13 @@ public abstract class AbstractBulkClient extends AbstractNativeClient implements
@Override
public void close() throws IOException {
ensureClientIsPresent();
if (closed.compareAndSet(false, true)) {
if (bulkMetric != null) {
logger.info("closing bulk metric");
bulkMetric.close();
}
ensureClientIsPresent();
if (bulkController != null) {
logger.info("closing bulk controller");
bulkController.close();
}
closeClient();
closeClient(settings);
}
}
@ -92,9 +77,9 @@ public abstract class AbstractBulkClient extends AbstractNativeClient implements
public void newIndex(IndexDefinition indexDefinition) throws IOException {
Settings settings = indexDefinition.getSettings() == null ? null :
Settings.builder().loadFromSource(indexDefinition.getSettings(), XContentType.JSON).build();
Map<String, ?> mappings = indexDefinition.getMappings() == null ? null :
Map<String, Object> mappings = indexDefinition.getMappings() == null ? null :
JsonXContent.jsonXContent.createParser(NamedXContentRegistry.EMPTY,
DeprecationHandler.THROW_UNSUPPORTED_OPERATION, indexDefinition.getMappings()).mapOrdered();
DeprecationHandler.THROW_UNSUPPORTED_OPERATION, indexDefinition.getMappings()).mapOrdered();
newIndex(indexDefinition.getFullIndexName(), settings, mappings);
}
@ -109,15 +94,17 @@ public abstract class AbstractBulkClient extends AbstractNativeClient implements
}
@Override
public void newIndex(String index, Settings settings, Map<String, ?> map) throws IOException {
newIndex(index, settings, map == null || map.isEmpty() ? null :
JsonXContent.contentBuilder().map(map));
public void newIndex(String index, Settings settings, XContentBuilder builder) throws IOException {
String mappingString = builder != null ? Strings.toString(builder) : null;
Map<String, Object> mappings = mappingString != null ? JsonXContent.jsonXContent.createParser(NamedXContentRegistry.EMPTY,
DeprecationHandler.THROW_UNSUPPORTED_OPERATION, mappingString).mapOrdered() : null;
newIndex(index, settings, mappings);
}
@Override
public void newIndex(String index, Settings settings, XContentBuilder builder) throws IOException {
public void newIndex(String index, Settings settings, Map<String, Object> mapping) {
if (index == null) {
logger.warn("no index name given to create index");
logger.warn("no index name give to create index");
return;
}
ensureClientIsPresent();
@ -126,15 +113,13 @@ public abstract class AbstractBulkClient extends AbstractNativeClient implements
if (settings != null) {
createIndexRequestBuilder.setSettings(settings);
}
if (builder != null) {
// NOTE: addMapping(type, ...) API is very fragile. Use XConteBuilder for safe typing.
createIndexRequestBuilder.addMapping(TYPE_NAME, builder);
if (mapping != null) {
createIndexRequestBuilder.addMapping("_doc", mapping);
}
CreateIndexResponse createIndexResponse = createIndexRequestBuilder.execute().actionGet();
logger.info("index {} created: {}", index,
Strings.toString(createIndexResponse.toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS)));
waitForCluster("YELLOW", 30L, TimeUnit.SECONDS);
waitForShards(30L, TimeUnit.SECONDS);
if (createIndexResponse.isAcknowledged()) {
logger.info("index {} created", index);
}
}
@Override
@ -145,84 +130,75 @@ public abstract class AbstractBulkClient extends AbstractNativeClient implements
@Override
public void startBulk(String index, long startRefreshIntervalSeconds, long stopRefreshIntervalSeconds)
throws IOException {
if (bulkController != null) {
ensureClientIsPresent();
bulkController.startBulkMode(index, startRefreshIntervalSeconds, stopRefreshIntervalSeconds);
}
ensureClientIsPresent();
bulkController.startBulkMode(index, startRefreshIntervalSeconds, stopRefreshIntervalSeconds);
}
@Override
public void stopBulk(IndexDefinition indexDefinition) throws IOException {
if (bulkController != null) {
ensureClientIsPresent();
bulkController.stopBulkMode(indexDefinition);
}
ensureClientIsPresent();
bulkController.stopBulkMode(indexDefinition);
}
@Override
public void stopBulk(String index, long timeout, TimeUnit timeUnit) throws IOException {
if (bulkController != null) {
ensureClientIsPresent();
bulkController.stopBulkMode(index, timeout, timeUnit);
}
ensureClientIsPresent();
bulkController.stopBulkMode(index, timeout, timeUnit);
}
@Override
public BulkClient index(String index, String id, boolean create, String source) {
return index(new IndexRequest(index, TYPE_NAME, id).create(create)
.source(source.getBytes(StandardCharsets.UTF_8), XContentType.JSON));
return index(index, id, create, new BytesArray(source.getBytes(StandardCharsets.UTF_8)));
}
@Override
public BulkClient 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).type("_doc").id(id).create(create)
.source(source, XContentType.JSON));
}
@Override
public BulkClient index(IndexRequest indexRequest) {
ensureClientIsPresent();
bulkController.index(indexRequest);
bulkController.bulkIndex(indexRequest);
return this;
}
@Override
public BulkClient delete(String index, String id) {
return delete(new DeleteRequest(index, TYPE_NAME, id));
return delete(new DeleteRequest().index(index).type("_doc").id(id));
}
@Override
public BulkClient delete(DeleteRequest deleteRequest) {
ensureClientIsPresent();
bulkController.delete(deleteRequest);
bulkController.bulkDelete(deleteRequest);
return this;
}
@Override
public BulkClient update(String index, String id, BytesReference source) {
return update(new UpdateRequest(index, TYPE_NAME, id)
return update(new UpdateRequest().index(index).type("_doc").id(id)
.doc(source, XContentType.JSON));
}
@Override
public BulkClient update(String index, String id, String source) {
return update(new UpdateRequest(index, TYPE_NAME, id)
return update(new UpdateRequest().index(index).type("_doc").id(id)
.doc(source.getBytes(StandardCharsets.UTF_8), XContentType.JSON));
}
@Override
public BulkClient update(UpdateRequest updateRequest) {
ensureClientIsPresent();
bulkController.update(updateRequest);
bulkController.bulkUpdate(updateRequest);
return this;
}
@Override
public boolean waitForResponses(long timeout, TimeUnit timeUnit) {
ensureClientIsPresent();
boolean success = bulkController.waitForResponses(timeout, timeUnit);
logger.info("waited for all bulk responses: " + success);
return success;
return bulkController.waitForBulkResponses(timeout, timeUnit);
}
@Override

View file

@ -1,6 +1,7 @@
package org.xbib.elx.common;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.get.GetAction;
import org.elasticsearch.action.get.GetRequestBuilder;
import org.elasticsearch.action.get.GetResponse;
@ -15,9 +16,12 @@ import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollAction;
import org.elasticsearch.action.search.SearchScrollRequestBuilder;
import org.elasticsearch.action.search.ShardSearchFailure;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.search.SearchHit;
import org.xbib.elx.api.SearchClient;
import org.xbib.elx.api.SearchMetric;
import java.io.IOException;
import java.util.Arrays;
import java.util.Comparator;
import java.util.Optional;
@ -28,13 +32,45 @@ import java.util.function.Predicate;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
public abstract class AbstractSearchClient extends AbstractNativeClient implements SearchClient {
public abstract class AbstractSearchClient extends AbstractBasicClient implements SearchClient {
private SearchMetric searchMetric;
@Override
public SearchMetric getSearchMetric() {
return searchMetric;
}
@Override
public void init(Settings settings) throws IOException {
super.init(settings);
this.searchMetric = new DefaultSearchMetric();
searchMetric.init(settings);
}
@Override
public void close() throws IOException {
super.close();
if (searchMetric != null) {
searchMetric.close();
}
}
@Override
public Optional<GetResponse> get(Consumer<GetRequestBuilder> getRequestBuilderConsumer) {
GetRequestBuilder getRequestBuilder = new GetRequestBuilder(client, GetAction.INSTANCE);
getRequestBuilderConsumer.accept(getRequestBuilder);
GetResponse getResponse = getRequestBuilder.execute().actionGet();
ActionFuture<GetResponse> actionFuture = getRequestBuilder.execute();
searchMetric.getCurrentQueries().inc();
GetResponse getResponse = actionFuture.actionGet();
searchMetric.getCurrentQueries().dec();
searchMetric.getQueries().inc();
searchMetric.markTotalQueries(1);
if (getResponse.isExists()) {
searchMetric.getSucceededQueries().inc();
} else {
searchMetric.getEmptyQueries().inc();
}
return getResponse.isExists() ? Optional.of(getResponse) : Optional.empty();
}
@ -42,23 +78,46 @@ public abstract class AbstractSearchClient extends AbstractNativeClient implemen
public Optional<MultiGetResponse> multiGet(Consumer<MultiGetRequestBuilder> multiGetRequestBuilderConsumer) {
MultiGetRequestBuilder multiGetRequestBuilder = new MultiGetRequestBuilder(client, MultiGetAction.INSTANCE);
multiGetRequestBuilderConsumer.accept(multiGetRequestBuilder);
MultiGetResponse multiGetItemResponse = multiGetRequestBuilder.execute().actionGet();
return multiGetItemResponse.getResponses().length == 0 ? Optional.empty() : Optional.of(multiGetItemResponse);
ActionFuture<MultiGetResponse> actionFuture = multiGetRequestBuilder.execute();
searchMetric.getCurrentQueries().inc();
MultiGetResponse multiGetItemResponse = actionFuture.actionGet();
searchMetric.getCurrentQueries().dec();
searchMetric.getQueries().inc();
searchMetric.markTotalQueries(1);
boolean isempty = multiGetItemResponse.getResponses().length == 0;
if (isempty) {
searchMetric.getEmptyQueries().inc();
} else {
searchMetric.getSucceededQueries().inc();
}
return isempty ? Optional.empty() : Optional.of(multiGetItemResponse);
}
@Override
public Optional<SearchResponse> search(Consumer<SearchRequestBuilder> queryBuilder) {
SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client, SearchAction.INSTANCE);
queryBuilder.accept(searchRequestBuilder);
SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
ActionFuture<SearchResponse> actionFuture = searchRequestBuilder.execute();
searchMetric.getCurrentQueries().inc();
SearchResponse searchResponse = actionFuture.actionGet();
searchMetric.getCurrentQueries().dec();
searchMetric.getQueries().inc();
searchMetric.markTotalQueries(1);
if (searchResponse.getFailedShards() > 0) {
StringBuilder sb = new StringBuilder("Search failed:");
for (ShardSearchFailure failure : searchResponse.getShardFailures()) {
sb.append("\n").append(failure.reason());
}
searchMetric.getEmptyQueries().inc();
throw new ElasticsearchException(sb.toString());
}
return searchResponse.getHits().getHits().length == 0 ? Optional.empty() : Optional.of(searchResponse);
boolean isempty = searchResponse.getHits().getHits().length == 0;
if (isempty) {
searchMetric.getEmptyQueries().inc();
} else {
searchMetric.getSucceededQueries().inc();
}
return isempty ? Optional.empty() : Optional.of(searchResponse);
}
@Override
@ -67,12 +126,32 @@ public abstract class AbstractSearchClient extends AbstractNativeClient implemen
SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client, SearchAction.INSTANCE);
queryBuilder.accept(searchRequestBuilder);
searchRequestBuilder.setScroll(scrollTime).setSize(scrollSize);
SearchResponse originalSearchResponse = searchRequestBuilder.execute().actionGet();
ActionFuture<SearchResponse> actionFuture = searchRequestBuilder.execute();
searchMetric.getCurrentQueries().inc();
SearchResponse originalSearchResponse = actionFuture.actionGet();
searchMetric.getCurrentQueries().dec();
searchMetric.getQueries().inc();
searchMetric.markTotalQueries(1);
boolean isempty = originalSearchResponse.getHits().getTotalHits() == 0;
if (isempty) {
searchMetric.getEmptyQueries().inc();
} else {
searchMetric.getSucceededQueries().inc();
}
Stream<SearchResponse> infiniteResponses = Stream.iterate(originalSearchResponse,
searchResponse -> new SearchScrollRequestBuilder(client, SearchScrollAction.INSTANCE)
.setScrollId(searchResponse.getScrollId())
.setScroll(scrollTime)
.execute().actionGet());
searchResponse -> {
SearchScrollRequestBuilder searchScrollRequestBuilder =
new SearchScrollRequestBuilder(client, SearchScrollAction.INSTANCE)
.setScrollId(searchResponse.getScrollId())
.setScroll(scrollTime);
ActionFuture<SearchResponse> actionFuture1 = searchScrollRequestBuilder.execute();
searchMetric.getCurrentQueries().inc();
SearchResponse searchResponse1 = actionFuture1.actionGet();
searchMetric.getCurrentQueries().dec();
searchMetric.getQueries().inc();
searchMetric.markTotalQueries(1);
return searchResponse1;
});
Predicate<SearchResponse> condition = searchResponse -> searchResponse.getHits().getHits().length > 0;
Consumer<SearchResponse> lastAction = searchResponse -> {
ClearScrollRequestBuilder clearScrollRequestBuilder =

View file

@ -10,10 +10,11 @@ import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.xbib.elx.api.AdminClientProvider;
import org.xbib.elx.api.BulkClientProvider;
import org.xbib.elx.api.NativeClient;
import org.xbib.elx.api.BasicClient;
import org.xbib.elx.api.SearchClientProvider;
import java.io.IOException;
import java.util.Map;
import java.util.ServiceLoader;
@SuppressWarnings("rawtypes")
@ -23,7 +24,7 @@ public class ClientBuilder {
private final ElasticsearchClient client;
private ClassLoader classLoader;
private final ClassLoader classLoader;
private final Settings.Builder settingsBuilder;
@ -56,11 +57,6 @@ public class ClientBuilder {
return new ClientBuilder(client);
}
public ClientBuilder setClassLoader(ClassLoader classLoader) {
this.classLoader = classLoader;
return this;
}
public ClientBuilder setAdminClientProvider(Class<? extends AdminClientProvider> adminClientProvider) {
this.adminClientProvider = adminClientProvider;
return this;
@ -111,8 +107,27 @@ public class ClientBuilder {
return this;
}
public ClientBuilder put(Map<String, ?> map) {
for (Map.Entry<String, ?> entry : map.entrySet()) {
if (entry.getValue() == null) {
continue;
}
if (entry.getValue() instanceof String ||
entry.getValue() instanceof Integer ||
entry.getValue() instanceof Long ||
entry.getValue() instanceof Float ||
entry.getValue() instanceof TimeValue) {
settingsBuilder.put(entry.getKey(), entry.getValue().toString());
} else {
logger.log(Level.WARN, "skipping " + entry.getValue() +
" because invalid class type " + entry.getValue().getClass().getName());
}
}
return this;
}
@SuppressWarnings("unchecked")
public <C extends NativeClient> C build() throws IOException {
public <C extends BasicClient> C build() throws IOException {
Settings settings = settingsBuilder.build();
logger.log(Level.INFO, "settings = " + settings.toDelimitedString(','));
if (adminClientProvider != null) {
@ -145,6 +160,6 @@ public class ClientBuilder {
}
}
}
throw new IllegalArgumentException("no provider found");
throw new IllegalArgumentException("no provider");
}
}

View file

@ -2,9 +2,6 @@ package org.xbib.elx.common;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
@ -13,6 +10,7 @@ import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.xbib.elx.api.BulkClient;
import org.xbib.elx.api.BulkController;
import org.xbib.elx.api.BulkListener;
import org.xbib.elx.api.BulkMetric;
import org.xbib.elx.api.BulkProcessor;
import org.xbib.elx.api.IndexDefinition;
@ -29,10 +27,12 @@ public class DefaultBulkController implements BulkController {
private static final Logger logger = LogManager.getLogger(DefaultBulkController.class);
private final BulkClient client;
private final BulkClient bulkClient;
private final BulkMetric bulkMetric;
private BulkProcessor bulkProcessor;
private final List<String> indexNames;
private final Map<String, Long> startBulkRefreshIntervals;
@ -43,17 +43,11 @@ public class DefaultBulkController implements BulkController {
private final TimeUnit maxWaitTimeUnit;
private BulkProcessor bulkProcessor;
private BulkListener bulkListener;
private final AtomicBoolean active;
private boolean enableBulkLogging;
public DefaultBulkController(BulkClient client, BulkMetric bulkMetric) {
this.client = client;
this.bulkMetric = bulkMetric;
public DefaultBulkController(BulkClient bulkClient) {
this.bulkClient = bulkClient;
this.bulkMetric = new DefaultBulkMetric();
this.indexNames = new ArrayList<>();
this.active = new AtomicBoolean(false);
this.startBulkRefreshIntervals = new HashMap<>();
@ -62,13 +56,19 @@ public class DefaultBulkController implements BulkController {
this.maxWaitTimeUnit = TimeUnit.SECONDS;
}
@Override
public BulkMetric getBulkMetric() {
return bulkMetric;
}
@Override
public Throwable getLastBulkError() {
return bulkListener.getLastBulkError();
return bulkProcessor.getBulkListener().getLastBulkError();
}
@Override
public void init(Settings settings) {
bulkMetric.init(settings);
int maxActionsPerRequest = settings.getAsInt(Parameters.MAX_ACTIONS_PER_REQUEST.name(),
Parameters.DEFAULT_MAX_ACTIONS_PER_REQUEST.getNum());
int maxConcurrentRequests = settings.getAsInt(Parameters.MAX_CONCURRENT_REQUESTS.name(),
@ -78,10 +78,10 @@ public class DefaultBulkController implements BulkController {
ByteSizeValue maxVolumePerRequest = settings.getAsBytesSize(Parameters.MAX_VOLUME_PER_REQUEST.name(),
ByteSizeValue.parseBytesSizeValue(Parameters.DEFAULT_MAX_VOLUME_PER_REQUEST.getString(),
"maxVolumePerRequest"));
this.enableBulkLogging = settings.getAsBoolean(Parameters.ENABLE_BULK_LOGGING.name(),
boolean enableBulkLogging = settings.getAsBoolean(Parameters.ENABLE_BULK_LOGGING.name(),
Parameters.ENABLE_BULK_LOGGING.getValue());
this.bulkListener = new BulkListener();
this.bulkProcessor = DefaultBulkProcessor.builder(client.getClient(), bulkListener)
BulkListener bulkListener = new DefaultBulkListener(this, bulkMetric, enableBulkLogging);
this.bulkProcessor = DefaultBulkProcessor.builder(bulkClient.getClient(), bulkListener)
.setBulkActions(maxActionsPerRequest)
.setConcurrentRequests(maxConcurrentRequests)
.setFlushInterval(flushIngestInterval)
@ -96,6 +96,11 @@ public class DefaultBulkController implements BulkController {
}
}
@Override
public void inactivate() {
this.active.set(false);
}
@Override
public void startBulkMode(IndexDefinition indexDefinition) throws IOException {
startBulkMode(indexDefinition.getFullIndexName(), indexDefinition.getStartRefreshInterval(),
@ -111,72 +116,56 @@ public class DefaultBulkController implements BulkController {
startBulkRefreshIntervals.put(indexName, startRefreshIntervalInSeconds);
stopBulkRefreshIntervals.put(indexName, stopRefreshIntervalInSeconds);
if (startRefreshIntervalInSeconds != 0L) {
client.updateIndexSetting(indexName, "refresh_interval", startRefreshIntervalInSeconds + "s",
bulkClient.updateIndexSetting(indexName, "refresh_interval", startRefreshIntervalInSeconds + "s",
30L, TimeUnit.SECONDS);
}
}
}
@Override
public void index(IndexRequest indexRequest) {
public void bulkIndex(IndexRequest indexRequest) {
ensureActiveAndBulk();
if (!active.get()) {
throw new IllegalStateException("inactive");
}
try {
if (bulkMetric != null) {
bulkMetric.getCurrentIngest().inc(indexRequest.index(), indexRequest.type(), indexRequest.id());
}
bulkMetric.getCurrentIngest().inc(indexRequest.index(), indexRequest.type(), indexRequest.id());
bulkProcessor.add(indexRequest);
} catch (Exception e) {
bulkListener.lastBulkError = e;
active.set(false);
if (logger.isErrorEnabled()) {
logger.error("bulk add of index failed: " + e.getMessage(), e);
}
inactivate();
}
}
@Override
public void delete(DeleteRequest deleteRequest) {
if (!active.get()) {
throw new IllegalStateException("inactive");
}
public void bulkDelete(DeleteRequest deleteRequest) {
ensureActiveAndBulk();
try {
if (bulkMetric != null) {
bulkMetric.getCurrentIngest().inc(deleteRequest.index(), deleteRequest.type(), deleteRequest.id());
}
bulkMetric.getCurrentIngest().inc(deleteRequest.index(), deleteRequest.type(), deleteRequest.id());
bulkProcessor.add(deleteRequest);
} catch (Exception e) {
bulkListener.lastBulkError = e;
active.set(false);
if (logger.isErrorEnabled()) {
logger.error("bulk add of delete failed: " + e.getMessage(), e);
}
inactivate();
}
}
@Override
public void update(UpdateRequest updateRequest) {
if (!active.get()) {
throw new IllegalStateException("inactive");
}
public void bulkUpdate(UpdateRequest updateRequest) {
ensureActiveAndBulk();
try {
if (bulkMetric != null) {
bulkMetric.getCurrentIngest().inc(updateRequest.index(), updateRequest.type(), updateRequest.id());
}
bulkMetric.getCurrentIngest().inc(updateRequest.index(), updateRequest.type(), updateRequest.id());
bulkProcessor.add(updateRequest);
} catch (Exception e) {
bulkListener.lastBulkError = e;
active.set(false);
if (logger.isErrorEnabled()) {
logger.error("bulk add of update failed: " + e.getMessage(), e);
}
inactivate();
}
}
@Override
public boolean waitForResponses(long timeout, TimeUnit timeUnit) {
public boolean waitForBulkResponses(long timeout, TimeUnit timeUnit) {
try {
return bulkProcessor.awaitFlush(timeout, timeUnit);
} catch (InterruptedException e) {
@ -195,11 +184,11 @@ public class DefaultBulkController implements BulkController {
@Override
public void stopBulkMode(String index, long timeout, TimeUnit timeUnit) throws IOException {
flush();
if (waitForResponses(timeout, timeUnit)) {
if (waitForBulkResponses(timeout, timeUnit)) {
if (indexNames.contains(index)) {
Long secs = stopBulkRefreshIntervals.get(index);
if (secs != null && secs != 0L) {
client.updateIndexSetting(index, "refresh_interval", secs + "s",
bulkClient.updateIndexSetting(index, "refresh_interval", secs + "s",
30L, TimeUnit.SECONDS);
}
indexNames.remove(index);
@ -217,11 +206,11 @@ public class DefaultBulkController implements BulkController {
@Override
public void close() throws IOException {
flush();
if (client.waitForResponses(maxWaitTime, maxWaitTimeUnit)) {
if (bulkClient.waitForResponses(maxWaitTime, maxWaitTimeUnit)) {
for (String index : indexNames) {
Long secs = stopBulkRefreshIntervals.get(index);
if (secs != null && secs != 0L)
client.updateIndexSetting(index, "refresh_interval", secs + "s",
bulkClient.updateIndexSetting(index, "refresh_interval", secs + "s",
30L, TimeUnit.SECONDS);
}
indexNames.clear();
@ -238,92 +227,5 @@ public class DefaultBulkController implements BulkController {
if (bulkProcessor == null) {
throw new UnsupportedOperationException("bulk processor not present");
}
if (bulkListener == null) {
throw new UnsupportedOperationException("bulk listener not present");
}
}
private class BulkListener implements DefaultBulkProcessor.Listener {
private final Logger logger = LogManager.getLogger(BulkListener.class.getName());
private Throwable lastBulkError = null;
@Override
public void beforeBulk(long executionId, BulkRequest request) {
long l = 0;
if (bulkMetric != null) {
l = bulkMetric.getCurrentIngest().getCount();
bulkMetric.getCurrentIngest().inc();
int n = request.numberOfActions();
bulkMetric.getSubmitted().inc(n);
bulkMetric.getCurrentIngestNumDocs().inc(n);
bulkMetric.getTotalIngestSizeInBytes().inc(request.estimatedSizeInBytes());
}
if (enableBulkLogging && logger.isDebugEnabled()) {
logger.debug("before bulk [{}] [actions={}] [bytes={}] [concurrent requests={}]",
executionId,
request.numberOfActions(),
request.estimatedSizeInBytes(),
l);
}
}
@Override
public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
long l = 0;
if (bulkMetric != null) {
l = bulkMetric.getCurrentIngest().getCount();
bulkMetric.getCurrentIngest().dec();
bulkMetric.getSucceeded().inc(response.getItems().length);
}
int n = 0;
for (BulkItemResponse itemResponse : response.getItems()) {
if (bulkMetric != null) {
bulkMetric.getCurrentIngest().dec(itemResponse.getIndex(), itemResponse.getType(), itemResponse.getId());
}
if (itemResponse.isFailed()) {
n++;
if (bulkMetric != null) {
bulkMetric.getSucceeded().dec(1);
bulkMetric.getFailed().inc(1);
}
}
}
if (enableBulkLogging && logger.isDebugEnabled() && bulkMetric != null) {
logger.debug("after bulk [{}] [succeeded={}] [failed={}] [{}ms] {} concurrent requests",
executionId,
bulkMetric.getSucceeded().getCount(),
bulkMetric.getFailed().getCount(),
response.getTook().millis(),
l);
}
if (n > 0) {
if (enableBulkLogging && logger.isErrorEnabled()) {
logger.error("bulk [{}] failed with {} failed items, failure message = {}",
executionId, n, response.buildFailureMessage());
}
} else {
if (bulkMetric != null) {
bulkMetric.getCurrentIngestNumDocs().dec(response.getItems().length);
}
}
}
@Override
public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
if (bulkMetric != null) {
bulkMetric.getCurrentIngest().dec();
}
lastBulkError = failure;
active.set(false);
if (enableBulkLogging && logger.isErrorEnabled()) {
logger.error("after bulk [" + executionId + "] error", failure);
}
}
Throwable getLastBulkError() {
return lastBulkError;
}
}
}

View file

@ -0,0 +1,95 @@
package org.xbib.elx.common;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.xbib.elx.api.BulkController;
import org.xbib.elx.api.BulkListener;
import org.xbib.elx.api.BulkMetric;
public class DefaultBulkListener implements BulkListener {
private final Logger logger = LogManager.getLogger(DefaultBulkListener.class.getName());
private final BulkController bulkController;
private final BulkMetric bulkMetric;
private final boolean isBulkLoggingEnabled;
private Throwable lastBulkError = null;
public DefaultBulkListener(BulkController bulkController,
BulkMetric bulkMetric,
boolean isBulkLoggingEnabled) {
this.bulkController = bulkController;
this.bulkMetric = bulkMetric;
this.isBulkLoggingEnabled = isBulkLoggingEnabled;
}
@Override
public void beforeBulk(long executionId, BulkRequest request) {
long l = bulkMetric.getCurrentIngest().getCount();
bulkMetric.getCurrentIngest().inc();
int n = request.numberOfActions();
bulkMetric.getSubmitted().inc(n);
bulkMetric.getCurrentIngestNumDocs().inc(n);
bulkMetric.getTotalIngestSizeInBytes().inc(request.estimatedSizeInBytes());
if (isBulkLoggingEnabled && logger.isDebugEnabled()) {
logger.debug("before bulk [{}] [actions={}] [bytes={}] [concurrent requests={}]",
executionId,
request.numberOfActions(),
request.estimatedSizeInBytes(),
l);
}
}
@Override
public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
long l = bulkMetric.getCurrentIngest().getCount();
bulkMetric.getCurrentIngest().dec();
bulkMetric.getSucceeded().inc(response.getItems().length);
int n = 0;
for (BulkItemResponse itemResponse : response.getItems()) {
bulkMetric.getCurrentIngest().dec(itemResponse.getIndex(), itemResponse.getType(), itemResponse.getId());
if (itemResponse.isFailed()) {
n++;
bulkMetric.getSucceeded().dec(1);
bulkMetric.getFailed().inc(1);
}
}
if (isBulkLoggingEnabled && logger.isDebugEnabled()) {
logger.debug("after bulk [{}] [succeeded={}] [failed={}] [{}ms] {} concurrent requests",
executionId,
bulkMetric.getSucceeded().getCount(),
bulkMetric.getFailed().getCount(),
response.getTook().millis(),
l);
}
if (n > 0) {
if (isBulkLoggingEnabled && logger.isErrorEnabled()) {
logger.error("bulk [{}] failed with {} failed items, failure message = {}",
executionId, n, response.buildFailureMessage());
}
} else {
bulkMetric.getCurrentIngestNumDocs().dec(response.getItems().length);
}
}
@Override
public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
bulkMetric.getCurrentIngest().dec();
lastBulkError = failure;
if (logger.isErrorEnabled()) {
logger.error("after bulk [" + executionId + "] error", failure);
}
bulkController.inactivate();
}
@Override
public Throwable getLastBulkError() {
return lastBulkError;
}
}

View file

@ -44,6 +44,11 @@ public class DefaultBulkMetric implements BulkMetric {
start();
}
@Override
public void markTotalIngest(long n) {
totalIngest.mark(n);
}
@Override
public Metered getTotalIngest() {
return totalIngest;

View file

@ -15,7 +15,9 @@ import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.util.concurrent.EsExecutors;
import org.elasticsearch.common.util.concurrent.FutureUtils;
import org.xbib.elx.api.BulkListener;
import org.xbib.elx.api.BulkProcessor;
import org.xbib.elx.api.BulkRequestHandler;
import java.util.Objects;
import java.util.concurrent.Executors;
@ -26,13 +28,17 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
/**
* A bulk processor is a thread safe bulk processing class, allowing to easily set when to "flush" a new bulk request
* (either based on number of actions, based on the size, or time), and to easily control the number of concurrent bulk
* A bulk processor is a thread safe bulk processing class, allowing to easily
* set when to "flush" a new bulk request
* (either based on number of actions, based on the size, or time), and
* to easily control the number of concurrent bulk
* requests allowed to be executed in parallel.
* In order to create a new bulk processor, use the {@link Builder}.
*/
public class DefaultBulkProcessor implements BulkProcessor {
private final BulkListener bulkListener;
private final int bulkActions;
private final long bulkSize;
@ -49,16 +55,22 @@ public class DefaultBulkProcessor implements BulkProcessor {
private volatile boolean closed;
private DefaultBulkProcessor(ElasticsearchClient client, Listener listener, String name, int concurrentRequests,
int bulkActions, ByteSizeValue bulkSize, TimeValue flushInterval) {
private DefaultBulkProcessor(ElasticsearchClient client,
BulkListener bulkListener,
String name,
int concurrentRequests,
int bulkActions,
ByteSizeValue bulkSize,
TimeValue flushInterval) {
this.bulkListener = bulkListener;
this.executionIdGen = new AtomicLong();
this.closed = false;
this.bulkActions = bulkActions;
this.bulkSize = bulkSize.getBytes();
this.bulkRequest = new BulkRequest();
this.bulkRequestHandler = concurrentRequests == 0 ?
new SyncBulkRequestHandler(client, listener) :
new AsyncBulkRequestHandler(client, listener, concurrentRequests);
new SyncBulkRequestHandler(client, bulkListener) :
new AsyncBulkRequestHandler(client, bulkListener, concurrentRequests);
if (flushInterval != null) {
this.scheduler = (ScheduledThreadPoolExecutor) Executors.newScheduledThreadPool(1,
EsExecutors.daemonThreadFactory(Settings.EMPTY,
@ -73,10 +85,15 @@ public class DefaultBulkProcessor implements BulkProcessor {
}
}
public static Builder builder(ElasticsearchClient client, Listener listener) {
Objects.requireNonNull(client, "The client you specified while building a BulkProcessor is null");
Objects.requireNonNull(listener, "A listener for the BulkProcessor is required but null");
return new Builder(client, listener);
public static Builder builder(ElasticsearchClient client,
BulkListener bulkListener) {
Objects.requireNonNull(bulkListener, "A listener for the BulkProcessor is required but null");
return new Builder(client, bulkListener);
}
@Override
public BulkListener getBulkListener() {
return bulkListener;
}
/**
@ -140,19 +157,7 @@ public class DefaultBulkProcessor implements BulkProcessor {
*/
@Override
public DefaultBulkProcessor add(ActionRequest request) {
return add(request, null);
}
/**
* 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);
internalAdd(request);
return this;
}
@ -186,14 +191,14 @@ public class DefaultBulkProcessor implements BulkProcessor {
}
}
private synchronized void internalAdd(ActionRequest request, Object payload) {
private synchronized void internalAdd(ActionRequest request) {
ensureOpen();
if (request instanceof IndexRequest) {
bulkRequest.add((IndexRequest) request, payload);
bulkRequest.add((IndexRequest) request);
} else if (request instanceof DeleteRequest) {
bulkRequest.add((DeleteRequest) request, payload);
bulkRequest.add((DeleteRequest) request);
} else if (request instanceof UpdateRequest) {
bulkRequest.add((UpdateRequest) request, payload);
bulkRequest.add((UpdateRequest) request);
} else {
throw new UnsupportedOperationException();
}
@ -228,7 +233,7 @@ public class DefaultBulkProcessor implements BulkProcessor {
private final ElasticsearchClient client;
private final Listener listener;
private final BulkListener bulkListener;
private String name;
@ -245,11 +250,11 @@ public class DefaultBulkProcessor implements BulkProcessor {
* to be notified on the completion of bulk requests.
*
* @param client the client
* @param listener the listener
* @param bulkListener the listener
*/
Builder(ElasticsearchClient client, Listener listener) {
Builder(ElasticsearchClient client, BulkListener bulkListener) {
this.client = client;
this.listener = listener;
this.bulkListener = bulkListener;
}
/**
@ -319,7 +324,7 @@ public class DefaultBulkProcessor implements BulkProcessor {
* @return a bulk processor
*/
public DefaultBulkProcessor build() {
return new DefaultBulkProcessor(client, listener, name, concurrentRequests, bulkActions, bulkSize, flushInterval);
return new DefaultBulkProcessor(client, bulkListener, name, concurrentRequests, bulkActions, bulkSize, flushInterval);
}
}
@ -343,25 +348,25 @@ public class DefaultBulkProcessor implements BulkProcessor {
private final ElasticsearchClient client;
private final DefaultBulkProcessor.Listener listener;
private final BulkListener bulkListener;
SyncBulkRequestHandler(ElasticsearchClient client, DefaultBulkProcessor.Listener listener) {
Objects.requireNonNull(listener, "A listener is required for SyncBulkRequestHandler but null");
SyncBulkRequestHandler(ElasticsearchClient client, BulkListener bulkListener) {
Objects.requireNonNull(bulkListener, "A listener is required for SyncBulkRequestHandler but null");
this.client = client;
this.listener = listener;
this.bulkListener = bulkListener;
}
@Override
public void execute(BulkRequest bulkRequest, long executionId) {
boolean afterCalled = false;
try {
listener.beforeBulk(executionId, bulkRequest);
bulkListener.beforeBulk(executionId, bulkRequest);
BulkResponse bulkResponse = client.execute(BulkAction.INSTANCE, bulkRequest).actionGet();
afterCalled = true;
listener.afterBulk(executionId, bulkRequest, bulkResponse);
bulkListener.afterBulk(executionId, bulkRequest, bulkResponse);
} catch (Exception e) {
if (!afterCalled) {
listener.afterBulk(executionId, bulkRequest, e);
bulkListener.afterBulk(executionId, bulkRequest, e);
}
}
}
@ -376,13 +381,15 @@ public class DefaultBulkProcessor implements BulkProcessor {
private final ElasticsearchClient client;
private final DefaultBulkProcessor.Listener listener;
private final BulkListener listener;
private final Semaphore semaphore;
private final int concurrentRequests;
private AsyncBulkRequestHandler(ElasticsearchClient client, DefaultBulkProcessor.Listener listener, int concurrentRequests) {
private AsyncBulkRequestHandler(ElasticsearchClient client,
BulkListener listener,
int concurrentRequests) {
Objects.requireNonNull(listener, "A listener is required for AsyncBulkRequestHandler but null");
this.client = client;
this.listener = listener;

View file

@ -0,0 +1,97 @@
package org.xbib.elx.common;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.common.settings.Settings;
import org.xbib.elx.api.SearchMetric;
import org.xbib.metrics.api.Count;
import org.xbib.metrics.api.Metered;
import org.xbib.metrics.common.CountMetric;
import org.xbib.metrics.common.Meter;
import java.util.concurrent.Executors;
public class DefaultSearchMetric implements SearchMetric {
private static final Logger logger = LogManager.getLogger(DefaultSearchMetric.class.getName());
private final Meter totalQuery;
private final Count currentQuery;
private final Count queries;
private final Count succeededQueries;
private final Count emptyQueries;
private Long started;
private Long stopped;
public DefaultSearchMetric() {
totalQuery = new Meter(Executors.newSingleThreadScheduledExecutor());
currentQuery = new CountMetric();
queries = new CountMetric();
succeededQueries = new CountMetric();
emptyQueries = new CountMetric();
}
@Override
public void init(Settings settings) {
logger.info("init");
start();
}
@Override
public void markTotalQueries(long n) {
totalQuery.mark(n);
}
@Override
public Metered getTotalQueries() {
return totalQuery;
}
@Override
public Count getCurrentQueries() {
return currentQuery;
}
@Override
public Count getQueries() {
return queries;
}
@Override
public Count getSucceededQueries() {
return succeededQueries;
}
@Override
public Count getEmptyQueries() {
return emptyQueries;
}
@Override
public long elapsed() {
return started != null ? ((stopped != null ? stopped : System.nanoTime()) - started) : -1L;
}
@Override
public void start() {
this.started = System.nanoTime();
totalQuery.start(5L);
}
@Override
public void stop() {
this.stopped = System.nanoTime();
totalQuery.stop();
}
@Override
public void close() {
stop();
totalQuery.shutdown();
}
}

View file

@ -2,22 +2,13 @@ package org.xbib.elx.common;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.settings.Settings;
import org.xbib.elx.api.*;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
/**
* A mocked client, it does not perform any actions on a cluster. Useful for testing.
*/
public class MockAdminClient extends MockNativeClient implements AdminClient {
@Override
public void setClient(ElasticsearchClient client) {
}
public class MockAdminClient extends AbstractAdminClient {
@Override
public ElasticsearchClient getClient() {
@ -29,42 +20,16 @@ public class MockAdminClient extends MockNativeClient implements AdminClient {
}
@Override
public String getClusterName() {
protected ElasticsearchClient createClient(Settings settings) {
return null;
}
@Override
public String getHealthColor(long maxWaitTime, TimeUnit timeUnit) {
return null;
protected void closeClient(Settings settings) {
}
@Override
public IndexDefinition buildIndexDefinitionFromSettings(String index, Settings settings) throws IOException {
return null;
}
@Override
public Map<String, ?> getMapping(String index) {
return null;
}
@Override
public Map<String, ?> getMapping(String index, String type) {
return null;
}
@Override
public AdminClient deleteIndex(IndexDefinition indexDefinition) {
return this;
}
@Override
public AdminClient deleteIndex(String index) {
return this;
}
@Override
public AdminClient updateReplicaLevel(IndexDefinition indexDefinition, int level) throws IOException {
public MockAdminClient deleteIndex(String index) {
return this;
}
@ -73,56 +38,6 @@ public class MockAdminClient extends MockNativeClient implements AdminClient {
return true;
}
@Override
public String resolveAlias(String alias) {
return null;
}
@Override
public String resolveMostRecentIndex(String alias) {
return null;
}
@Override
public Map<String, String> getAliases(String index) {
return null;
}
@Override
public IndexShiftResult shiftIndex(IndexDefinition indexDefinition, List<String> additionalAliases) {
return null;
}
@Override
public IndexShiftResult shiftIndex(IndexDefinition indexDefinition, List<String> additionalAliases, IndexAliasAdder indexAliasAdder) {
return null;
}
@Override
public IndexShiftResult shiftIndex(String index, String fullIndexName, List<String> additionalAliases) {
return null;
}
@Override
public IndexShiftResult shiftIndex(String index, String fullIndexName, List<String> additionalAliases, IndexAliasAdder adder) {
return null;
}
@Override
public IndexPruneResult pruneIndex(IndexDefinition indexDefinition) {
return null;
}
@Override
public IndexPruneResult pruneIndex(String index, String fullIndexName, int delta, int mintokeep, boolean perform) {
return null;
}
@Override
public Long mostRecentDocument(String index, String timestampfieldname) throws IOException {
return null;
}
@Override
public void waitForCluster(String healthColor, long timeValue, TimeUnit timeUnit) {
}
@ -133,35 +48,10 @@ public class MockAdminClient extends MockNativeClient implements AdminClient {
}
@Override
public long getSearchableDocs(String index) {
return 0;
}
@Override
public boolean isIndexExists(String index) {
return false;
}
@Override
public AdminClient updateReplicaLevel(String index, int level, long maxWaitTime, TimeUnit timeUnit) {
public MockAdminClient updateReplicaLevel(String index, int level, long maxWaitTime, TimeUnit timeUnit) {
return this;
}
@Override
public int getReplicaLevel(IndexDefinition indexDefinition) {
return 0;
}
@Override
public int getReplicaLevel(String index) {
return 0;
}
@Override
public boolean forceMerge(IndexDefinition indexDefinition) {
return false;
}
@Override
public void close() {
// nothing to do

View file

@ -4,22 +4,13 @@ import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.xbib.elx.api.BulkClient;
import org.xbib.elx.api.BulkController;
import org.xbib.elx.api.BulkMetric;
import org.xbib.elx.api.IndexDefinition;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.TimeUnit;
/**
* A mocked client, it does not perform any actions on a cluster. Useful for testing.
*/
public class MockBulkClient extends MockNativeClient implements BulkClient {
public class MockBulkClient extends AbstractBulkClient {
@Override
public ElasticsearchClient getClient() {
@ -35,53 +26,17 @@ public class MockBulkClient extends MockNativeClient implements BulkClient {
return null;
}
@Override
public void waitForShards(long maxWaitTime, TimeUnit timeUnit) {
}
@Override
protected ElasticsearchClient createClient(Settings settings) {
return null;
}
@Override
protected void closeClient() {
}
@Override
public BulkMetric getBulkMetric() {
return null;
}
@Override
public BulkController getBulkController() {
return null;
}
@Override
public void newIndex(String index) throws IOException {
}
@Override
public void newIndex(IndexDefinition indexDefinition) throws IOException {
}
@Override
public void newIndex(String index, Settings settings) throws IOException {
}
@Override
public void newIndex(String index, Settings settings, XContentBuilder mapping) throws IOException {
}
@Override
public void newIndex(String index, Settings settings, Map<String, ?> mapping) throws IOException {
}
@Override
public BulkClient index(String index, String id, boolean create, BytesReference source) {
return null;
protected void closeClient(Settings settings) {
}
@Override
@ -109,30 +64,15 @@ public class MockBulkClient extends MockNativeClient implements BulkClient {
return this;
}
@Override
public BulkClient update(String index, String id, BytesReference source) {
return null;
}
@Override
public MockBulkClient update(UpdateRequest updateRequest) {
return this;
}
@Override
public void startBulk(IndexDefinition indexDefinition) throws IOException {
}
@Override
public void startBulk(String index, long startRefreshInterval, long stopRefreshIterval) {
}
@Override
public void stopBulk(IndexDefinition indexDefinition) throws IOException {
}
@Override
public void stopBulk(String index, long maxWaitTime, TimeUnit timeUnit) {
}
@ -142,11 +82,6 @@ public class MockBulkClient extends MockNativeClient implements BulkClient {
return true;
}
@Override
public void updateIndexSetting(String index, String key, Object value, long timeout, TimeUnit timeUnit) throws IOException {
}
@Override
public void refreshIndex(String index) {
}

View file

@ -1,74 +0,0 @@
package org.xbib.elx.common;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.settings.Settings;
import org.xbib.elx.api.NativeClient;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
public class MockNativeClient extends AbstractNativeClient implements NativeClient {
@Override
protected void ensureClientIsPresent() {
}
@Override
public void setClient(ElasticsearchClient client) {
}
@Override
public ElasticsearchClient getClient() {
return null;
}
@Override
protected ElasticsearchClient createClient(Settings settings) throws IOException {
return null;
}
@Override
protected void closeClient() throws IOException {
}
@Override
public void init(Settings settings) throws IOException {
}
@Override
public String getClusterName() {
return null;
}
@Override
public String getHealthColor(long maxWaitTime, TimeUnit timeUnit) {
return null;
}
@Override
public void waitForCluster(String healthColor, long maxWaitTime, TimeUnit timeUnit) {
}
@Override
public void waitForShards(long maxWaitTime, TimeUnit timeUnit) {
}
@Override
public long getSearchableDocs(String index) {
return 0;
}
@Override
public boolean isIndexExists(String index) {
return false;
}
@Override
public void close() throws IOException {
}
}

View file

@ -1,25 +1,14 @@
package org.xbib.elx.common;
import org.elasticsearch.action.get.GetRequestBuilder;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetRequestBuilder;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.search.SearchHit;
import org.xbib.elx.api.SearchClient;
import java.util.Optional;
import java.util.function.Consumer;
import java.util.stream.Stream;
import java.util.concurrent.TimeUnit;
/**
* A mocked client, it does not perform any actions on a cluster. Useful for testing.
*/
public class MockSearchClient extends MockNativeClient implements SearchClient {
public class MockSearchClient extends AbstractSearchClient {
@Override
public ElasticsearchClient getClient() {
@ -35,42 +24,21 @@ public class MockSearchClient extends MockNativeClient implements SearchClient {
return null;
}
@Override
public void waitForShards(long maxWaitTime, TimeUnit timeUnit) {
}
@Override
protected ElasticsearchClient createClient(Settings settings) {
return null;
}
@Override
protected void closeClient() {
protected void closeClient(Settings settings) {
}
@Override
public void close() {
// nothing to do
}
@Override
public Optional<GetResponse> get(Consumer<GetRequestBuilder> getRequestBuilder) {
return Optional.empty();
}
@Override
public Optional<MultiGetResponse> multiGet(Consumer<MultiGetRequestBuilder> multiGetRequestBuilder) {
return Optional.empty();
}
@Override
public Optional<SearchResponse> search(Consumer<SearchRequestBuilder> searchRequestBuilder) {
return Optional.empty();
}
@Override
public Stream<SearchHit> search(Consumer<SearchRequestBuilder> searchRequestBuilder, TimeValue scrollTime, int scrollSize) {
return null;
}
@Override
public Stream<String> getIds(Consumer<SearchRequestBuilder> queryBuilder) {
return null;
}
}

View file

@ -1,7 +1,13 @@
package org.xbib.elx.common.test;
import org.junit.jupiter.api.Test;
import org.xbib.elx.common.*;
import org.xbib.elx.common.ClientBuilder;
import org.xbib.elx.common.MockAdminClient;
import org.xbib.elx.common.MockAdminClientProvider;
import org.xbib.elx.common.MockBulkClient;
import org.xbib.elx.common.MockBulkClientProvider;
import org.xbib.elx.common.MockSearchClient;
import org.xbib.elx.common.MockSearchClientProvider;
import java.io.IOException;

View file

@ -6,13 +6,12 @@ import org.elasticsearch.action.bulk.BulkAction;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchAction;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@ -33,7 +32,7 @@ class SearchTest {
ElasticsearchClient client = helper.client("1");
BulkRequestBuilder builder = new BulkRequestBuilder(client, BulkAction.INSTANCE);
for (int i = 0; i < 1000; i++) {
IndexRequest indexRequest = new IndexRequest("pages", "row")
IndexRequest indexRequest = new IndexRequest().index("pages").type("_doc")
.source(XContentFactory.jsonBuilder()
.startObject()
.field("user1", "joerg")
@ -54,16 +53,14 @@ class SearchTest {
client.execute(RefreshAction.INSTANCE, new RefreshRequest()).actionGet();
for (int i = 0; i < 1; i++) {
QueryBuilder queryStringBuilder = QueryBuilders.queryStringQuery("rs:" + 1234);
SearchSourceBuilder searchSource = new SearchSourceBuilder();
searchSource.query(queryStringBuilder);
searchSource.sort("rowcount", SortOrder.DESC);
searchSource.from(i * 10);
searchSource.size(10);
SearchRequest searchRequest = new SearchRequest();
searchRequest.indices("pages");
searchRequest.types("row");
searchRequest.source(searchSource);
SearchResponse searchResponse = client.execute(SearchAction.INSTANCE, searchRequest).actionGet();
SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client, SearchAction.INSTANCE)
.setIndices("pages")
.setQuery(queryStringBuilder)
.addSort("rowcount", SortOrder.DESC)
.setFrom(i * 10)
.setSize(10)
.setTrackTotalHits(true);
SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
assertTrue(searchResponse.getHits().getTotalHits() > 0);
}
}

View file

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

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.ClusterStateRequest;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.analysis.common.CommonAnalysisPlugin;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.client.support.AbstractClient;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
@ -37,7 +36,7 @@ import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@ -215,7 +214,7 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
.put(getNodeSettings())
.put("node.name", id)
.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);
AbstractClient client = (AbstractClient) node.client();
nodes.put(id, node);

View file

@ -5,12 +5,11 @@ import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
import org.elasticsearch.action.index.IndexAction;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchAction;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@ -45,20 +44,18 @@ class WildcardTest {
}
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").type("_doc").id(id)
.source(XContentFactory.jsonBuilder().startObject().field("field", fieldValue).endObject()))
.actionGet();
client.execute(RefreshAction.INSTANCE, new RefreshRequest()).actionGet();
}
private long count(ElasticsearchClient client, QueryBuilder queryBuilder) {
SearchSourceBuilder builder = new SearchSourceBuilder();
builder.query(queryBuilder);
SearchRequest searchRequest = new SearchRequest();
searchRequest.indices("index");
searchRequest.types("type");
searchRequest.source(builder);
return client.execute(SearchAction.INSTANCE, searchRequest).actionGet().getHits().getTotalHits();
SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client, SearchAction.INSTANCE)
.setIndices("index")
.setQuery(queryBuilder)
.setTrackTotalHits(true);
return searchRequestBuilder.execute().actionGet().getHits().getTotalHits();
}
private void validateCount(ElasticsearchClient client, QueryBuilder queryBuilder, long expectedHits) {

View file

@ -2,13 +2,12 @@ import org.apache.tools.ant.taskdefs.condition.Os
dependencies{
api project(':elx-common')
api "org.xbib:netty-http-client:${project.property('xbib-netty-http.version')}"
implementation "org.xbib.elasticsearch:transport-netty4:${rootProject.property('elasticsearch-server.version')}"
implementation "org.xbib:netty-http-client:${project.property('xbib-netty-http.version')}"
testRuntimeOnly "org.bouncycastle:bcpkix-jdk15on:${project.property('bouncycastle.version')}"
runtimeOnly "org.bouncycastle:bcpkix-jdk15on:${project.property('bouncycastle.version')}"
if (Os.isFamily(Os.FAMILY_MAC)) {
testRuntimeOnly "io.netty:netty-tcnative-boringssl-static:${project.property('tcnative-legacy-macosx.version')}"
runtimeOnly "io.netty:netty-tcnative-boringssl-static:${project.property('tcnative-legacy-macosx.version')}"
} else if (Os.isFamily(Os.FAMILY_UNIX)) {
testRuntimeOnly "io.netty:netty-tcnative-boringssl-static:${project.property('tcnative.version')}"
runtimeOnly "io.netty:netty-tcnative-boringssl-static:${project.property('tcnative.version')}"
}
testImplementation "org.xbib.elasticsearch:elasticsearch-analysis-common:${rootProject.property('elasticsearch-server.version')}"
}

View file

@ -1,6 +1,7 @@
package org.elasticsearch.action.admin.indices.get;
import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
import org.apache.logging.log4j.Level;
import org.apache.lucene.util.CollectionUtil;
import org.elasticsearch.cluster.metadata.AliasMetaData;
import org.elasticsearch.cluster.metadata.MappingMetaData;
@ -10,6 +11,7 @@ import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
@ -34,15 +36,10 @@ public class HttpGetIndexAction extends HttpAction<GetIndexRequest, GetIndexResp
}
@Override
protected CheckedFunction<XContentParser, GetIndexResponse, IOException> entityParser() {
protected CheckedFunction<XContentParser, GetIndexResponse, IOException> entityParser(HttpResponse httpResponse) {
return this::fromXContent;
}
@Override
protected GetIndexResponse emptyResponse() {
return new GetIndexResponse();
}
private GetIndexResponse fromXContent(XContentParser parser) throws IOException {
ImmutableOpenMap.Builder<String, List<AliasMetaData>> aliases = ImmutableOpenMap.builder();
ImmutableOpenMap.Builder<String, ImmutableOpenMap<String, MappingMetaData>> mappings = ImmutableOpenMap.builder();
@ -68,6 +65,7 @@ public class HttpGetIndexAction extends HttpAction<GetIndexRequest, GetIndexResp
parser.nextToken();
}
}
logger.log(Level.INFO, "indices = " + indices);
return new GetIndexResponse(indices.toArray(new String[0]),
mappings.build(),
aliases.build(),

View file

@ -7,6 +7,7 @@ import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException;
import java.util.Map;
@ -29,15 +30,10 @@ public class HttpGetMappingsAction extends HttpAction<GetMappingsRequest, GetMap
}
@Override
protected CheckedFunction<XContentParser, GetMappingsResponse, IOException> entityParser() {
protected CheckedFunction<XContentParser, GetMappingsResponse, IOException> entityParser(HttpResponse httpResponse) {
return this::fromXContent;
}
@Override
protected GetMappingsResponse emptyResponse() {
return new GetMappingsResponse();
}
@SuppressWarnings("unchecked")
private GetMappingsResponse fromXContent(XContentParser parser) throws IOException {
XContentParser.Token token = parser.nextToken();

View file

@ -25,6 +25,7 @@ import org.elasticsearch.rest.RestStatus;
import org.xbib.net.URL;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.client.api.Transport;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
@ -71,13 +72,13 @@ public abstract class HttpAction<R extends ActionRequest, T extends ActionRespon
} else {
ElasticsearchStatusException statusException = parseToError(httpActionContext);
if (statusException.status().equals(RestStatus.NOT_FOUND)) {
listener.onResponse(emptyResponse());
listener.onResponse(parseToResponse(httpActionContext));
} else {
listener.onFailure(statusException);
}
}
});
Transport transport = httpActionContext.getHelper().internalClient().execute(httpRequest);
Transport transport = httpActionContext.getExtendedHttpClient().internalClient().execute(httpRequest);
httpActionContext.setHttpClientTransport(transport);
if (transport.isFailed()) {
listener.onFailure(new Exception(transport.getFailure()));
@ -155,11 +156,11 @@ public abstract class HttpAction<R extends ActionRequest, T extends ActionRespon
throw new IllegalStateException("unsupported content-type: " + mediaType);
}
try (XContentParser parser = xContentType.xContent()
.createParser(httpActionContext.getHelper().getRegistry(),
.createParser(httpActionContext.getExtendedHttpClient().getRegistry(),
DeprecationHandler.THROW_UNSUPPORTED_OPERATION,
httpActionContext.getHttpResponse().getBody().toString(StandardCharsets.UTF_8))) {
return entityParser().apply(parser);
} catch (IOException e) {
return entityParser(httpActionContext.getHttpResponse()).apply(parser);
} catch (Exception e) {
logger.error(e.getMessage(), e);
return null;
}
@ -167,7 +168,7 @@ public abstract class HttpAction<R extends ActionRequest, T extends ActionRespon
protected ElasticsearchStatusException parseToError(HttpActionContext<R, T> httpActionContext) {
try (XContentParser parser = XContentFactory.xContent(XContentType.JSON)
.createParser(httpActionContext.getHelper().getRegistry(),
.createParser(httpActionContext.getExtendedHttpClient().getRegistry(),
DeprecationHandler.THROW_UNSUPPORTED_OPERATION,
httpActionContext.getHttpResponse().getBody().toString(StandardCharsets.UTF_8))) {
return errorParser().apply(parser);
@ -183,8 +184,5 @@ public abstract class HttpAction<R extends ActionRequest, T extends ActionRespon
protected abstract Request.Builder createHttpRequest(String baseUrl, R request) throws IOException;
protected abstract CheckedFunction<XContentParser, T, IOException> entityParser();
protected abstract T emptyResponse();
protected abstract CheckedFunction<XContentParser, T, IOException> entityParser(HttpResponse httpResponse);
}

View file

@ -13,7 +13,7 @@ import org.xbib.netty.http.common.HttpResponse;
*/
public class HttpActionContext<R extends ActionRequest, T extends ActionResponse> {
private final HttpClientHelper helper;
private final HttpClientHelper extendedHttpClient;
private final R request;
@ -23,14 +23,14 @@ public class HttpActionContext<R extends ActionRequest, T extends ActionResponse
private HttpResponse httpResponse;
HttpActionContext(HttpClientHelper helper, R request, String url) {
this.helper = helper;
public HttpActionContext(HttpClientHelper extendedHttpClient, R request, String url) {
this.extendedHttpClient = extendedHttpClient;
this.request = request;
this.url = url;
}
public HttpClientHelper getHelper() {
return helper;
public HttpClientHelper getExtendedHttpClient() {
return extendedHttpClient;
}
public R getRequest() {

View file

@ -31,12 +31,12 @@ public class HttpAdminClient extends AbstractAdminClient implements Elasticsearc
@Override
protected ElasticsearchClient createClient(Settings settings) throws IOException {
return helper.createClient(settings);
return this;
}
@Override
protected void closeClient() throws IOException {
helper.closeClient();
protected void closeClient(Settings settings) throws IOException {
helper.closeClient(settings);
}
@Override
@ -51,7 +51,7 @@ public class HttpAdminClient extends AbstractAdminClient implements Elasticsearc
@Override
public <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> RequestBuilder prepareExecute(Action<Request, Response, RequestBuilder> action) {
return helper.prepareExecute(action);
return action.newRequestBuilder(this);
}
@Override

View file

@ -3,6 +3,7 @@ package org.xbib.elx.http;
import org.xbib.elx.api.AdminClientProvider;
public class HttpAdminClientProvider implements AdminClientProvider<HttpAdminClient> {
@Override
public HttpAdminClient getClient() {
return new HttpAdminClient();

View file

@ -31,12 +31,12 @@ public class HttpBulkClient extends AbstractBulkClient implements ElasticsearchC
@Override
protected ElasticsearchClient createClient(Settings settings) throws IOException {
return helper.createClient(settings);
return this;
}
@Override
protected void closeClient() throws IOException {
helper.closeClient();
protected void closeClient(Settings settings) throws IOException {
helper.closeClient(settings);
}
@Override
@ -50,8 +50,10 @@ public class HttpBulkClient extends AbstractBulkClient implements ElasticsearchC
}
@Override
public <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> RequestBuilder prepareExecute(Action<Request, Response, RequestBuilder> action) {
return helper.prepareExecute(action);
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

View file

@ -11,11 +11,9 @@ import org.elasticsearch.action.ActionRequestBuilder;
import org.elasticsearch.action.ActionResponse;
import org.elasticsearch.action.GenericAction;
import org.elasticsearch.action.support.PlainActionFuture;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
import org.elasticsearch.threadpool.ThreadPool;
import org.xbib.elx.common.AbstractAdminClient;
import org.xbib.net.URL;
import org.xbib.netty.http.client.Client;
import java.io.IOException;
@ -32,7 +30,7 @@ import java.util.stream.Stream;
/**
* Elasticsearch HTTP client.
*/
public class HttpClientHelper extends AbstractAdminClient implements ElasticsearchClient {
public class HttpClientHelper {
private static final Logger logger = LogManager.getLogger(HttpClientHelper.class);
@ -58,10 +56,8 @@ public class HttpClientHelper extends AbstractAdminClient implements Elasticsear
this.actionMap = new HashMap<>();
}
@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public void init(Settings settings) throws IOException {
super.init(settings);
public void init(Settings settings) {
if (settings.hasValue("url")) {
this.url = settings.get("url");
} else if (settings.hasValue("host")) {
@ -96,22 +92,10 @@ public class HttpClientHelper extends AbstractAdminClient implements Elasticsear
return nettyHttpClient;
}
@Override
public ElasticsearchClient getClient() {
return this;
}
@Override
protected ElasticsearchClient createClient(Settings settings) {
return this;
}
@Override
protected void closeClient() throws IOException {
protected void closeClient(Settings settings) throws IOException {
nettyHttpClient.shutdownGracefully();
}
@Override
public <Request extends ActionRequest, Response extends ActionResponse,
RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> ActionFuture<Response>
execute(Action<Request, Response, RequestBuilder> action, Request request) {
@ -120,21 +104,12 @@ public class HttpClientHelper extends AbstractAdminClient implements Elasticsear
return actionFuture;
}
@Override
public <Request extends ActionRequest, Response extends ActionResponse,
RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> void
execute(Action<Request, Response, RequestBuilder> action, Request request, ActionListener<Response> 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
public ThreadPool threadPool() {
logger.log(Level.TRACE, "returning null for threadPool() request");
return null;

View file

@ -30,13 +30,13 @@ public class HttpSearchClient extends AbstractSearchClient implements Elasticsea
}
@Override
protected ElasticsearchClient createClient(Settings settings) throws IOException {
return helper.createClient(settings);
protected ElasticsearchClient createClient(Settings settings) {
return this;
}
@Override
protected void closeClient() throws IOException {
helper.closeClient();
protected void closeClient(Settings settings) throws IOException {
helper.closeClient(settings);
}
@Override
@ -51,7 +51,7 @@ public class HttpSearchClient extends AbstractSearchClient implements Elasticsea
@Override
public <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> RequestBuilder prepareExecute(Action<Request, Response, RequestBuilder> action) {
return helper.prepareExecute(action);
return action.newRequestBuilder(this);
}
@Override

View file

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

View file

@ -1,39 +1,17 @@
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.NodesInfoRequest;
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.transport.TransportAddress;
import org.elasticsearch.common.unit.ByteSizeValue;
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.HttpActionContext;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
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.List;
import java.util.Map;
import java.util.Set;
/**
*
@ -95,84 +73,7 @@ public class HttpNodesInfoAction extends HttpAction<NodesInfoRequest, NodesInfoR
}
@Override
protected CheckedFunction<XContentParser, NodesInfoResponse, IOException> entityParser() {
protected CheckedFunction<XContentParser, NodesInfoResponse, IOException> entityParser(HttpResponse httpResponse) {
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);
}
}

View file

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

View file

@ -1,7 +1,6 @@
package org.xbib.elx.http.action.admin.cluster.state;
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.ClusterStateRequest;
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.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException;
import java.util.ArrayList;
@ -73,15 +73,10 @@ public class HttpClusterStateAction extends HttpAction<ClusterStateRequest, Clus
}
@Override
protected CheckedFunction<XContentParser, ClusterStateResponse, IOException> entityParser() {
protected CheckedFunction<XContentParser, ClusterStateResponse, IOException> entityParser(HttpResponse httpResponse) {
return this::fromXContent;
}
@Override
protected ClusterStateResponse emptyResponse() {
return new ClusterStateResponse();
}
private ClusterStateResponse fromXContent(XContentParser parser) throws IOException {
XContentParser.Token token = parser.nextToken();
ensureExpectedToken(XContentParser.Token.START_OBJECT, token, parser::getTokenLocation);

View file

@ -12,6 +12,7 @@ import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException;
@ -37,12 +38,7 @@ public class HttpIndicesAliasesAction extends HttpAction<IndicesAliasesRequest,
}
@Override
protected CheckedFunction<XContentParser, IndicesAliasesResponse, IOException> entityParser() {
protected CheckedFunction<XContentParser, IndicesAliasesResponse, IOException> entityParser(HttpResponse httpResponse) {
return IndicesAliasesResponse::fromXContent;
}
@Override
protected IndicesAliasesResponse emptyResponse() {
return new IndicesAliasesResponse();
}
}

View file

@ -10,6 +10,7 @@ import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException;
import java.util.ArrayList;
@ -34,16 +35,10 @@ public class HttpGetAliasAction extends HttpAction<GetAliasesRequest, GetAliases
}
@Override
protected CheckedFunction<XContentParser, GetAliasesResponse, IOException> entityParser() {
protected CheckedFunction<XContentParser, GetAliasesResponse, IOException> entityParser(HttpResponse httpResponse) {
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 {
if (parser.currentToken() == null) {
parser.nextToken();

View file

@ -11,6 +11,7 @@ import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException;
@ -29,12 +30,7 @@ public class HttpCreateIndexAction extends HttpAction<CreateIndexRequest, Create
}
@Override
protected CheckedFunction<XContentParser, CreateIndexResponse, IOException> entityParser() {
protected CheckedFunction<XContentParser, CreateIndexResponse, IOException> entityParser(HttpResponse httpResponse) {
return CreateIndexResponse::fromXContent;
}
@Override
protected CreateIndexResponse emptyResponse() {
return new CreateIndexResponse();
}
}

View file

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

View file

@ -10,6 +10,7 @@ import org.elasticsearch.rest.RestStatus;
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.common.HttpResponse;
import java.io.IOException;
@ -26,22 +27,21 @@ public class HttpIndicesExistsAction extends HttpAction<IndicesExistsRequest, In
return newHeadRequest(url, index);
}
@Override
protected CheckedFunction<XContentParser, IndicesExistsResponse, IOException> entityParser() {
return this::fromXContent;
}
@Override
protected IndicesExistsResponse emptyResponse() {
return new IndicesExistsResponse(false); // used for 404 Not found
}
@Override
protected ElasticsearchStatusException parseToError(HttpActionContext<IndicesExistsRequest, IndicesExistsResponse> httpActionContext) {
return new ElasticsearchStatusException("not found", RestStatus.NOT_FOUND);
}
private IndicesExistsResponse fromXContent(XContentParser parser) throws IOException {
return new IndicesExistsResponse(true); // used for 200 OK
@Override
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);
}
}

View file

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

View file

@ -10,6 +10,7 @@ import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentParserUtils;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException;
import java.util.HashMap;
@ -30,16 +31,10 @@ public class HttpGetSettingsAction extends HttpAction<GetSettingsRequest, GetSet
}
@Override
protected CheckedFunction<XContentParser, GetSettingsResponse, IOException> entityParser() {
protected CheckedFunction<XContentParser, GetSettingsResponse, IOException> entityParser(HttpResponse httpResponse) {
return this::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<>();

View file

@ -11,6 +11,7 @@ import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException;
import java.io.UncheckedIOException;
@ -37,12 +38,7 @@ public class HttpUpdateSettingsAction extends HttpAction<UpdateSettingsRequest,
}
@Override
protected CheckedFunction<XContentParser, UpdateSettingsResponse, IOException> entityParser() {
protected CheckedFunction<XContentParser, UpdateSettingsResponse, IOException> entityParser(HttpResponse httpResponse) {
return UpdateSettingsResponse::fromXContent;
}
@Override
protected UpdateSettingsResponse emptyResponse() {
return new UpdateSettingsResponse();
}
}

View file

@ -2,7 +2,6 @@ package org.xbib.elx.http.action.bulk;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.bulk.BulkAction;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
@ -14,6 +13,7 @@ import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException;
@ -32,16 +32,13 @@ public class HttpBulkAction extends HttpAction<BulkRequest, BulkResponse> {
IndexRequest indexRequest = (IndexRequest) actionRequest;
bulkContent.append("{\"").append(indexRequest.opType().getLowercase()).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) {
bulkContent.append(",\"_id\":\"").append(indexRequest.id()).append("\"");
}
if (indexRequest.routing() != null) {
bulkContent.append(",\"_routing\":\"").append(indexRequest.routing()).append("\"");
}
if (indexRequest.parent() != null) {
bulkContent.append(",\"_parent\":\"").append(indexRequest.parent()).append("\"");
}
if (indexRequest.version() > 0) {
bulkContent.append(",\"_version\":\"").append(indexRequest.version()).append("\"");
if (indexRequest.versionType() != null) {
@ -92,14 +89,7 @@ public class HttpBulkAction extends HttpAction<BulkRequest, BulkResponse> {
}
@Override
protected CheckedFunction<XContentParser, BulkResponse, IOException> entityParser() {
protected CheckedFunction<XContentParser, BulkResponse, IOException> entityParser(HttpResponse httpResponse) {
return BulkResponse::fromXContent;
}
@Override
protected BulkResponse emptyResponse() {
BulkItemResponse[] responses = null;
long took = 0L;
return new BulkResponse(responses, took);
}
}

View file

@ -8,6 +8,7 @@ import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException;
@ -20,16 +21,11 @@ public class HttpExistsAction extends HttpAction<GetRequest, GetResponse> {
@Override
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
protected CheckedFunction<XContentParser, GetResponse, IOException> entityParser() {
protected CheckedFunction<XContentParser, GetResponse, IOException> entityParser(HttpResponse httpResponse) {
return GetResponse::fromXContent;
}
@Override
protected GetResponse emptyResponse() {
return new GetResponse();
}
}

View file

@ -8,6 +8,7 @@ import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException;
@ -20,16 +21,11 @@ public class HttpGetAction extends HttpAction<GetRequest, GetResponse> {
@Override
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
protected CheckedFunction<XContentParser, GetResponse, IOException> entityParser() {
protected CheckedFunction<XContentParser, GetResponse, IOException> entityParser(HttpResponse httpResponse) {
return GetResponse::fromXContent;
}
@Override
protected GetResponse emptyResponse() {
return new GetResponse();
}
}

View file

@ -8,6 +8,7 @@ import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException;
@ -20,17 +21,12 @@ public class HttpIndexAction extends HttpAction<IndexRequest, IndexResponse> {
@Override
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());
}
@Override
protected CheckedFunction<XContentParser, IndexResponse, IOException> entityParser() {
protected CheckedFunction<XContentParser, IndexResponse, IOException> entityParser(HttpResponse httpResponse) {
return IndexResponse::fromXContent;
}
@Override
protected IndexResponse emptyResponse() {
return new IndexResponse();
}
}

View file

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

View file

@ -11,6 +11,7 @@ import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException;
public class HttpClearScrollAction extends HttpAction<ClearScrollRequest, ClearScrollResponse> {
@ -28,12 +29,7 @@ public class HttpClearScrollAction extends HttpAction<ClearScrollRequest, ClearS
}
@Override
protected CheckedFunction<XContentParser, ClearScrollResponse, IOException> entityParser() {
protected CheckedFunction<XContentParser, ClearScrollResponse, IOException> entityParser(HttpResponse httpResponse) {
return ClearScrollResponse::fromXContent;
}
@Override
protected ClearScrollResponse emptyResponse() {
return new ClearScrollResponse(true, 0);
}
}

View file

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

View file

@ -11,6 +11,7 @@ import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.common.HttpResponse;
import java.io.IOException;
public class HttpSearchScrollAction extends HttpAction<SearchScrollRequest, SearchResponse> {
@ -28,12 +29,7 @@ public class HttpSearchScrollAction extends HttpAction<SearchScrollRequest, Sear
}
@Override
protected SearchResponse emptyResponse() {
return new SearchResponse();
}
@Override
protected CheckedFunction<XContentParser, SearchResponse, IOException> entityParser() {
protected CheckedFunction<XContentParser, SearchResponse, IOException> entityParser(HttpResponse httpResponse) {
return SearchResponse::fromXContent;
}
}

View file

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

View file

@ -53,7 +53,7 @@ class BulkClientTest {
client.flush();
client.waitForResponses(30L, TimeUnit.SECONDS);
} finally {
assertEquals(1, client.getBulkMetric().getSucceeded().getCount());
assertEquals(1, client.getBulkController().getBulkMetric().getSucceeded().getCount());
if (client.getBulkController().getLastBulkError() != null) {
logger.error("error", client.getBulkController().getLastBulkError());
}
@ -85,7 +85,7 @@ class BulkClientTest {
.build()) {
XContentBuilder builder = JsonXContent.contentBuilder()
.startObject()
.startObject("doc")
.startObject("_doc")
.startObject("properties")
.startObject("location")
.field("type", "geo_point")
@ -94,7 +94,7 @@ class BulkClientTest {
.endObject()
.endObject();
bulkClient.newIndex("test", Settings.EMPTY, builder);
assertTrue(adminClient.getMapping("test", "doc").containsKey("properties"));
assertTrue(adminClient.getMapping("test", "_doc").containsKey("properties"));
}
}
@ -115,7 +115,7 @@ class BulkClientTest {
client.flush();
client.waitForResponses(30L, TimeUnit.SECONDS);
} finally {
assertEquals(numactions, client.getBulkMetric().getSucceeded().getCount());
assertEquals(numactions, client.getBulkController().getBulkMetric().getSucceeded().getCount());
if (client.getBulkController().getLastBulkError() != null) {
logger.error("error", client.getBulkController().getLastBulkError());
}
@ -170,7 +170,7 @@ class BulkClientTest {
logger.warn("latch timeout");
}
bulkClient.stopBulk("test", 30L, TimeUnit.SECONDS);
assertEquals(maxthreads * actions, bulkClient.getBulkMetric().getSucceeded().getCount());
assertEquals(maxthreads * actions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
} finally {
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());

View file

@ -51,7 +51,7 @@ class DuplicateIDTest {
assertTrue(hits < ACTIONS);
} finally {
client.close();
assertEquals(numactions, client.getBulkMetric().getSucceeded().getCount());
assertEquals(numactions, client.getBulkController().getBulkMetric().getSucceeded().getCount());
if (client.getBulkController().getLastBulkError() != null) {
logger.error("error", client.getBulkController().getLastBulkError());
}

View file

@ -53,7 +53,7 @@ class SearchTest {
bulkClient.refreshIndex("test");
assertEquals(numactions, bulkClient.getSearchableDocs("test"));
}
assertEquals(numactions, bulkClient.getBulkMetric().getSucceeded().getCount());
assertEquals(numactions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}
@ -77,6 +77,9 @@ class SearchTest {
idcount.incrementAndGet();
});
assertEquals(numactions, idcount.get());
assertEquals(13, searchClient.getSearchMetric().getQueries().getCount());
assertEquals(2, searchClient.getSearchMetric().getSucceededQueries().getCount());
assertEquals(0, searchClient.getSearchMetric().getEmptyQueries().getCount());
}
}
}

View file

@ -63,8 +63,8 @@ class SmokeTest {
adminClient.updateReplicaLevel(indexDefinition, 2);
int replica = adminClient.getReplicaLevel(indexDefinition);
assertEquals(2, replica);
assertEquals(0, bulkClient.getBulkMetric().getFailed().getCount());
assertEquals(6, bulkClient.getBulkMetric().getSucceeded().getCount());
assertEquals(0, bulkClient.getBulkController().getBulkMetric().getFailed().getCount());
assertEquals(6, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}

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.ClusterStateRequest;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.analysis.common.CommonAnalysisPlugin;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.client.support.AbstractClient;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
@ -38,6 +37,7 @@ import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@ -158,7 +158,7 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
return helper;
}
class Helper {
static class Helper {
String home;
@ -228,7 +228,7 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
.put(getNodeSettings())
.put("node.name", id)
.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);
AbstractClient client = (AbstractClient) node.client();
nodes.put(id, node);

View file

@ -1,5 +1,5 @@
dependencies {
api project(':elx-common')
implementation "org.xbib.elasticsearch:transport-netty4:${rootProject.property('elasticsearch-server.version')}"
testImplementation "org.xbib.elasticsearch:elasticsearch-analysis-common:${rootProject.property('elasticsearch-server.version')}"
//testImplementation "org.xbib.elasticsearch:elasticsearch-analysis-common:${rootProject.property('elasticsearch-server.version')}"
}

View file

@ -4,6 +4,7 @@ module org.xbib.elx.node {
requires org.xbib.elx.common;
requires org.xbib.elasticsearch.log4j;
requires org.xbib.elasticsearch.server;
requires org.xbib.elasticsearch.transport.nettyfour;
provides org.xbib.elx.api.AdminClientProvider with
org.xbib.elx.node.NodeAdminClientProvider;
provides org.xbib.elx.api.BulkClientProvider with

View file

@ -1,69 +1,25 @@
package org.xbib.elx.node;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.env.Environment;
import org.elasticsearch.node.Node;
import org.elasticsearch.plugins.Plugin;
import org.xbib.elx.common.AbstractAdminClient;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
public class NodeAdminClient extends AbstractAdminClient {
private static final Logger logger = LogManager.getLogger(NodeAdminClient.class.getName());
private final NodeClientHelper helper;
private Node node;
@Override
protected ElasticsearchClient createClient(Settings settings) throws IOException {
if (settings == null) {
return null;
}
String version = System.getProperty("os.name")
+ " " + System.getProperty("java.vm.name")
+ " " + System.getProperty("java.vm.vendor")
+ " " + System.getProperty("java.runtime.version")
+ " " + System.getProperty("java.vm.version");
Settings effectiveSettings = Settings.builder().put(settings)
.put("node.client", true)
.put("node.master", false)
.put("node.data", false)
.build();
XContentBuilder builder = XContentFactory.jsonBuilder();
effectiveSettings.toXContent(builder, new ToXContent.MapParams(Collections.singletonMap("flat_settings", "true")));
logger.info("creating node client on {} with effective settings {}",
version, Strings.toString(builder));
Collection<Class<? extends Plugin>> plugins = Collections.emptyList();
this.node = new BulkNode(new Environment(effectiveSettings, null), plugins);
try {
node.start();
} catch (Exception e) {
throw new IOException(e);
}
return node.client();
public NodeAdminClient() {
this.helper = new NodeClientHelper();
}
@Override
protected void closeClient() throws IOException {
if (node != null) {
logger.debug("closing node client");
node.close();
}
protected ElasticsearchClient createClient(Settings settings) {
return helper.createClient(settings, null);
}
private static class BulkNode extends Node {
BulkNode(Environment env, Collection<Class<? extends Plugin>> classpathPlugins) {
super(env, classpathPlugins);
}
@Override
protected void closeClient(Settings settings) throws IOException {
helper.closeClient(settings);
}
}

View file

@ -1,68 +1,25 @@
package org.xbib.elx.node;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.env.Environment;
import org.elasticsearch.node.Node;
import org.elasticsearch.plugins.Plugin;
import org.xbib.elx.common.AbstractBulkClient;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
public class NodeBulkClient extends AbstractBulkClient {
private static final Logger logger = LogManager.getLogger(NodeBulkClient.class.getName());
private final NodeClientHelper helper;
private Node node;
@Override
protected ElasticsearchClient createClient(Settings settings) throws IOException {
if (settings == null) {
return null;
}
String version = System.getProperty("os.name")
+ " " + System.getProperty("java.vm.name")
+ " " + System.getProperty("java.vm.vendor")
+ " " + System.getProperty("java.runtime.version")
+ " " + System.getProperty("java.vm.version");
Settings effectiveSettings = Settings.builder().put(settings)
.put("node.client", true)
.put("node.master", false)
.put("node.data", false)
.build();
XContentBuilder builder = XContentFactory.jsonBuilder();
effectiveSettings.toXContent(builder, new ToXContent.MapParams(Collections.singletonMap("flat_settings", "true")));
logger.info("creating node client on {} with effective settings {}",
version, Strings.toString(builder));
Collection<Class<? extends Plugin>> plugins = Collections.emptyList();
this.node = new BulkNode(new Environment(effectiveSettings, null), plugins);
try {
node.start();
} catch (Exception e) {
throw new IOException(e);
}
return node.client();
public NodeBulkClient() {
this.helper = new NodeClientHelper();
}
@Override
protected void closeClient() throws IOException {
if (node != null) {
logger.debug("closing node client");
node.close();
}
protected ElasticsearchClient createClient(Settings settings) {
return helper.createClient(settings, null);
}
private static class BulkNode extends Node {
BulkNode(Environment env, Collection<Class<? extends Plugin>> classpathPlugins) {
super(env, classpathPlugins);
}
@Override
protected void closeClient(Settings settings) throws IOException {
helper.closeClient(settings);
}
}

View file

@ -0,0 +1,97 @@
package org.xbib.elx.node;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.network.NetworkModule;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.concurrent.EsExecutors;
import org.elasticsearch.env.Environment;
import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeValidationException;
import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.transport.netty4.Netty4Plugin;
import org.xbib.elx.common.Parameters;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
public class NodeClientHelper {
private static final Logger logger = LogManager.getLogger(NodeClientHelper.class.getName());
private static Object configurationObject;
private static Node node;
private static final Map<String, ElasticsearchClient> clientMap = new HashMap<>();
public ElasticsearchClient createClient(Settings settings, Object object) {
if (configurationObject == null) {
configurationObject = object;
}
if (configurationObject instanceof ElasticsearchClient) {
return (ElasticsearchClient) configurationObject;
}
return clientMap.computeIfAbsent(settings.get("cluster.name"),
key -> innerCreateClient(settings));
}
public void closeClient(Settings settings) throws IOException {
ElasticsearchClient client = clientMap.remove(settings.get("cluster.name"));
if (client != null) {
logger.debug("closing node...");
node.close();
node = null;
}
}
private ElasticsearchClient innerCreateClient(Settings settings) {
String version = System.getProperty("os.name")
+ " " + System.getProperty("java.vm.name")
+ " " + System.getProperty("java.vm.vendor")
+ " " + System.getProperty("java.runtime.version")
+ " " + System.getProperty("java.vm.version");
Settings effectiveSettings = Settings.builder()
.put(settings.filter(key -> !isPrivateSettings(key)))
.put("node.master", false)
.put("node.data", false)
// "node.processors"
.put(EsExecutors.PROCESSORS_SETTING.getKey(),
settings.get(EsExecutors.PROCESSORS_SETTING.getKey(),
String.valueOf(Runtime.getRuntime().availableProcessors())))
// "transport.type"
.put(NetworkModule.TRANSPORT_TYPE_KEY,
Netty4Plugin.NETTY_TRANSPORT_NAME)
.build();
logger.info("creating node client on {} with effective settings {}",
version, effectiveSettings.toDelimitedString(','));
Collection<Class<? extends Plugin>> plugins =
Collections.singletonList(Netty4Plugin.class);
node = new BulkNode(new Environment(effectiveSettings, null), plugins);
try {
node.start();
return node.client();
} catch (NodeValidationException e) {
logger.log(Level.ERROR, e.getMessage(), e);
}
return null;
}
private static boolean isPrivateSettings(String key) {
return key.equals(Parameters.MAX_ACTIONS_PER_REQUEST.name()) ||
key.equals(Parameters.MAX_CONCURRENT_REQUESTS.name()) ||
key.equals(Parameters.MAX_VOLUME_PER_REQUEST.name()) ||
key.equals(Parameters.FLUSH_INTERVAL.name());
}
private static class BulkNode extends Node {
BulkNode(Environment env, Collection<Class<? extends Plugin>> classpathPlugins) {
super(env, classpathPlugins);
}
}
}

View file

@ -1,68 +1,25 @@
package org.xbib.elx.node;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.env.Environment;
import org.elasticsearch.node.Node;
import org.elasticsearch.plugins.Plugin;
import org.xbib.elx.common.AbstractSearchClient;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
public class NodeSearchClient extends AbstractSearchClient {
private static final Logger logger = LogManager.getLogger(NodeSearchClient.class.getName());
private final NodeClientHelper helper;
private Node node;
@Override
protected ElasticsearchClient createClient(Settings settings) throws IOException {
if (settings == null) {
return null;
}
String version = System.getProperty("os.name")
+ " " + System.getProperty("java.vm.name")
+ " " + System.getProperty("java.vm.vendor")
+ " " + System.getProperty("java.runtime.version")
+ " " + System.getProperty("java.vm.version");
Settings effectiveSettings = Settings.builder().put(settings)
.put("node.client", true)
.put("node.master", false)
.put("node.data", false)
.build();
XContentBuilder builder = XContentFactory.jsonBuilder();
effectiveSettings.toXContent(builder, new ToXContent.MapParams(Collections.singletonMap("flat_settings", "true")));
logger.info("creating node client on {} with effective settings {}",
version, Strings.toString(builder));
Collection<Class<? extends Plugin>> plugins = Collections.emptyList();
this.node = new BulkNode(new Environment(effectiveSettings, null), plugins);
try {
node.start();
} catch (Exception e) {
throw new IOException(e);
}
return node.client();
public NodeSearchClient() {
this.helper = new NodeClientHelper();
}
@Override
protected void closeClient() throws IOException {
if (node != null) {
logger.debug("closing node client");
node.close();
}
protected ElasticsearchClient createClient(Settings settings) {
return helper.createClient(settings, null);
}
private static class BulkNode extends Node {
BulkNode(Environment env, Collection<Class<? extends Plugin>> classpathPlugins) {
super(env, classpathPlugins);
}
@Override
protected void closeClient(Settings settings) throws IOException {
helper.closeClient(settings);
}
}

View file

@ -43,6 +43,7 @@ class BulkClientTest {
void testSingleDoc() throws Exception {
final NodeBulkClient bulkClient = ClientBuilder.builder(helper.client("1"))
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(30))
.build();
@ -52,7 +53,7 @@ class BulkClientTest {
bulkClient.flush();
bulkClient.waitForResponses(30L, TimeUnit.SECONDS);
} finally {
assertEquals(1, bulkClient.getBulkMetric().getSucceeded().getCount());
assertEquals(1, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}
@ -65,6 +66,7 @@ class BulkClientTest {
void testNewIndex() throws Exception {
final NodeBulkClient bulkClient = ClientBuilder.builder(helper.client("1"))
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(5))
.build();
bulkClient.newIndex("test");
@ -75,22 +77,22 @@ class BulkClientTest {
void testMapping() throws Exception {
try (NodeAdminClient adminClient = ClientBuilder.builder(helper.client("1"))
.setAdminClientProvider(NodeAdminClientProvider.class)
.put(helper.getNodeSettings())
.build();
NodeBulkClient bulkClient = ClientBuilder.builder(helper.client("1"))
.setBulkClientProvider(NodeBulkClientProvider.class)
.build()) {
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.build()) {
XContentBuilder builder = JsonXContent.contentBuilder()
.startObject()
.startObject("doc")
.startObject("properties")
.startObject("location")
.field("type", "geo_point")
.endObject()
.endObject()
.endObject()
.endObject();
bulkClient.newIndex("test", Settings.EMPTY, builder);
assertTrue(adminClient.getMapping("test", "doc").containsKey("properties"));
assertTrue(adminClient.getMapping("test", "_doc").containsKey("properties"));
}
}
@ -99,6 +101,7 @@ class BulkClientTest {
long numactions = ACTIONS;
final NodeBulkClient bulkClient = ClientBuilder.builder(helper.client("1"))
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(60))
.build();
@ -111,7 +114,7 @@ class BulkClientTest {
bulkClient.flush();
bulkClient.waitForResponses(30L, TimeUnit.SECONDS);
} finally {
assertEquals(numactions, bulkClient.getBulkMetric().getSucceeded().getCount());
assertEquals(numactions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}
@ -130,6 +133,7 @@ class BulkClientTest {
logger.info("maxthreads={} maxactions={} maxloop={}", maxthreads, maxActionsPerRequest, actions);
final NodeBulkClient bulkClient = ClientBuilder.builder(helper.client("1"))
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.put(Parameters.MAX_CONCURRENT_REQUESTS.name(), maxthreads)
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), maxActionsPerRequest)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(60))
@ -165,14 +169,14 @@ class BulkClientTest {
logger.warn("latch timeout");
}
bulkClient.stopBulk("test", 30L, TimeUnit.SECONDS);
assertEquals(maxthreads * actions, bulkClient.getBulkMetric().getSucceeded().getCount());
assertEquals(maxthreads * actions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
bulkClient.refreshIndex("test");
assertEquals(maxthreads * actions, bulkClient.getSearchableDocs("test"));
} finally {
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}
assertNull(bulkClient.getBulkController().getLastBulkError());
bulkClient.refreshIndex("test");
assertEquals(maxthreads * actions, bulkClient.getSearchableDocs("test"));
bulkClient.close();
}
}

View file

@ -35,6 +35,7 @@ class DuplicateIDTest {
long numactions = ACTIONS;
final NodeBulkClient bulkClient = ClientBuilder.builder(helper.client("1"))
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), MAX_ACTIONS_PER_REQUEST)
.build();
try {
@ -49,7 +50,7 @@ class DuplicateIDTest {
assertTrue(bulkClient.getSearchableDocs("test") < ACTIONS);
} finally {
bulkClient.close();
assertEquals(numactions, bulkClient.getBulkMetric().getSucceeded().getCount());
assertEquals(numactions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}

View file

@ -37,9 +37,11 @@ class IndexPruneTest {
void testPrune() throws IOException {
final NodeAdminClient adminClient = ClientBuilder.builder(helper.client("1"))
.setAdminClientProvider(NodeAdminClientProvider.class)
.put(helper.getNodeSettings())
.build();
final NodeBulkClient bulkClient = ClientBuilder.builder(helper.client("1"))
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.build();
try {
Settings settings = Settings.builder()

View file

@ -36,9 +36,11 @@ class IndexShiftTest {
void testIndexShift() throws Exception {
final NodeAdminClient adminClient = ClientBuilder.builder(helper.client("1"))
.setAdminClientProvider(NodeAdminClientProvider.class)
.put(helper.getNodeSettings())
.build();
final NodeBulkClient bulkClient = ClientBuilder.builder(helper.client("1"))
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.build();
try {
Settings settings = Settings.builder()

View file

@ -39,6 +39,7 @@ class SearchTest {
long numactions = ACTIONS;
final NodeBulkClient bulkClient = ClientBuilder.builder(helper.client("1"))
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), MAX_ACTIONS_PER_REQUEST)
.build();
try (bulkClient) {
@ -52,13 +53,14 @@ class SearchTest {
bulkClient.refreshIndex("test");
assertEquals(numactions, bulkClient.getSearchableDocs("test"));
}
assertEquals(numactions, bulkClient.getBulkMetric().getSucceeded().getCount());
assertEquals(numactions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}
assertNull(bulkClient.getBulkController().getLastBulkError());
try (NodeSearchClient searchClient = ClientBuilder.builder(helper.client("1"))
.setSearchClientProvider(NodeSearchClientProvider.class)
.put(helper.getNodeSettings())
.build()) {
Stream<SearchHit> stream = searchClient.search(qb -> qb
.setIndices("test")
@ -75,6 +77,9 @@ class SearchTest {
idcount.incrementAndGet();
});
assertEquals(numactions, idcount.get());
assertEquals(13, searchClient.getSearchMetric().getQueries().getCount());
assertEquals(2, searchClient.getSearchMetric().getSucceededQueries().getCount());
assertEquals(0, searchClient.getSearchMetric().getEmptyQueries().getCount());
}
}
}

View file

@ -32,9 +32,11 @@ class SmokeTest {
void smokeTest() throws Exception {
try (NodeAdminClient adminClient = ClientBuilder.builder(helper.client("1"))
.setAdminClientProvider(NodeAdminClientProvider.class)
.put(helper.getNodeSettings())
.build();
NodeBulkClient bulkClient = ClientBuilder.builder(helper.client("1"))
.setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings())
.build()) {
IndexDefinition indexDefinition =
adminClient.buildIndexDefinitionFromSettings("test_smoke", Settings.EMPTY);
@ -61,8 +63,8 @@ class SmokeTest {
adminClient.updateReplicaLevel(indexDefinition, 2);
int replica = adminClient.getReplicaLevel(indexDefinition);
assertEquals(2, replica);
assertEquals(0, bulkClient.getBulkMetric().getFailed().getCount());
assertEquals(6, bulkClient.getBulkMetric().getSucceeded().getCount());
assertEquals(0, bulkClient.getBulkController().getBulkMetric().getFailed().getCount());
assertEquals(6, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}

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.ClusterStateRequest;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.analysis.common.CommonAnalysisPlugin;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.client.support.AbstractClient;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
@ -38,6 +37,7 @@ import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@ -194,6 +194,7 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
.put("path.home", getHome())
.put("discovery.zen.master_election.ignore_non_master_pings", "true")
.put("transport.netty.epoll", "false")
.put("node.max_local_storage_nodes", 2) // for server and client
.build();
}
@ -219,7 +220,7 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
.put(getNodeSettings())
.put("node.name", id)
.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);
AbstractClient client = (AbstractClient) node.client();
nodes.put(id, node);

View file

@ -1,5 +1,5 @@
dependencies {
api project(':elx-common')
implementation "org.xbib.elasticsearch:transport-netty4:${rootProject.property('elasticsearch-server.version')}"
testImplementation "org.xbib.elasticsearch:elasticsearch-analysis-common:${rootProject.property('elasticsearch-server.version')}"
//testImplementation "org.xbib.elasticsearch:elasticsearch-analysis-common:${rootProject.property('elasticsearch-server.version')}"
}

View file

@ -19,17 +19,13 @@ public class TransportAdminClient extends AbstractAdminClient {
}
@Override
protected ElasticsearchClient createClient(Settings settings) throws IOException {
public ElasticsearchClient createClient(Settings settings) throws IOException {
return helper.createClient(settings);
}
@Override
protected void closeClient() {
if (getClient() != null) {
TransportClient client = (TransportClient) getClient();
client.close();
client.threadPool().shutdown();
}
public void closeClient(Settings settings) {
helper.closeClient(settings);
}
@Override

View file

@ -18,22 +18,18 @@ public class TransportBulkClient extends AbstractBulkClient {
}
@Override
protected ElasticsearchClient createClient(Settings settings) throws IOException {
public ElasticsearchClient createClient(Settings settings) throws IOException {
return helper.createClient(settings);
}
@Override
protected void closeClient() {
if (getClient() != null) {
TransportClient client = (TransportClient) getClient();
client.close();
client.threadPool().shutdown();
}
}
@Override
public void init(Settings settings) throws IOException {
super.init(settings);
helper.init((TransportClient) getClient(), settings);
}
@Override
protected void closeClient(Settings settings) {
helper.closeClient(settings);
}
}

View file

@ -7,6 +7,7 @@ import org.elasticsearch.Version;
import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
import org.elasticsearch.action.admin.cluster.state.ClusterStateRequestBuilder;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.client.transport.NoNodeAvailableException;
import org.elasticsearch.client.transport.TransportClient;
@ -24,17 +25,24 @@ import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.transport.netty4.Netty4Plugin;
import org.xbib.elx.common.util.NetworkUtils;
import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* Transport client with additional methods using the BulkProcessor.
*/
public class TransportClientHelper {
private static final Logger logger = LogManager.getLogger(TransportClientHelper.class.getName());
private static final Logger logger = LogManager.getLogger(TransportAdminClient.class.getName());
private static final Map<String, ElasticsearchClient> clientMap = new HashMap<>();
protected ElasticsearchClient createClient(Settings settings) throws IOException {
if (settings != null) {
@ -54,6 +62,16 @@ public class TransportClientHelper {
return null;
}
public void closeClient(Settings settings) {
ElasticsearchClient client = clientMap.remove(settings.get("cluster.name"));
if (client != null) {
if (client instanceof Client) {
((Client) client).close();
}
client.threadPool().shutdownNow();
}
}
public void init(TransportClient transportClient, Settings settings) throws IOException {
Collection<TransportAddress> addrs = findAddresses(settings);
if (!connect(transportClient, addrs, settings.getAsBoolean("autodiscover", false))) {
@ -96,13 +114,15 @@ public class TransportClientHelper {
logger.info("connected to nodes = {}", nodes);
if (nodes != null && !nodes.isEmpty()) {
if (autodiscover) {
logger.debug("trying to auto-discover all nodes...");
logger.debug("trying to discover all nodes...");
ClusterStateRequestBuilder clusterStateRequestBuilder =
new ClusterStateRequestBuilder(transportClient, ClusterStateAction.INSTANCE);
ClusterStateResponse clusterStateResponse = clusterStateRequestBuilder.execute().actionGet();
DiscoveryNodes discoveryNodes = clusterStateResponse.getState().getNodes();
addDiscoveryNodes(transportClient, discoveryNodes);
logger.info("after auto-discovery: connected to {}", transportClient.connectedNodes());
for (DiscoveryNode discoveryNode : discoveryNodes) {
transportClient.addTransportAddress(discoveryNode.getAddress());
}
logger.info("after discovery: connected to {}", transportClient.connectedNodes());
}
return true;
}
@ -114,7 +134,7 @@ public class TransportClientHelper {
// "cluster.name"
.put(ClusterName.CLUSTER_NAME_SETTING.getKey(),
settings.get(ClusterName.CLUSTER_NAME_SETTING.getKey()))
// "processors"
// "node.processors"
.put(EsExecutors.PROCESSORS_SETTING.getKey(),
settings.get(EsExecutors.PROCESSORS_SETTING.getKey(),
String.valueOf(Runtime.getRuntime().availableProcessors())))
@ -124,12 +144,6 @@ public class TransportClientHelper {
.build();
}
private void addDiscoveryNodes(TransportClient transportClient, DiscoveryNodes discoveryNodes) {
for (DiscoveryNode discoveryNode : discoveryNodes) {
transportClient.addTransportAddress(discoveryNode.getAddress());
}
}
static class MyTransportClient extends TransportClient {
MyTransportClient(Settings settings, Collection<Class<? extends Plugin>> plugins) {

View file

@ -18,22 +18,18 @@ public class TransportSearchClient extends AbstractSearchClient {
}
@Override
protected ElasticsearchClient createClient(Settings settings) throws IOException {
public ElasticsearchClient createClient(Settings settings) throws IOException {
return helper.createClient(settings);
}
@Override
protected void closeClient() {
if (getClient() != null) {
TransportClient client = (TransportClient) getClient();
client.close();
client.threadPool().shutdown();
}
}
@Override
public void init(Settings settings) throws IOException {
super.init(settings);
helper.init((TransportClient) getClient(), settings);
}
@Override
public void closeClient(Settings settings) {
helper.closeClient(settings);
}
}

View file

@ -54,7 +54,7 @@ class BulkClientTest {
bulkClient.flush();
bulkClient.waitForResponses(30L, TimeUnit.SECONDS);
} finally {
assertEquals(1, bulkClient.getBulkMetric().getSucceeded().getCount());
assertEquals(1, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}
@ -86,16 +86,14 @@ class BulkClientTest {
.build()) {
XContentBuilder builder = JsonXContent.contentBuilder()
.startObject()
.startObject("doc")
.startObject("properties")
.startObject("location")
.field("type", "geo_point")
.endObject()
.endObject()
.endObject()
.endObject();
bulkClient.newIndex("test", Settings.EMPTY, builder);
assertTrue(adminClient.getMapping("test", "doc").containsKey("properties"));
assertTrue(adminClient.getMapping("test", "_doc").containsKey("properties"));
}
}
@ -119,7 +117,7 @@ class BulkClientTest {
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
assertEquals(numactions, bulkClient.getBulkMetric().getSucceeded().getCount());
assertEquals(numactions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}
@ -176,7 +174,7 @@ class BulkClientTest {
logger.warn("latch timeout");
}
bulkClient.stopBulk("test", 30L, TimeUnit.SECONDS);
assertEquals(maxthreads * actions, bulkClient.getBulkMetric().getSucceeded().getCount());
assertEquals(maxthreads * actions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
} catch (NoNodeAvailableException e) {
logger.warn("skipping, no node available");
} catch (Exception e) {

View file

@ -50,7 +50,7 @@ class DuplicateIDTest {
assertTrue(bulkClient.getSearchableDocs("test_dup") < ACTIONS);
} finally {
bulkClient.close();
assertEquals(numactions, bulkClient.getBulkMetric().getSucceeded().getCount());
assertEquals(numactions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}

View file

@ -53,7 +53,7 @@ class SearchTest {
bulkClient.refreshIndex("test");
assertEquals(numactions, bulkClient.getSearchableDocs("test"));
}
assertEquals(numactions, bulkClient.getBulkMetric().getSucceeded().getCount());
assertEquals(numactions, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}
@ -77,6 +77,9 @@ class SearchTest {
idcount.incrementAndGet();
});
assertEquals(numactions, idcount.get());
assertEquals(13, searchClient.getSearchMetric().getQueries().getCount());
assertEquals(2, searchClient.getSearchMetric().getSucceededQueries().getCount());
assertEquals(0, searchClient.getSearchMetric().getEmptyQueries().getCount());
}
}
}

View file

@ -63,8 +63,8 @@ class SmokeTest {
adminClient.updateReplicaLevel(indexDefinition, 2);
int replica = adminClient.getReplicaLevel(indexDefinition);
assertEquals(2, replica);
assertEquals(0, bulkClient.getBulkMetric().getFailed().getCount());
assertEquals(6, bulkClient.getBulkMetric().getSucceeded().getCount());
assertEquals(0, bulkClient.getBulkController().getBulkMetric().getFailed().getCount());
assertEquals(6, bulkClient.getBulkController().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkController().getLastBulkError() != null) {
logger.error("error", bulkClient.getBulkController().getLastBulkError());
}

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.ClusterStateRequest;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.analysis.common.CommonAnalysisPlugin;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.client.support.AbstractClient;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
@ -38,6 +37,7 @@ import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@ -227,7 +227,7 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
.put(getNodeSettings())
.put("node.name", id)
.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);
AbstractClient client = (AbstractClient) node.client();
nodes.put(id, node);

View file

@ -1,6 +1,6 @@
group = org.xbib
name = elx
version = 6.3.2.7
version = 6.3.2.8
gradle.wrapper.version = 6.4.1
elasticsearch-server.version = 6.3.2.4
@ -8,6 +8,6 @@ tcnative.version = 2.0.29.Final
tcnative-legacy-macosx.version = 2.0.26.Final
bouncycastle.version = 1.64
xbib-metrics.version = 2.1.0
xbib-netty-http.version = 4.1.49.1
log4j.version = 2.13.1
xbib-netty-http.version = 4.1.50.1
log4j.version = 2.13.3
asciidoclet.version = 1.6.0.0