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

update for new 6.3.2.4 version

Browse source
This commit is contained in:
Jörg Prante 2020-04-06 11:16:33 +02:00
parent 1912c018d5
commit 7b71bf9481
54 changed files with 262 additions and 209 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -11,4 +11,5 @@
/.project
/.gradle
build
out
plugins

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

@ -29,7 +29,7 @@ public interface BulkProcessor extends Closeable, Flushable {
/**
* A listener for the execution.
*/
public interface Listener {
interface Listener {
/**
* Callback before the bulk is executed.

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

@ -2,6 +2,7 @@ package org.xbib.elx.common;
import com.carrotsearch.hppc.cursors.ObjectCursor;
import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.ElasticsearchTimeoutException;
@ -194,16 +195,17 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
@Override
public AbstractExtendedClient init(Settings settings) throws IOException {
logger.log(Level.INFO, "initializing with settings = " + settings.toDelimitedString(','));
if (client == null) {
client = createClient(settings);
}
if (bulkMetric == null) {
this.bulkMetric = new DefaultBulkMetric();
this.bulkMetric.init(settings);
bulkMetric = new DefaultBulkMetric();
bulkMetric.init(settings);
}
if (bulkController == null) {
this.bulkController = new DefaultBulkController(this, bulkMetric);
this.bulkController.init(settings);
bulkController = new DefaultBulkController(this, bulkMetric);
bulkController.init(settings);
}
return this;
}
@ -235,8 +237,7 @@ public abstract class AbstractExtendedClient implements ExtendedClient {
public String getClusterName() {
ensureActive();
try {
ClusterStateRequest clusterStateRequest = new ClusterStateRequest();
clusterStateRequest.clear();
ClusterStateRequest clusterStateRequest = new ClusterStateRequest().clear();
ClusterStateResponse clusterStateResponse =
client.execute(ClusterStateAction.INSTANCE, clusterStateRequest).actionGet();
return clusterStateResponse.getClusterName().value();

11
elx-common/src/main/java/org/xbib/elx/common/ClientBuilder.java
View file

@ -1,5 +1,8 @@
package org.xbib.elx.common;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.Version;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.settings.Settings;
@ -16,6 +19,8 @@ import java.util.ServiceLoader;
@SuppressWarnings("rawtypes")
public class ClientBuilder {
private static final Logger logger = LogManager.getLogger(ClientBuilder.class);
private final ElasticsearchClient client;
private final Settings.Builder settingsBuilder;
@ -97,6 +102,10 @@ public class ClientBuilder {
if (provider == null) {
throw new IllegalArgumentException("no provider");
}
return (C) providerMap.get(provider).getExtendedClient().setClient(client).init(settingsBuilder.build());
Settings settings = settingsBuilder.build();
logger.log(Level.INFO, "settings = " + settings.toDelimitedString(','));
return (C) providerMap.get(provider).getExtendedClient()
.setClient(client)
.init(settings);
}
}

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

@ -49,6 +49,8 @@ public class DefaultBulkController implements BulkController {
private AtomicBoolean active;
private boolean enableBulkLogging;
public DefaultBulkController(ExtendedClient client, BulkMetric bulkMetric) {
this.client = client;
this.bulkMetric = bulkMetric;
@ -76,11 +78,8 @@ 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"));
if (logger.isInfoEnabled()) {
logger.info("bulk processor up with maxActionsPerRequest = {} maxConcurrentRequests = {} " +
"flushIngestInterval = {} maxVolumePerRequest = {}",
maxActionsPerRequest, maxConcurrentRequests, flushIngestInterval, maxVolumePerRequest);
}
this.enableBulkLogging = settings.getAsBoolean(Parameters.ENABLE_BULK_LOGGING.name(),
Parameters.ENABLE_BULK_LOGGING.getValue());
this.bulkListener = new BulkListener();
this.bulkProcessor = DefaultBulkProcessor.builder(client.getClient(), bulkListener)
.setBulkActions(maxActionsPerRequest)
@ -89,6 +88,12 @@ public class DefaultBulkController implements BulkController {
.setBulkSize(maxVolumePerRequest)
.build();
this.active.set(true);
if (logger.isInfoEnabled()) {
logger.info("bulk processor up with maxActionsPerRequest = {} maxConcurrentRequests = {} " +
"flushIngestInterval = {} maxVolumePerRequest = {} bulk logging = {} logger debug = {} from settings = {}",
maxActionsPerRequest, maxConcurrentRequests, flushIngestInterval, maxVolumePerRequest,
enableBulkLogging, logger.isDebugEnabled(), settings.toDelimitedString(','));
}
}
@Override
@ -115,6 +120,9 @@ public class DefaultBulkController implements BulkController {
@Override
public void index(IndexRequest indexRequest) {
ensureActiveAndBulk();
if (!active.get()) {
throw new IllegalStateException("inactive");
}
try {
if (bulkMetric != null) {
bulkMetric.getCurrentIngest().inc(indexRequest.index(), indexRequest.type(), indexRequest.id());
@ -237,7 +245,7 @@ public class DefaultBulkController implements BulkController {
private class BulkListener implements DefaultBulkProcessor.Listener {
private final Logger logger = LogManager.getLogger("org.xbib.elx.BulkProcessor.Listener");
private final Logger logger = LogManager.getLogger(BulkListener.class.getName());
private Throwable lastBulkError = null;
@ -252,7 +260,7 @@ public class DefaultBulkController implements BulkController {
bulkMetric.getCurrentIngestNumDocs().inc(n);
bulkMetric.getTotalIngestSizeInBytes().inc(request.estimatedSizeInBytes());
}
if (logger.isDebugEnabled()) {
if (enableBulkLogging && logger.isDebugEnabled()) {
logger.debug("before bulk [{}] [actions={}] [bytes={}] [concurrent requests={}]",
executionId,
request.numberOfActions(),
@ -282,7 +290,7 @@ public class DefaultBulkController implements BulkController {
}
}
}
if (bulkMetric != null && logger.isDebugEnabled()) {
if (enableBulkLogging && logger.isDebugEnabled() && bulkMetric != null) {
logger.debug("after bulk [{}] [succeeded={}] [failed={}] [{}ms] {} concurrent requests",
executionId,
bulkMetric.getSucceeded().getCount(),
@ -291,7 +299,7 @@ public class DefaultBulkController implements BulkController {
l);
}
if (n > 0) {
if (logger.isErrorEnabled()) {
if (enableBulkLogging && logger.isErrorEnabled()) {
logger.error("bulk [{}] failed with {} failed items, failure message = {}",
executionId, n, response.buildFailureMessage());
}
@ -309,7 +317,7 @@ public class DefaultBulkController implements BulkController {
}
lastBulkError = failure;
active.set(false);
if (logger.isErrorEnabled()) {
if (enableBulkLogging && logger.isErrorEnabled()) {
logger.error("after bulk [" + executionId + "] error", failure);
}
}

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

@ -124,7 +124,7 @@ public class DefaultBulkProcessor implements BulkProcessor {
closed = true;
if (scheduledFuture != null) {
FutureUtils.cancel(scheduledFuture);
this.scheduler.shutdown();
scheduler.shutdown();
}
if (bulkRequest.numberOfActions() > 0) {
execute();

12
elx-common/src/main/java/org/xbib/elx/common/Parameters.java
View file

@ -2,6 +2,8 @@ package org.xbib.elx.common;
public enum Parameters {
ENABLE_BULK_LOGGING(false),
DEFAULT_MAX_ACTIONS_PER_REQUEST(1000),
DEFAULT_MAX_CONCURRENT_REQUESTS(Runtime.getRuntime().availableProcessors()),
@ -18,10 +20,16 @@ public enum Parameters {
FLUSH_INTERVAL("flush_interval");
boolean flag;
int num;
String string;
Parameters(boolean flag) {
this.flag = flag;
}
Parameters(int num) {
this.num = num;
}
@ -30,6 +38,10 @@ public enum Parameters {
this.string = string;
}
boolean getValue() {
return flag;
}
int getNum() {
return num;
}

2
elx-common/src/main/java/org/xbib/elx/common/io/package-info.java
View file

@ -1,4 +1,4 @@
/**
*
* I/O helpers for Elasticsearch client extensions.
*/
package org.xbib.elx.common.io;

2
elx-common/src/main/java/org/xbib/elx/common/package-info.java
View file

@ -1,4 +1,4 @@
/**
*
* Common classes for Elasticsearch client extensions.
*/
package org.xbib.elx.common;

2
elx-common/src/main/java/org/xbib/elx/common/util/package-info.java
View file

@ -1,4 +1,4 @@
/**
*
* Utilities for Elasticsearch client extensions.
*/
package org.xbib.elx.common.util;

1
elx-common/src/test/java/org/xbib/elx/common/test/AliasTest.java
View file

@ -79,7 +79,6 @@ class AliasTest {
.aliases(aliases);
indicesAliasesRequest.addAliasAction(aliasAction);
client.execute(IndicesAliasesAction.INSTANCE, indicesAliasesRequest).actionGet();
GetAliasesRequest getAliasesRequest = new GetAliasesRequest();
getAliasesRequest.aliases(alias);
GetAliasesResponse getAliasesResponse =

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

@ -158,7 +158,7 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
return helper;
}
class Helper {
static class Helper {
String home;

9
elx-http/build.gradle
View file

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

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

@ -6,7 +6,7 @@ import org.elasticsearch.common.ParseField;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
import java.util.Map;
@ -23,7 +23,7 @@ public class HttpGetMappingsAction extends HttpAction<GetMappingsRequest, GetMap
}
@Override
protected RequestBuilder createHttpRequest(String url, GetMappingsRequest request) {
protected Request.Builder createHttpRequest(String url, GetMappingsRequest request) {
String index = request.indices() != null ? "/" + String.join(",", request.indices()) : "";
return newGetRequest(url, index + "/_mapping");
}

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

@ -18,7 +18,6 @@ import org.elasticsearch.threadpool.ThreadPool;
import org.xbib.elx.common.AbstractExtendedClient;
import org.xbib.net.URL;
import org.xbib.netty.http.client.Client;
import org.xbib.netty.http.client.ClientBuilder;
import java.io.IOException;
import java.util.ArrayList;
@ -75,7 +74,7 @@ public class ExtendedHttpClient extends AbstractExtendedClient implements Elasti
httpAction.setSettings(settings);
actionMap.put(httpAction.getActionInstance(), httpAction);
}
ClientBuilder clientBuilder = Client.builder();
Client.Builder clientBuilder = Client.builder();
if (settings.hasValue("debug")) {
clientBuilder.enableDebug();
}

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

@ -22,9 +22,9 @@ import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.rest.BytesRestResponse;
import org.elasticsearch.rest.RestStatus;
import org.xbib.netty.http.client.Request;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.transport.Transport;
import org.xbib.net.URL;
import org.xbib.netty.http.client.api.Request;
import org.xbib.netty.http.client.api.Transport;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
@ -56,17 +56,17 @@ public abstract class HttpAction<R extends ActionRequest, T extends ActionRespon
listener.onFailure(validationException);
return;
}
RequestBuilder httpRequestBuilder =
Request.Builder httpRequestBuilder =
createHttpRequest(httpActionContext.getUrl(), httpActionContext.getRequest());
Request httpRequest = httpRequestBuilder.build();
httpRequest.setResponseListener(fullHttpResponse -> {
if (logger.isDebugEnabled()) {
logger.log(Level.DEBUG, "got response: " + fullHttpResponse.status().code() +
" headers = " + fullHttpResponse.headers().entries() +
" content = " + fullHttpResponse.content().toString(StandardCharsets.UTF_8));
logger.log(Level.DEBUG, "got response: " + fullHttpResponse.getStatus().getCode() +
" headers = " + fullHttpResponse.getHeaders() +
" content = " + fullHttpResponse.getBody().toString(StandardCharsets.UTF_8));
}
httpActionContext.setHttpResponse(fullHttpResponse);
if (fullHttpResponse.status().equals(HttpResponseStatus.OK)) {
if (fullHttpResponse.getStatus().getCode() == HttpResponseStatus.OK.code()) {
listener.onResponse(parseToResponse(httpActionContext));
} else {
ElasticsearchStatusException statusException = parseToError(httpActionContext);
@ -88,68 +88,68 @@ public abstract class HttpAction<R extends ActionRequest, T extends ActionRespon
}
}
protected RequestBuilder newGetRequest(String url, String path) {
protected Request.Builder newGetRequest(String url, String path) {
return newRequest(HttpMethod.GET, url, path);
}
protected RequestBuilder newGetRequest(String url, String path, BytesReference content) {
protected Request.Builder newGetRequest(String url, String path, BytesReference content) {
return newRequest(HttpMethod.GET, url, path, content);
}
protected RequestBuilder newHeadRequest(String url, String path) {
protected Request.Builder newHeadRequest(String url, String path) {
return newRequest(HttpMethod.HEAD, url, path);
}
protected RequestBuilder newPostRequest(String url, String path) {
protected Request.Builder newPostRequest(String url, String path) {
return newRequest(HttpMethod.POST, url, path);
}
protected RequestBuilder newPostRequest(String url, String path, BytesReference content) {
protected Request.Builder newPostRequest(String url, String path, BytesReference content) {
return newRequest(HttpMethod.POST, url, path, content);
}
protected RequestBuilder newPostRequest(String url, String path, String content) {
protected Request.Builder newPostRequest(String url, String path, String content) {
return newRequest(HttpMethod.POST, url, path, content);
}
protected RequestBuilder newPutRequest(String url, String path) {
protected Request.Builder newPutRequest(String url, String path) {
return newRequest(HttpMethod.PUT, url, path);
}
protected RequestBuilder newPutRequest(String url, String path, String content) {
protected Request.Builder newPutRequest(String url, String path, String content) {
return newRequest(HttpMethod.PUT, url, path, content);
}
protected RequestBuilder newPutRequest(String url, String path, BytesReference content) {
protected Request.Builder newPutRequest(String url, String path, BytesReference content) {
return newRequest(HttpMethod.PUT, url, path, content);
}
protected RequestBuilder newDeleteRequest(String url, String path) {
protected Request.Builder newDeleteRequest(String url, String path) {
return newRequest(HttpMethod.DELETE, url, path);
}
protected RequestBuilder newDeleteRequest(String url, String path, BytesReference content) {
protected Request.Builder newDeleteRequest(String url, String path, BytesReference content) {
return newRequest(HttpMethod.DELETE, url, path, content);
}
protected RequestBuilder newRequest(HttpMethod method, String baseUrl, String path) {
return Request.builder(method).url(baseUrl).uri(path);
protected Request.Builder newRequest(HttpMethod method, String baseUrl, String path) {
return Request.builder(method).url(URL.from(baseUrl).resolve(path));
}
protected RequestBuilder newRequest(HttpMethod method, String baseUrl, String path, BytesReference content) {
return Request.builder(method).url(baseUrl).uri(path).content(content.toBytesRef().bytes, APPLICATION_JSON);
protected Request.Builder newRequest(HttpMethod method, String baseUrl, String path, BytesReference content) {
return Request.builder(method).url(URL.from(baseUrl).resolve(path)).content(content.toBytesRef().bytes, APPLICATION_JSON);
}
protected RequestBuilder newRequest(HttpMethod method, String baseUrl, String path, String content) {
return Request.builder(method).url(baseUrl).uri(path).content(content, APPLICATION_JSON);
protected Request.Builder newRequest(HttpMethod method, String baseUrl, String path, String content) {
return Request.builder(method).url(URL.from(baseUrl).resolve(path)).content(content, APPLICATION_JSON);
}
protected RequestBuilder newRequest(HttpMethod method, String baseUrl, String path, ByteBuf byteBuf) {
return Request.builder(method).url(baseUrl).uri(path).content(byteBuf, APPLICATION_JSON);
protected Request.Builder newRequest(HttpMethod method, String baseUrl, String path, ByteBuf byteBuf) {
return Request.builder(method).url(URL.from(baseUrl).resolve(path)).content(byteBuf, APPLICATION_JSON);
}
protected T parseToResponse(HttpActionContext<R, T> httpActionContext) {
String mediaType = httpActionContext.getHttpResponse().headers().get(HttpHeaderNames.CONTENT_TYPE);
String mediaType = httpActionContext.getHttpResponse().getHeaders().getHeader(HttpHeaderNames.CONTENT_TYPE);
XContentType xContentType = XContentType.fromMediaTypeOrFormat(mediaType);
if (xContentType == null) {
throw new IllegalStateException("unsupported content-type: " + mediaType);
@ -157,7 +157,7 @@ public abstract class HttpAction<R extends ActionRequest, T extends ActionRespon
try (XContentParser parser = xContentType.xContent()
.createParser(httpActionContext.getExtendedHttpClient().getRegistry(),
DeprecationHandler.THROW_UNSUPPORTED_OPERATION,
httpActionContext.getHttpResponse().content().toString(StandardCharsets.UTF_8))) {
httpActionContext.getHttpResponse().getBody().toString(StandardCharsets.UTF_8))) {
return entityParser().apply(parser);
} catch (IOException e) {
logger.error(e.getMessage(), e);
@ -169,7 +169,7 @@ public abstract class HttpAction<R extends ActionRequest, T extends ActionRespon
try (XContentParser parser = XContentFactory.xContent(XContentType.JSON)
.createParser(httpActionContext.getExtendedHttpClient().getRegistry(),
DeprecationHandler.THROW_UNSUPPORTED_OPERATION,
httpActionContext.getHttpResponse().content().toString(StandardCharsets.UTF_8))) {
httpActionContext.getHttpResponse().getBody().toString(StandardCharsets.UTF_8))) {
return errorParser().apply(parser);
} catch (IOException e) {
logger.error(e.getMessage(), e);
@ -181,7 +181,7 @@ public abstract class HttpAction<R extends ActionRequest, T extends ActionRespon
return BytesRestResponse::errorFromXContent;
}
protected abstract RequestBuilder createHttpRequest(String baseUrl, R request) throws IOException;
protected abstract Request.Builder createHttpRequest(String baseUrl, R request) throws IOException;
protected abstract CheckedFunction<XContentParser, T, IOException> entityParser();

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

@ -1,9 +1,9 @@
package org.xbib.elx.http;
import io.netty.handler.codec.http.FullHttpResponse;
import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.action.ActionResponse;
import org.xbib.netty.http.client.transport.Transport;
import org.xbib.netty.http.client.api.Transport;
import org.xbib.netty.http.common.HttpResponse;
/**
* HTTP action context.
@ -21,7 +21,7 @@ public class HttpActionContext<R extends ActionRequest, T extends ActionResponse
private Transport httpClientTransport;
private FullHttpResponse httpResponse;
private HttpResponse httpResponse;
HttpActionContext(ExtendedHttpClient extendedHttpClient, R request, String url) {
this.extendedHttpClient = extendedHttpClient;
@ -49,12 +49,12 @@ public class HttpActionContext<R extends ActionRequest, T extends ActionResponse
return httpClientTransport;
}
public HttpActionContext<R, T> setHttpResponse(FullHttpResponse fullHttpResponse) {
this.httpResponse = fullHttpResponse;
public HttpActionContext<R, T> setHttpResponse(HttpResponse httpResponse) {
this.httpResponse = httpResponse;
return this;
}
public FullHttpResponse getHttpResponse() {
public HttpResponse getHttpResponse() {
return httpResponse;
}
}

2
elx-http/src/main/java/org/xbib/elx/http/HttpActionFuture.java
View file

@ -7,7 +7,7 @@ import org.elasticsearch.action.ActionListener;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.util.concurrent.BaseFuture;
import org.elasticsearch.common.util.concurrent.UncategorizedExecutionException;
import org.xbib.netty.http.client.transport.Transport;
import org.xbib.netty.http.client.api.Transport;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

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

@ -6,7 +6,7 @@ import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
@ -18,7 +18,7 @@ public class HttpClusterHealthAction extends HttpAction<ClusterHealthRequest, Cl
}
@Override
protected RequestBuilder createHttpRequest(String url, ClusterHealthRequest request) {
protected Request.Builder createHttpRequest(String url, ClusterHealthRequest request) {
return newGetRequest(url, "/_cluster/health");
}

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

@ -23,7 +23,7 @@ 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.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
import java.io.UncheckedIOException;
@ -53,7 +53,7 @@ public class HttpNodesInfoAction extends HttpAction<NodesInfoRequest, NodesInfoR
* @return HTTP request
*/
@Override
protected RequestBuilder createHttpRequest(String url, NodesInfoRequest request) {
protected Request.Builder createHttpRequest(String url, NodesInfoRequest request) {
StringBuilder path = new StringBuilder("/_nodes");
if (request.nodesIds() != null) {
String nodeIds = String.join(",", request.nodesIds());
@ -104,13 +104,17 @@ public class HttpNodesInfoAction extends HttpAction<NodesInfoRequest, NodesInfoR
return new NodesInfoResponse();
}
/**
* Broken.
*/
@SuppressWarnings("unchecked")
protected NodesInfoResponse createResponse(HttpActionContext<NodesInfoRequest, NodesInfoResponse> httpContext) {
Map<String, Object> map = null;
String string = (String)map.get("cluster_name");
ClusterName clusterName = new ClusterName(string);
//String string = (String)map.get("cluster_name");
ClusterName clusterName = new ClusterName("");
List<NodeInfo> nodeInfoList = new LinkedList<>();
map = (Map<String, Object>)map.get("nodes");
//map = (Map<String, Object>)map.get("nodes");
for (Map.Entry<String, Object> entry : map.entrySet()) {
String nodeId = entry.getKey();
String ephemeralId = null;

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

@ -9,7 +9,7 @@ import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
import java.io.UncheckedIOException;
@ -24,7 +24,7 @@ public class HttpClusterUpdateSettingsAction extends HttpAction<ClusterUpdateSet
}
@Override
protected RequestBuilder createHttpRequest(String url, ClusterUpdateSettingsRequest request) {
protected Request.Builder createHttpRequest(String url, ClusterUpdateSettingsRequest request) {
try {
XContentBuilder builder = jsonBuilder();
builder.startObject().startObject("persistent");

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

@ -19,7 +19,7 @@ import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.NamedObjectNotFoundException;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
import java.util.ArrayList;
@ -48,7 +48,7 @@ public class HttpClusterStateAction extends HttpAction<ClusterStateRequest, Clus
}
@Override
protected RequestBuilder createHttpRequest(String url, ClusterStateRequest request) {
protected Request.Builder createHttpRequest(String url, ClusterStateRequest request) {
List<String> list = new ArrayList<>();
if (request.metaData()) {
list.add("metadata");

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

@ -11,7 +11,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
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.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
@ -23,7 +23,7 @@ public class HttpIndicesAliasesAction extends HttpAction<IndicesAliasesRequest,
}
@Override
protected RequestBuilder createHttpRequest(String url, IndicesAliasesRequest request) {
protected Request.Builder createHttpRequest(String url, IndicesAliasesRequest request) {
try {
XContentBuilder builder = JsonXContent.contentBuilder();
request.toXContent(builder, ToXContent.EMPTY_PARAMS);

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

@ -9,7 +9,7 @@ import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
import java.util.ArrayList;
@ -20,7 +20,7 @@ import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpect
public class HttpGetAliasAction extends HttpAction<GetAliasesRequest, GetAliasesResponse> {
@Override
protected RequestBuilder createHttpRequest(String url, GetAliasesRequest request) {
protected Request.Builder createHttpRequest(String url, GetAliasesRequest request) {
// beware of this inconsistency, request.indices() always return empty array
String index = request.indices() != null ? String.join(",", request.indices()) + "/" : "";
String aliases = request.aliases() != null ? String.join(",", request.aliases()) + "/" : "";

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

@ -10,7 +10,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
@ -22,7 +22,7 @@ public class HttpCreateIndexAction extends HttpAction<CreateIndexRequest, Create
}
@Override
protected RequestBuilder createHttpRequest(String url, CreateIndexRequest createIndexRequest) throws IOException {
protected Request.Builder createHttpRequest(String url, CreateIndexRequest createIndexRequest) throws IOException {
XContentBuilder builder = XContentFactory.jsonBuilder();
builder = createIndexRequest.toXContent(builder, ToXContent.EMPTY_PARAMS);
return newPutRequest(url, "/" + createIndexRequest.index(), BytesReference.bytes(builder));

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

@ -6,7 +6,7 @@ import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
@ -18,7 +18,7 @@ public class HttpDeleteIndexAction extends HttpAction<DeleteIndexRequest, Delete
}
@Override
protected RequestBuilder createHttpRequest(String url, DeleteIndexRequest deleteIndexRequest) {
protected Request.Builder createHttpRequest(String url, DeleteIndexRequest deleteIndexRequest) {
return newDeleteRequest(url, "/" + String.join(",", deleteIndexRequest.indices()));
}

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

@ -9,7 +9,7 @@ import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.rest.RestStatus;
import org.xbib.elx.http.HttpAction;
import org.xbib.elx.http.HttpActionContext;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
@ -21,7 +21,7 @@ public class HttpIndicesExistsAction extends HttpAction<IndicesExistsRequest, In
}
@Override
protected RequestBuilder createHttpRequest(String url, IndicesExistsRequest request) {
protected Request.Builder createHttpRequest(String url, IndicesExistsRequest request) {
String index = String.join(",", request.indices());
return newHeadRequest(url, index);
}

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

@ -6,7 +6,7 @@ import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
@ -18,7 +18,7 @@ public class HttpRefreshIndexAction extends HttpAction<RefreshRequest, RefreshRe
}
@Override
protected RequestBuilder createHttpRequest(String url, RefreshRequest request) {
protected Request.Builder createHttpRequest(String url, RefreshRequest request) {
String index = request.indices() != null ? String.join(",", request.indices()) + "/" : "";
return newPostRequest(url, "/" + index + "_refresh");
}

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

@ -9,7 +9,7 @@ import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentParserUtils;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
import java.util.HashMap;
@ -23,7 +23,7 @@ public class HttpGetSettingsAction extends HttpAction<GetSettingsRequest, GetSet
}
@Override
protected RequestBuilder createHttpRequest(String url, GetSettingsRequest request) {
protected Request.Builder createHttpRequest(String url, GetSettingsRequest request) {
// beware, request.indices() is always an empty array
String index = request.indices() != null ? String.join(",", request.indices()) + "/" : "";
return newGetRequest(url, index + "_settings");

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

@ -10,7 +10,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
import java.io.UncheckedIOException;
@ -23,7 +23,7 @@ public class HttpUpdateSettingsAction extends HttpAction<UpdateSettingsRequest,
}
@Override
protected RequestBuilder createHttpRequest(String url, UpdateSettingsRequest request) {
protected Request.Builder createHttpRequest(String url, UpdateSettingsRequest request) {
try {
XContentBuilder builder = XContentFactory.jsonBuilder();
builder.startObject();

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

@ -13,7 +13,7 @@ import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
@ -25,7 +25,7 @@ public class HttpBulkAction extends HttpAction<BulkRequest, BulkResponse> {
}
@Override
protected RequestBuilder createHttpRequest(String url, BulkRequest request) throws IOException {
protected Request.Builder createHttpRequest(String url, BulkRequest request) throws IOException {
StringBuilder bulkContent = new StringBuilder();
for (DocWriteRequest<?> actionRequest : request.requests()) {
if (actionRequest instanceof IndexRequest) {

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

@ -7,7 +7,7 @@ import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
@ -19,7 +19,7 @@ public class HttpExistsAction extends HttpAction<GetRequest, GetResponse> {
}
@Override
protected RequestBuilder createHttpRequest(String url, GetRequest request) {
protected Request.Builder createHttpRequest(String url, GetRequest request) {
return newHeadRequest(url, "/" + request.index() + "/" + request.type() + "/" + request.id());
}

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

@ -7,7 +7,7 @@ import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
@ -19,7 +19,7 @@ public class HttpGetAction extends HttpAction<GetRequest, GetResponse> {
}
@Override
protected RequestBuilder createHttpRequest(String url, GetRequest request) {
protected Request.Builder createHttpRequest(String url, GetRequest request) {
return newGetRequest(url, "/" + request.index() + "/" + request.type() + "/" + request.id());
}

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

@ -7,7 +7,7 @@ import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
@ -19,7 +19,7 @@ public class HttpIndexAction extends HttpAction<IndexRequest, IndexResponse> {
}
@Override
protected RequestBuilder createHttpRequest(String url, IndexRequest request) {
protected Request.Builder createHttpRequest(String url, IndexRequest request) {
return newPutRequest(url, "/" + request.index() + "/" + request.type() + "/" + request.id(),
request.source());
}

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

@ -7,7 +7,7 @@ import org.elasticsearch.action.main.MainResponse;
import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
@ -19,7 +19,7 @@ public class HttpMainAction extends HttpAction<MainRequest, MainResponse> {
}
@Override
protected RequestBuilder createHttpRequest(String url, MainRequest request) {
protected Request.Builder createHttpRequest(String url, MainRequest request) {
return newGetRequest(url, "/");
}

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

@ -6,7 +6,7 @@ import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.XContentParser;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
@ -18,7 +18,7 @@ public class HttpSearchAction extends HttpAction<SearchRequest, SearchResponse>
}
@Override
protected RequestBuilder createHttpRequest(String url, SearchRequest request) {
protected Request.Builder createHttpRequest(String url, SearchRequest request) {
// request.indices() always empty array
String index = request.indices() != null ? String.join(",", request.indices()) + "/" : "";
return newPostRequest(url, index + "_search", request.source().toString());

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

@ -11,7 +11,7 @@ import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType;
import org.xbib.elx.http.HttpAction;
import org.xbib.netty.http.client.RequestBuilder;
import org.xbib.netty.http.client.api.Request;
import java.io.IOException;
@ -23,7 +23,7 @@ public class HttpUpdateAction extends HttpAction<UpdateRequest, UpdateResponse>
}
@Override
protected RequestBuilder createHttpRequest(String url, UpdateRequest updateRequest) {
protected Request.Builder createHttpRequest(String url, UpdateRequest updateRequest) {
try {
// The Java API allows update requests with different content types
// set for the partial document and the upsert document. This client

2
elx-http/src/test/resources/log4j2.xml
View file

@ -6,7 +6,7 @@
</Console>
</appenders>
<Loggers>
<Root level="info">
<Root level="debug">
<AppenderRef ref="Console" />
</Root>
</Loggers>

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

@ -1,4 +1,4 @@
/**
*
* Node client extensions.
*/
package org.xbib.elx.node;

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

@ -141,7 +141,7 @@ class ClientTest {
void testThreadedRandomDocs() throws Exception {
int maxthreads = Runtime.getRuntime().availableProcessors();
Long maxActionsPerRequest = MAX_ACTIONS_PER_REQUEST;
final Long actions = ACTIONS;
final long actions = ACTIONS;
logger.info("maxthreads={} maxactions={} maxloop={}", maxthreads, maxActionsPerRequest, actions);
final ExtendedNodeClient client = ClientBuilder.builder(helper.client("1"))
.provider(ExtendedNodeClientProvider.class)
@ -169,13 +169,13 @@ class ClientTest {
});
}
logger.info("waiting for latch...");
if (latch.await(60L, TimeUnit.SECONDS)) {
if (latch.await(30L, TimeUnit.SECONDS)) {
logger.info("flush...");
client.flush();
client.waitForResponses(60L, TimeUnit.SECONDS);
client.waitForResponses(30L, TimeUnit.SECONDS);
logger.info("got all responses, executor service shutdown...");
executorService.shutdown();
executorService.awaitTermination(60L, TimeUnit.SECONDS);
executorService.awaitTermination(30L, TimeUnit.SECONDS);
logger.info("pool is shut down");
} else {
logger.warn("latch timeout");

28
elx-node/src/test/java/org/xbib/elx/node/test/IndexPruneTest.java
View file

@ -45,22 +45,22 @@ class IndexPruneTest {
.put("index.number_of_shards", 1)
.put("index.number_of_replicas", 0)
.build();
client.newIndex("test1", settings);
client.shiftIndex("test", "test1", Collections.emptyList());
client.newIndex("test2", settings);
client.shiftIndex("test", "test2", Collections.emptyList());
client.newIndex("test3", settings);
client.shiftIndex("test", "test3", Collections.emptyList());
client.newIndex("test4", settings);
client.shiftIndex("test", "test4", Collections.emptyList());
client.newIndex("test_prune1", settings);
client.shiftIndex("test_prune", "test_prune1", Collections.emptyList());
client.newIndex("test_prune2", settings);
client.shiftIndex("test_prune", "test_prune2", Collections.emptyList());
client.newIndex("test_prune3", settings);
client.shiftIndex("test_prune", "test_prune3", Collections.emptyList());
client.newIndex("test_prune4", settings);
client.shiftIndex("test_prune", "test_prune4", Collections.emptyList());
IndexPruneResult indexPruneResult =
client.pruneIndex("test", "test4", 2, 2, true);
assertTrue(indexPruneResult.getDeletedIndices().contains("test1"));
assertTrue(indexPruneResult.getDeletedIndices().contains("test2"));
assertFalse(indexPruneResult.getDeletedIndices().contains("test3"));
assertFalse(indexPruneResult.getDeletedIndices().contains("test4"));
client.pruneIndex("test_prune", "test_prune4", 2, 2, true);
assertTrue(indexPruneResult.getDeletedIndices().contains("test_prune1"));
assertTrue(indexPruneResult.getDeletedIndices().contains("test_prune2"));
assertFalse(indexPruneResult.getDeletedIndices().contains("test_prune3"));
assertFalse(indexPruneResult.getDeletedIndices().contains("test_prune4"));
List<Boolean> list = new ArrayList<>();
for (String index : Arrays.asList("test1", "test2", "test3", "test4")) {
for (String index : Arrays.asList("test_prune1", "test_prune2", "test_prune3", "test_prune4")) {
IndicesExistsRequest indicesExistsRequest = new IndicesExistsRequest();
indicesExistsRequest.indices(index);
IndicesExistsResponse indicesExistsResponse =

16
elx-node/src/test/java/org/xbib/elx/node/test/IndexShiftTest.java
View file

@ -41,21 +41,21 @@ class IndexShiftTest {
.put("index.number_of_shards", 1)
.put("index.number_of_replicas", 0)
.build();
client.newIndex("test1234", settings);
client.newIndex("test_shift", settings);
for (int i = 0; i < 1; i++) {
client.index("test1234", helper.randomString(1), false,
client.index("test_shift", helper.randomString(1), false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
client.flush();
client.waitForResponses(30L, TimeUnit.SECONDS);
IndexShiftResult indexShiftResult =
client.shiftIndex("test", "test1234", Arrays.asList("a", "b", "c"));
client.shiftIndex("test", "test_shift", Arrays.asList("a", "b", "c"));
assertTrue(indexShiftResult.getNewAliases().contains("a"));
assertTrue(indexShiftResult.getNewAliases().contains("b"));
assertTrue(indexShiftResult.getNewAliases().contains("c"));
assertTrue(indexShiftResult.getMovedAliases().isEmpty());
Map<String, String> aliases = client.getAliases("test1234");
Map<String, String> aliases = client.getAliases("test_shift");
assertTrue(aliases.containsKey("a"));
assertTrue(aliases.containsKey("b"));
assertTrue(aliases.containsKey("c"));
@ -68,15 +68,15 @@ class IndexShiftTest {
assertTrue(aliases.containsKey("c"));
assertTrue(aliases.containsKey("test"));
client.newIndex("test5678", settings);
client.newIndex("test_shift2", settings);
for (int i = 0; i < 1; i++) {
client.index("test5678", helper.randomString(1), false,
client.index("test_shift2", helper.randomString(1), false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
client.flush();
client.waitForResponses(30L, TimeUnit.SECONDS);
indexShiftResult = client.shiftIndex("test", "test5678", Arrays.asList("d", "e", "f"),
indexShiftResult = client.shiftIndex("test", "test_shift2", Arrays.asList("d", "e", "f"),
(request, index, alias) -> request.addAliasAction(IndicesAliasesRequest.AliasActions.add()
.index(index).alias(alias).filter(QueryBuilders.termQuery("my_key", alias)))
);
@ -87,7 +87,7 @@ class IndexShiftTest {
assertTrue(indexShiftResult.getMovedAliases().contains("b"));
assertTrue(indexShiftResult.getMovedAliases().contains("c"));
aliases = client.getAliases("test5678");
aliases = client.getAliases("test_shift2");
assertTrue(aliases.containsKey("a"));
assertTrue(aliases.containsKey("b"));
assertTrue(aliases.containsKey("c"));

18
elx-node/src/test/java/org/xbib/elx/node/test/SmokeTest.java
View file

@ -34,20 +34,20 @@ class SmokeTest {
.build();
try {
assertEquals(helper.getClusterName(), client.getClusterName());
client.newIndex("test");
client.index("test", "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest
client.newIndex("test_smoke");
client.index("test_smoke", "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest
client.flush();
client.waitForResponses(30, TimeUnit.SECONDS);
client.checkMapping("test");
client.update("test", "1", "{ \"name\" : \"Another name\"}");
client.delete("test", "1");
client.checkMapping("test_smoke");
client.update("test_smoke", "1", "{ \"name\" : \"Another name\"}");
client.delete("test_smoke", "1");
client.flush();
client.waitForResponses(30, TimeUnit.SECONDS);
client.waitForRecovery("test", 10L, TimeUnit.SECONDS);
client.delete("test", "1");
client.waitForRecovery("test_smoke", 10L, TimeUnit.SECONDS);
client.delete("test_smoke", "1");
client.flush();
client.deleteIndex("test");
IndexDefinition indexDefinition = client.buildIndexDefinitionFromSettings("test", Settings.builder()
client.deleteIndex("test_smoke");
IndexDefinition indexDefinition = client.buildIndexDefinitionFromSettings("test_smoke", Settings.builder()
.build());
assertEquals(0, indexDefinition.getReplicaLevel());
client.newIndex(indexDefinition);

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

@ -157,7 +157,7 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
return helper;
}
class Helper {
static class Helper {
String home;

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

@ -1,5 +1,6 @@
package org.xbib.elx.transport;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.Version;
@ -48,11 +49,11 @@ public class ExtendedTransportClient extends AbstractExtendedClient {
+ " " + System.getProperty("java.vm.version")
+ " Elasticsearch " + Version.CURRENT.toString();
Settings transportClientSettings = getTransportClientSettings(settings);
XContentBuilder settingsBuilder = XContentFactory.jsonBuilder().startObject();
//XContentBuilder settingsBuilder = XContentFactory.jsonBuilder().startObject();
XContentBuilder effectiveSettingsBuilder = XContentFactory.jsonBuilder().startObject();
logger.info("creating transport client on {} with custom settings {} and effective settings {}",
logger.log(Level.INFO, "creating transport client on {} with settings {}",
systemIdentifier,
Strings.toString(settings.toXContent(settingsBuilder, ToXContent.EMPTY_PARAMS).endObject()),
//Strings.toString(settings.toXContent(settingsBuilder, ToXContent.EMPTY_PARAMS).endObject()),
Strings.toString(transportClientSettings.toXContent(effectiveSettingsBuilder,
ToXContent.EMPTY_PARAMS).endObject()));
return new MyTransportClient(transportClientSettings, Collections.singletonList(Netty4Plugin.class));
@ -138,11 +139,14 @@ public class ExtendedTransportClient extends AbstractExtendedClient {
private Settings getTransportClientSettings(Settings settings) {
return Settings.builder()
// "cluster.name"
.put(ClusterName.CLUSTER_NAME_SETTING.getKey(),
settings.get(ClusterName.CLUSTER_NAME_SETTING.getKey()))
// "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();
@ -154,7 +158,7 @@ public class ExtendedTransportClient extends AbstractExtendedClient {
}
}
class MyTransportClient extends TransportClient {
static class MyTransportClient extends TransportClient {
MyTransportClient(Settings settings, Collection<Class<? extends Plugin>> plugins) {
super(settings, plugins);

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

@ -1,5 +1,6 @@
package org.xbib.elx.transport.test;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsAction;
@ -37,9 +38,9 @@ class ClientTest {
private static final Logger logger = LogManager.getLogger(ClientTest.class.getName());
private static final Long ACTIONS = 1000L;
private static final Long ACTIONS = 10000L;
private static final Long MAX_ACTIONS_PER_REQUEST = 100L;
private static final Long MAX_ACTIONS_PER_REQUEST = 10000L;
private final TestExtension.Helper helper;
@ -51,9 +52,9 @@ class ClientTest {
void testSingleDoc() throws Exception {
final ExtendedTransportClient client = ClientBuilder.builder()
.provider(ExtendedTransportClientProvider.class)
.put(helper.getTransportSettings())
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(30))
.put(helper.getTransportSettings())
.build();
try {
client.newIndex("test");
@ -76,8 +77,8 @@ class ClientTest {
void testNewIndex() throws Exception {
final ExtendedTransportClient client = ClientBuilder.builder()
.provider(ExtendedTransportClientProvider.class)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(5))
.put(helper.getTransportSettings())
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(5))
.build();
client.newIndex("test");
client.close();
@ -87,8 +88,8 @@ class ClientTest {
void testMapping() throws Exception {
final ExtendedTransportClient client = ClientBuilder.builder()
.provider(ExtendedTransportClientProvider.class)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(5))
.put(helper.getTransportSettings())
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(5))
.build();
XContentBuilder builder = JsonXContent.contentBuilder()
.startObject()
@ -114,9 +115,9 @@ class ClientTest {
long numactions = ACTIONS;
final ExtendedTransportClient client = ClientBuilder.builder()
.provider(ExtendedTransportClientProvider.class)
.put(helper.getTransportSettings())
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), MAX_ACTIONS_PER_REQUEST)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(60))
.put(helper.getTransportSettings())
.build();
try {
client.newIndex("test");
@ -144,6 +145,7 @@ class ClientTest {
searchRequest.indices("test");
searchRequest.source(builder);
SearchResponse searchResponse = client.getClient().execute(SearchAction.INSTANCE, searchRequest).actionGet();
logger.log(Level.INFO, searchResponse.toString());
assertEquals(numactions, searchResponse.getHits().getTotalHits());
client.close();
}
@ -153,14 +155,15 @@ class ClientTest {
void testThreadedRandomDocs() throws Exception {
int maxthreads = Runtime.getRuntime().availableProcessors();
Long maxActionsPerRequest = MAX_ACTIONS_PER_REQUEST;
final Long actions = ACTIONS;
final long actions = ACTIONS;
logger.info("maxthreads={} maxactions={} maxloop={}", maxthreads, maxActionsPerRequest, actions);
final ExtendedTransportClient client = ClientBuilder.builder()
.provider(ExtendedTransportClientProvider.class)
.put(helper.getTransportSettings())
.put(Parameters.MAX_CONCURRENT_REQUESTS.name(), maxthreads * 2)
.put(Parameters.MAX_ACTIONS_PER_REQUEST.name(), maxActionsPerRequest)
.put(Parameters.FLUSH_INTERVAL.name(), TimeValue.timeValueSeconds(60))
.put(helper.getTransportSettings())
.put(Parameters.ENABLE_BULK_LOGGING.name(), "true")
.build();
try {
Settings settings = Settings.builder()
@ -182,13 +185,13 @@ class ClientTest {
});
}
logger.info("waiting for latch...");
if (latch.await(60L, TimeUnit.SECONDS)) {
if (latch.await(30L, TimeUnit.SECONDS)) {
logger.info("flush...");
client.flush();
client.waitForResponses(60L, TimeUnit.SECONDS);
client.waitForResponses(30L, TimeUnit.SECONDS);
logger.info("got all responses, executor service shutdown...");
executorService.shutdown();
executorService.awaitTermination(60L, TimeUnit.SECONDS);
executorService.awaitTermination(30L, TimeUnit.SECONDS);
logger.info("pool is shut down");
} else {
logger.warn("latch timeout");
@ -212,6 +215,7 @@ class ClientTest {
searchRequest.indices("test");
searchRequest.source(builder);
SearchResponse searchResponse = client.getClient().execute(SearchAction.INSTANCE, searchRequest).actionGet();
logger.log(Level.INFO, searchResponse.toString());
assertEquals(maxthreads * actions, searchResponse.getHits().getTotalHits());
client.close();
}

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

@ -45,20 +45,20 @@ class DuplicateIDTest {
.put(helper.getTransportSettings())
.build();
try {
client.newIndex("test");
client.newIndex("test_dup");
for (int i = 0; i < ACTIONS; i++) {
client.index("test", helper.randomString(1), false,
client.index("test_dup", helper.randomString(1), false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
client.flush();
client.waitForResponses(30L, TimeUnit.SECONDS);
client.refreshIndex("test");
client.refreshIndex("test_dup");
SearchSourceBuilder builder = new SearchSourceBuilder();
builder.query(QueryBuilders.matchAllQuery());
builder.size(0);
SearchRequest searchRequest = new SearchRequest();
searchRequest.indices("test");
searchRequest.types("test");
searchRequest.indices("test_dup");
searchRequest.types("test_dup");
searchRequest.source(builder);
SearchResponse searchResponse =
helper.client("1").execute(SearchAction.INSTANCE, searchRequest).actionGet();

28
elx-transport/src/test/java/org/xbib/elx/transport/test/IndexPruneTest.java
View file

@ -46,22 +46,22 @@ class IndexPruneTest {
.put("index.number_of_shards", 1)
.put("index.number_of_replicas", 0)
.build();
client.newIndex("test1", settings);
client.shiftIndex("test", "test1", Collections.emptyList());
client.newIndex("test2", settings);
client.shiftIndex("test", "test2", Collections.emptyList());
client.newIndex("test3", settings);
client.shiftIndex("test", "test3", Collections.emptyList());
client.newIndex("test4", settings);
client.shiftIndex("test", "test4", Collections.emptyList());
client.newIndex("test_prune1", settings);
client.shiftIndex("test_prune", "test_prune1", Collections.emptyList());
client.newIndex("test_prune2", settings);
client.shiftIndex("test_prune", "test_prune2", Collections.emptyList());
client.newIndex("test_prune3", settings);
client.shiftIndex("test_prune", "test_prune3", Collections.emptyList());
client.newIndex("test_prune4", settings);
client.shiftIndex("test_prune", "test_prune4", Collections.emptyList());
IndexPruneResult indexPruneResult =
client.pruneIndex("test", "test4", 2, 2, true);
assertTrue(indexPruneResult.getDeletedIndices().contains("test1"));
assertTrue(indexPruneResult.getDeletedIndices().contains("test2"));
assertFalse(indexPruneResult.getDeletedIndices().contains("test3"));
assertFalse(indexPruneResult.getDeletedIndices().contains("test4"));
client.pruneIndex("test_prune", "test_prune4", 2, 2, true);
assertTrue(indexPruneResult.getDeletedIndices().contains("test_prune1"));
assertTrue(indexPruneResult.getDeletedIndices().contains("test_prune2"));
assertFalse(indexPruneResult.getDeletedIndices().contains("test_prune3"));
assertFalse(indexPruneResult.getDeletedIndices().contains("test_prune4"));
List<Boolean> list = new ArrayList<>();
for (String index : Arrays.asList("test1", "test2", "test3", "test4")) {
for (String index : Arrays.asList("test_prune1", "test_prune2", "test_prune3", "test_prune4")) {
IndicesExistsRequest indicesExistsRequest = new IndicesExistsRequest();
indicesExistsRequest.indices(index);
IndicesExistsResponse indicesExistsResponse =

24
elx-transport/src/test/java/org/xbib/elx/transport/test/IndexShiftTest.java
View file

@ -42,38 +42,38 @@ class IndexShiftTest {
.put("index.number_of_shards", 1)
.put("index.number_of_replicas", 0)
.build();
client.newIndex("test1234", settings);
client.newIndex("test_shift1234", settings);
for (int i = 0; i < 1; i++) {
client.index("test1234", helper.randomString(1), false,
client.index("test_shift1234", helper.randomString(1), false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
client.flush();
client.waitForResponses(30L, TimeUnit.SECONDS);
IndexShiftResult indexShiftResult =
client.shiftIndex("test", "test1234", Arrays.asList("a", "b", "c"));
client.shiftIndex("test_shift", "test_shift1234", Arrays.asList("a", "b", "c"));
assertTrue(indexShiftResult.getNewAliases().contains("a"));
assertTrue(indexShiftResult.getNewAliases().contains("b"));
assertTrue(indexShiftResult.getNewAliases().contains("c"));
assertTrue(indexShiftResult.getMovedAliases().isEmpty());
Map<String, String> aliases = client.getAliases("test1234");
Map<String, String> aliases = client.getAliases("test_shift1234");
assertTrue(aliases.containsKey("a"));
assertTrue(aliases.containsKey("b"));
assertTrue(aliases.containsKey("c"));
assertTrue(aliases.containsKey("test"));
String resolved = client.resolveAlias("test");
assertTrue(aliases.containsKey("test_shift"));
String resolved = client.resolveAlias("test_shift");
aliases = client.getAliases(resolved);
assertTrue(aliases.containsKey("a"));
assertTrue(aliases.containsKey("b"));
assertTrue(aliases.containsKey("c"));
assertTrue(aliases.containsKey("test"));
client.newIndex("test5678", settings);
assertTrue(aliases.containsKey("test_shift"));
client.newIndex("test_shift5678", settings);
for (int i = 0; i < 1; i++) {
client.index("test5678", helper.randomString(1), false,
client.index("test_shift5678", helper.randomString(1), false,
"{ \"name\" : \"" + helper.randomString(32) + "\"}");
}
client.flush();
client.waitForResponses(30L, TimeUnit.SECONDS);
indexShiftResult = client.shiftIndex("test", "test5678", Arrays.asList("d", "e", "f"),
indexShiftResult = client.shiftIndex("test_shift", "test_shift5678", Arrays.asList("d", "e", "f"),
(request, index, alias) -> request.addAliasAction(IndicesAliasesRequest.AliasActions.add()
.index(index).alias(alias).filter(QueryBuilders.termQuery("my_key", alias)))
);
@ -83,14 +83,14 @@ class IndexShiftTest {
assertTrue(indexShiftResult.getMovedAliases().contains("a"));
assertTrue(indexShiftResult.getMovedAliases().contains("b"));
assertTrue(indexShiftResult.getMovedAliases().contains("c"));
aliases = client.getAliases("test5678");
aliases = client.getAliases("test_shift5678");
assertTrue(aliases.containsKey("a"));
assertTrue(aliases.containsKey("b"));
assertTrue(aliases.containsKey("c"));
assertTrue(aliases.containsKey("d"));
assertTrue(aliases.containsKey("e"));
assertTrue(aliases.containsKey("f"));
resolved = client.resolveAlias("test");
resolved = client.resolveAlias("test_shift");
aliases = client.getAliases(resolved);
assertTrue(aliases.containsKey("a"));
assertTrue(aliases.containsKey("b"));

14
elx-transport/src/test/java/org/xbib/elx/transport/test/SmokeTest.java
View file

@ -35,15 +35,15 @@ class SmokeTest {
.build();
try {
assertEquals(helper.getClusterName(), client.getClusterName());
client.newIndex("test");
client.index("test", "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest
client.update("test", "1", "{ \"name\" : \"Another name\"}");
client.delete("test", "1");
client.newIndex("test_smoke");
client.index("test_smoke", "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest
client.update("test_smoke", "1", "{ \"name\" : \"Another name\"}");
client.delete("test_smoke", "1");
client.flush();
client.waitForResponses(30, TimeUnit.SECONDS);
client.checkMapping("test");
client.deleteIndex("test");
IndexDefinition indexDefinition = client.buildIndexDefinitionFromSettings("test", Settings.builder()
client.checkMapping("test_smoke");
client.deleteIndex("test_smoke");
IndexDefinition indexDefinition = client.buildIndexDefinitionFromSettings("test_smoke", Settings.builder()
.build());
assertEquals(0, indexDefinition.getReplicaLevel());
client.newIndex(indexDefinition);

2
elx-transport/src/test/resources/log4j2.xml
View file

@ -6,7 +6,7 @@
</Console>
</appenders>
<Loggers>
<Root level="info">
<Root level="debug">
<AppenderRef ref="Console" />
</Root>
</Loggers>

13
gradle.properties
View file

@ -1,13 +1,16 @@
group = org.xbib
name = elx
version = 6.3.2.3
version = 6.3.2.4
profile = default
release = 0
elasticsearch-server.version = 6.3.2.3
log4j.version = 2.11.1
xbib-metrics.version = 1.2.0
xbib-netty-http.version = 4.1.35.0
elasticsearch-server.version = 6.3.2.4
log4j.version = 2.12.1
xbib-metrics.version = 2.0.0
xbib-netty-http.version = 4.1.48.0
tcnative.version = 2.0.29.Final
tcnative-legacy-macosx.version = 2.0.26.Final
bouncycastle.version = 1.64
# test
junit.version = 5.4.2