align with es221, cluster health, no IOException in Admin API

This commit is contained in:
Jörg Prante 2021-04-22 16:44:18 +02:00
parent da5dbaf71f
commit 40cdde3ae8
45 changed files with 179 additions and 295 deletions

View file

@ -3,13 +3,12 @@ package org.xbib.elx.api;
import org.elasticsearch.client.ElasticsearchClient; import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
import java.io.Closeable; import java.io.Closeable;
import java.io.IOException;
import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
public interface BasicClient extends Closeable { public interface BasicClient extends Closeable {
void putClusterSetting(String key, Object value, long timeout, TimeUnit timeUnit) throws IOException; void putClusterSetting(String key, Object value, long timeout, TimeUnit timeUnit);
/** /**
* Set an Elasticsearch client to extend from it. May be null for TransportClient. * Set an Elasticsearch client to extend from it. May be null for TransportClient.
@ -24,14 +23,7 @@ public interface BasicClient extends Closeable {
*/ */
ElasticsearchClient getClient(); ElasticsearchClient getClient();
/** void init(Settings settings);
* Initiative the extended client, the bulk metric and bulk controller,
* creates instances and connect to cluster, if required.
*
* @param settings settings
* @throws IOException if init fails
*/
void init(Settings settings) throws IOException;
/** /**
* Get cluster name. * Get cluster name.
@ -48,16 +40,7 @@ public interface BasicClient extends Closeable {
*/ */
String getHealthColor(long maxWaitTime, TimeUnit timeUnit); String getHealthColor(long maxWaitTime, TimeUnit timeUnit);
/** void waitForHealthyCluster();
* Wait for cluster being healthy.
*
* @param healthColor cluster health color to wait for
* @param maxWaitTime time value
* @param timeUnit time unit
*/
void waitForCluster(String healthColor, long maxWaitTime, TimeUnit timeUnit);
void waitForShards(long maxWaitTime, TimeUnit timeUnit);
long getSearchableDocs(IndexDefinition indexDefinition); long getSearchableDocs(IndexDefinition indexDefinition);

View file

@ -5,7 +5,6 @@ import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest; import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.bytes.BytesReference;
import java.io.Flushable; import java.io.Flushable;
import java.io.IOException;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
public interface BulkClient extends BasicClient, Flushable { public interface BulkClient extends BasicClient, Flushable {
@ -13,24 +12,21 @@ public interface BulkClient extends BasicClient, Flushable {
/** /**
* Create a new index. * Create a new index.
* @param indexDefinition the index definition * @param indexDefinition the index definition
* @throws IOException if settings/mapping is invalid or index creation fails
*/ */
void newIndex(IndexDefinition indexDefinition) throws IOException; void newIndex(IndexDefinition indexDefinition);
/** /**
* Start bulk mode for indexes. * Start bulk mode for indexes.
* @param indexDefinition index definition * @param indexDefinition index definition
* @throws IOException if bulk could not be started
*/ */
void startBulk(IndexDefinition indexDefinition) throws IOException; void startBulk(IndexDefinition indexDefinition);
/** /**
* Stop bulk mode. * Stop bulk mode.
* *
* @param indexDefinition index definition * @param indexDefinition index definition
* @throws IOException if bulk could not be startet
*/ */
void stopBulk(IndexDefinition indexDefinition) throws IOException; void stopBulk(IndexDefinition indexDefinition);
/** /**
* Add index request. Each request will be added to a queue for bulking requests. * Add index request. Each request will be added to a queue for bulking requests.
@ -131,9 +127,8 @@ public interface BulkClient extends BasicClient, Flushable {
* @param value the new value * @param value the new value
* @param timeout timeout * @param timeout timeout
* @param timeUnit time unit * @param timeUnit time unit
* @throws IOException if update index setting failed
*/ */
void updateIndexSetting(String index, String key, Object value, long timeout, TimeUnit timeUnit) throws IOException; void updateIndexSetting(String index, String key, Object value, long timeout, TimeUnit timeUnit);
/** /**
* Refresh the index. * Refresh the index.

View file

@ -4,16 +4,15 @@ import org.elasticsearch.action.DocWriteRequest;
import java.io.Closeable; import java.io.Closeable;
import java.io.Flushable; import java.io.Flushable;
import java.io.IOException;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
public interface BulkProcessor extends Closeable, Flushable { public interface BulkProcessor extends Closeable, Flushable {
void setEnabled(boolean enabled); void setEnabled(boolean enabled);
void startBulkMode(IndexDefinition indexDefinition) throws IOException; void startBulkMode(IndexDefinition indexDefinition);
void stopBulkMode(IndexDefinition indexDefinition) throws IOException; void stopBulkMode(IndexDefinition indexDefinition);
void add(DocWriteRequest<?> request); void add(DocWriteRequest<?> request);

View file

@ -105,7 +105,7 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
ensureClientIsPresent(); ensureClientIsPresent();
DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest().indices(index); DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest().indices(index);
client.execute(DeleteIndexAction.INSTANCE, deleteIndexRequest).actionGet(); client.execute(DeleteIndexAction.INSTANCE, deleteIndexRequest).actionGet();
waitForCluster("GREEN", 300L, TimeUnit.SECONDS); waitForHealthyCluster();
return this; return this;
} }
@ -121,7 +121,7 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
logger.info("update replica level for " + indexDefinition + " to " + level); logger.info("update replica level for " + indexDefinition + " to " + level);
updateIndexSetting(indexDefinition.getFullIndexName(), "number_of_replicas", level, updateIndexSetting(indexDefinition.getFullIndexName(), "number_of_replicas", level,
30L, TimeUnit.SECONDS); 30L, TimeUnit.SECONDS);
waitForCluster("GREEN", 300L, TimeUnit.SECONDS); waitForHealthyCluster();
return this; return this;
} }
@ -230,7 +230,6 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
if (index.equals(fullIndexName)) { if (index.equals(fullIndexName)) {
return new EmptyIndexShiftResult(); // nothing to shift to return new EmptyIndexShiftResult(); // nothing to shift to
} }
waitForCluster("YELLOW", 30L, TimeUnit.SECONDS);
// two situations: 1. a new alias 2. there is already an old index with the alias // two situations: 1. a new alias 2. there is already an old index with the alias
Optional<String> oldIndex = resolveAlias(index).stream().sorted().findFirst(); Optional<String> oldIndex = resolveAlias(index).stream().sorted().findFirst();
Map<String, String> oldAliasMap = oldIndex.map(this::getAliases).orElse(null); Map<String, String> oldAliasMap = oldIndex.map(this::getAliases).orElse(null);
@ -411,7 +410,7 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
if (forceMergeResponse.getFailedShards() > 0) { if (forceMergeResponse.getFailedShards() > 0) {
throw new IllegalStateException("failed shards after force merge: " + forceMergeResponse.getFailedShards()); throw new IllegalStateException("failed shards after force merge: " + forceMergeResponse.getFailedShards());
} }
waitForCluster("GREEN", 300L, TimeUnit.SECONDS); waitForHealthyCluster();
return true; return true;
} }
@ -423,7 +422,7 @@ public abstract class AbstractAdminClient extends AbstractBasicClient implements
UpdateSettingsRequest updateSettingsRequest = new UpdateSettingsRequest(index) UpdateSettingsRequest updateSettingsRequest = new UpdateSettingsRequest(index)
.settings(updateSettingsBuilder).timeout(toTimeValue(timeout, timeUnit)); .settings(updateSettingsBuilder).timeout(toTimeValue(timeout, timeUnit));
client.execute(UpdateSettingsAction.INSTANCE, updateSettingsRequest).actionGet(); client.execute(UpdateSettingsAction.INSTANCE, updateSettingsRequest).actionGet();
waitForCluster("GREEN", 300L, TimeUnit.SECONDS); waitForHealthyCluster();
} }
@Override @Override

View file

@ -70,7 +70,7 @@ public abstract class AbstractBasicClient implements BasicClient {
} }
@Override @Override
public void init(Settings settings) throws IOException { public void init(Settings settings) {
if (closed.compareAndSet(false, true)) { if (closed.compareAndSet(false, true)) {
logger.log(Level.INFO, "initializing with settings = " + settings.toDelimitedString(',')); logger.log(Level.INFO, "initializing with settings = " + settings.toDelimitedString(','));
this.settings = settings; this.settings = settings;
@ -99,13 +99,13 @@ public abstract class AbstractBasicClient implements BasicClient {
} }
@Override @Override
public void putClusterSetting(String key, Object value, long timeout, TimeUnit timeUnit) throws IOException { public void putClusterSetting(String key, Object value, long timeout, TimeUnit timeUnit) {
ensureClientIsPresent(); ensureClientIsPresent();
if (key == null) { if (key == null) {
throw new IOException("no key given"); throw new IllegalArgumentException("no key given");
} }
if (value == null) { if (value == null) {
throw new IOException("no value given"); throw new IllegalArgumentException("no value given");
} }
Settings.Builder updateSettingsBuilder = Settings.builder(); Settings.Builder updateSettingsBuilder = Settings.builder();
updateSettingsBuilder.put(key, value.toString()); updateSettingsBuilder.put(key, value.toString());
@ -115,8 +115,15 @@ public abstract class AbstractBasicClient implements BasicClient {
} }
@Override @Override
public void waitForCluster(String statusString, long maxWaitTime, TimeUnit timeUnit) { public void waitForHealthyCluster() {
ensureClientIsPresent(); ensureClientIsPresent();
String statusString = settings.get(Parameters.CLUSTER_TARGET_HEALTH.getName(),
Parameters.CLUSTER_TARGET_HEALTH.getString());
String waitTimeStr = settings.get(Parameters.CLUSTER_TARGET_HEALTH_TIMEOUT.getName(),
Parameters.CLUSTER_TARGET_HEALTH_TIMEOUT.getString());
TimeValue timeValue = TimeValue.parseTimeValue(waitTimeStr, TimeValue.timeValueMinutes(30L), "");
long maxWaitTime = timeValue.minutes();
TimeUnit timeUnit = TimeUnit.MINUTES;
logger.info("waiting for cluster status " + statusString + " for " + maxWaitTime + " " + timeUnit); logger.info("waiting for cluster status " + statusString + " for " + maxWaitTime + " " + timeUnit);
ClusterHealthStatus status = ClusterHealthStatus.fromString(statusString); ClusterHealthStatus status = ClusterHealthStatus.fromString(statusString);
TimeValue timeout = toTimeValue(maxWaitTime, timeUnit); TimeValue timeout = toTimeValue(maxWaitTime, timeUnit);
@ -132,24 +139,6 @@ public abstract class AbstractBasicClient implements BasicClient {
} }
} }
@Override
public void waitForShards(long maxWaitTime, TimeUnit timeUnit) {
ensureClientIsPresent();
logger.info("waiting for cluster shard settling");
TimeValue timeout = toTimeValue(maxWaitTime, timeUnit);
ClusterHealthRequest clusterHealthRequest = new ClusterHealthRequest()
.waitForNoInitializingShards(true)
.waitForNoRelocatingShards(true)
.timeout(timeout);
ClusterHealthResponse healthResponse =
client.execute(ClusterHealthAction.INSTANCE, clusterHealthRequest).actionGet();
if (healthResponse.isTimedOut()) {
String message = "timeout waiting for cluster shards: " + timeout;
logger.error(message);
throw new IllegalStateException(message);
}
}
@Override @Override
public String getHealthColor(long maxWaitTime, TimeUnit timeUnit) { public String getHealthColor(long maxWaitTime, TimeUnit timeUnit) {
ensureClientIsPresent(); ensureClientIsPresent();
@ -205,20 +194,20 @@ public abstract class AbstractBasicClient implements BasicClient {
} }
} }
protected abstract ElasticsearchClient createClient(Settings settings) throws IOException; protected abstract ElasticsearchClient createClient(Settings settings);
protected abstract void closeClient(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 { protected void updateIndexSetting(String index, String key, Object value, long timeout, TimeUnit timeUnit) {
ensureClientIsPresent(); ensureClientIsPresent();
if (index == null) { if (index == null) {
throw new IOException("no index name given"); throw new IllegalArgumentException("no index name given");
} }
if (key == null) { if (key == null) {
throw new IOException("no key given"); throw new IllegalArgumentException("no key given");
} }
if (value == null) { if (value == null) {
throw new IOException("no value given"); throw new IllegalArgumentException("no value given");
} }
Settings.Builder updateSettingsBuilder = Settings.builder(); Settings.Builder updateSettingsBuilder = Settings.builder();
updateSettingsBuilder.put(key, value.toString()); updateSettingsBuilder.put(key, value.toString());

View file

@ -1,5 +1,6 @@
package org.xbib.elx.common; package org.xbib.elx.common;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.admin.indices.create.CreateIndexAction; import org.elasticsearch.action.admin.indices.create.CreateIndexAction;
@ -46,7 +47,7 @@ public abstract class AbstractBulkClient extends AbstractBasicClient implements
} }
@Override @Override
public void init(Settings settings) throws IOException { public void init(Settings settings) {
if (closed.compareAndSet(true, false)) { if (closed.compareAndSet(true, false)) {
super.init(settings); super.init(settings);
bulkProcessor = new DefaultBulkProcessor(this, settings); bulkProcessor = new DefaultBulkProcessor(this, settings);
@ -79,7 +80,7 @@ public abstract class AbstractBulkClient extends AbstractBasicClient implements
} }
@Override @Override
public void newIndex(IndexDefinition indexDefinition) throws IOException { public void newIndex(IndexDefinition indexDefinition) {
if (isIndexDefinitionDisabled(indexDefinition)) { if (isIndexDefinitionDisabled(indexDefinition)) {
return; return;
} }
@ -91,24 +92,32 @@ public abstract class AbstractBulkClient extends AbstractBasicClient implements
CreateIndexRequestBuilder createIndexRequestBuilder = new CreateIndexRequestBuilder(client, CreateIndexAction.INSTANCE) CreateIndexRequestBuilder createIndexRequestBuilder = new CreateIndexRequestBuilder(client, CreateIndexAction.INSTANCE)
.setIndex(index); .setIndex(index);
if (indexDefinition.getSettings() == null) { if (indexDefinition.getSettings() == null) {
XContentBuilder builder = JsonXContent.contentBuilder() try {
.startObject() XContentBuilder builder = JsonXContent.contentBuilder()
.startObject("index") .startObject()
.field("number_of_shards", 1) .startObject("index")
.field("number_of_replicas", 0) .field("number_of_shards", 1)
.endObject() .field("number_of_replicas", 0)
.endObject(); .endObject()
indexDefinition.setSettings(Strings.toString(builder)); .endObject();
indexDefinition.setSettings(Strings.toString(builder));
} catch (IOException e) {
logger.log(Level.WARN, e.getMessage(), e);
}
} }
Settings settings = Settings.builder().loadFromSource(indexDefinition.getSettings(), XContentType.JSON).build(); Settings settings = Settings.builder().loadFromSource(indexDefinition.getSettings(), XContentType.JSON).build();
createIndexRequestBuilder.setSettings(settings); createIndexRequestBuilder.setSettings(settings);
if (indexDefinition.getMappings() != null) { try {
Map<String, Object> mappings = JsonXContent.jsonXContent.createParser(NamedXContentRegistry.EMPTY, if (indexDefinition.getMappings() != null) {
DeprecationHandler.THROW_UNSUPPORTED_OPERATION, indexDefinition.getMappings()).mapOrdered(); Map<String, Object> mappings = JsonXContent.jsonXContent.createParser(NamedXContentRegistry.EMPTY,
createIndexRequestBuilder.addMapping(TYPE_NAME, mappings); DeprecationHandler.THROW_UNSUPPORTED_OPERATION, indexDefinition.getMappings()).mapOrdered();
} else { createIndexRequestBuilder.addMapping(TYPE_NAME, mappings);
XContentBuilder builder = JsonXContent.contentBuilder().startObject().startObject(TYPE_NAME).endObject().endObject(); } else {
createIndexRequestBuilder.addMapping(TYPE_NAME, builder); XContentBuilder builder = JsonXContent.contentBuilder().startObject().startObject(TYPE_NAME).endObject().endObject();
createIndexRequestBuilder.addMapping(TYPE_NAME, builder);
}
} catch (IOException e) {
logger.log(Level.WARN, e.getMessage(), e);
} }
CreateIndexResponse createIndexResponse = createIndexRequestBuilder.execute().actionGet(); CreateIndexResponse createIndexResponse = createIndexRequestBuilder.execute().actionGet();
if (createIndexResponse.isAcknowledged()) { if (createIndexResponse.isAcknowledged()) {
@ -118,11 +127,11 @@ public abstract class AbstractBulkClient extends AbstractBasicClient implements
return; return;
} }
// we really need state GREEN. If yellow, we may trigger shard write errors and queue will exceed quickly. // we really need state GREEN. If yellow, we may trigger shard write errors and queue will exceed quickly.
waitForCluster("GREEN", 300L, TimeUnit.SECONDS); waitForHealthyCluster();
} }
@Override @Override
public void startBulk(IndexDefinition indexDefinition) throws IOException { public void startBulk(IndexDefinition indexDefinition) {
if (isIndexDefinitionDisabled(indexDefinition)) { if (isIndexDefinitionDisabled(indexDefinition)) {
return; return;
} }
@ -133,7 +142,7 @@ public abstract class AbstractBulkClient extends AbstractBasicClient implements
} }
@Override @Override
public void stopBulk(IndexDefinition indexDefinition) throws IOException { public void stopBulk(IndexDefinition indexDefinition) {
if (isIndexDefinitionDisabled(indexDefinition)) { if (isIndexDefinitionDisabled(indexDefinition)) {
return; return;
} }
@ -216,7 +225,7 @@ public abstract class AbstractBulkClient extends AbstractBasicClient implements
} }
@Override @Override
public void updateIndexSetting(String index, String key, Object value, long timeout, TimeUnit timeUnit) throws IOException { public void updateIndexSetting(String index, String key, Object value, long timeout, TimeUnit timeUnit) {
super.updateIndexSetting(index, key, value, timeout, timeUnit); super.updateIndexSetting(index, key, value, timeout, timeUnit);
} }

View file

@ -49,7 +49,7 @@ public abstract class AbstractSearchClient extends AbstractBasicClient implement
} }
@Override @Override
public void init(Settings settings) throws IOException { public void init(Settings settings) {
if (closed.compareAndSet(true, false)) { if (closed.compareAndSet(true, false)) {
super.init(settings); super.init(settings);
this.searchMetric = new DefaultSearchMetric(getScheduler(), settings); this.searchMetric = new DefaultSearchMetric(getScheduler(), settings);

View file

@ -97,7 +97,7 @@ public class DefaultBulkProcessor implements BulkProcessor {
} }
@Override @Override
public void startBulkMode(IndexDefinition indexDefinition) throws IOException { public void startBulkMode(IndexDefinition indexDefinition) {
String indexName = indexDefinition.getFullIndexName(); String indexName = indexDefinition.getFullIndexName();
int interval = indexDefinition.getStartBulkRefreshSeconds(); int interval = indexDefinition.getStartBulkRefreshSeconds();
if (interval != 0) { if (interval != 0) {
@ -110,7 +110,7 @@ public class DefaultBulkProcessor implements BulkProcessor {
} }
@Override @Override
public void stopBulkMode(IndexDefinition indexDefinition) throws IOException { public void stopBulkMode(IndexDefinition indexDefinition) {
String indexName = indexDefinition.getFullIndexName(); String indexName = indexDefinition.getFullIndexName();
int interval = indexDefinition.getStopBulkRefreshSeconds(); int interval = indexDefinition.getStopBulkRefreshSeconds();
flush(); flush();

View file

@ -125,7 +125,11 @@ public class DefaultSearchMetric implements SearchMetric {
private void log() { private void log() {
if (logger.isInfoEnabled()) { if (logger.isInfoEnabled()) {
logger.info("docs = " + getTotalQueries().getCount()); logger.info("queries = " + getTotalQueries().getCount() +
" succeeded = " + getSucceededQueries().getCount() +
" empty = " + getEmptyQueries().getCount() +
" failed = " + getFailedQueries() +
" timeouts = " + getTimeoutQueries().getCount());
} }
} }
} }

View file

@ -3,8 +3,6 @@ package org.xbib.elx.common;
import org.elasticsearch.client.ElasticsearchClient; import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
import java.util.concurrent.TimeUnit;
/** /**
* A mocked client, it does not perform any actions on a cluster. Useful for testing. * A mocked client, it does not perform any actions on a cluster. Useful for testing.
*/ */
@ -28,14 +26,6 @@ public class MockAdminClient extends AbstractAdminClient {
protected void closeClient(Settings settings) { protected void closeClient(Settings settings) {
} }
@Override
public void waitForCluster(String healthColor, long timeValue, TimeUnit timeUnit) {
}
@Override
public void waitForShards(long maxWaitTime, TimeUnit timeUnit) {
}
@Override @Override
public void close() { public void close() {
// nothing to do // nothing to do

View file

@ -26,10 +26,6 @@ public class MockBulkClient extends AbstractBulkClient {
return null; return null;
} }
@Override
public void waitForShards(long maxWaitTime, TimeUnit timeUnit) {
}
@Override @Override
protected ElasticsearchClient createClient(Settings settings) { protected ElasticsearchClient createClient(Settings settings) {
return null; return null;

View file

@ -3,8 +3,6 @@ package org.xbib.elx.common;
import org.elasticsearch.client.ElasticsearchClient; import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
import java.util.concurrent.TimeUnit;
/** /**
* A mocked client, it does not perform any actions on a cluster. Useful for testing. * A mocked client, it does not perform any actions on a cluster. Useful for testing.
*/ */
@ -24,10 +22,6 @@ public class MockSearchClient extends AbstractSearchClient {
return null; return null;
} }
@Override
public void waitForShards(long maxWaitTime, TimeUnit timeUnit) {
}
@Override @Override
protected ElasticsearchClient createClient(Settings settings) { protected ElasticsearchClient createClient(Settings settings) {
return null; return null;

View file

@ -2,6 +2,10 @@ package org.xbib.elx.common;
public enum Parameters { public enum Parameters {
CLUSTER_TARGET_HEALTH("cluster.target_health", String.class, "GREEN"),
CLUSTER_TARGET_HEALTH_TIMEOUT("cluster.target_health_timeout", String.class, "30m"),
DATE_TIME_FORMAT("dateTimeFormat", String.class, "yyyyMMdd"), DATE_TIME_FORMAT("dateTimeFormat", String.class, "yyyyMMdd"),
BULK_MAX_WAIT_RESPONSE("bulk.max_wait_response", String.class, "30s"), BULK_MAX_WAIT_RESPONSE("bulk.max_wait_response", String.class, "30s"),

View file

@ -24,18 +24,18 @@ public class HttpAdminClient extends AbstractAdminClient implements Elasticsearc
} }
@Override @Override
public void init(Settings settings) throws IOException { public void init(Settings settings) {
super.init(settings); super.init(settings);
helper.init(settings); helper.init(settings);
} }
@Override @Override
protected ElasticsearchClient createClient(Settings settings) throws IOException { protected ElasticsearchClient createClient(Settings settings) {
return this; return this;
} }
@Override @Override
protected void closeClient(Settings settings) throws IOException { protected void closeClient(Settings settings) {
helper.closeClient(settings); helper.closeClient(settings);
} }

View file

@ -24,7 +24,7 @@ public class HttpBulkClient extends AbstractBulkClient implements ElasticsearchC
} }
@Override @Override
public void init(Settings settings) throws IOException { public void init(Settings settings) {
super.init(settings); super.init(settings);
helper.init(settings); helper.init(settings);
} }

View file

@ -111,9 +111,13 @@ public class HttpClientHelper {
return nettyHttpClient; return nettyHttpClient;
} }
protected void closeClient(Settings settings) throws IOException { protected void closeClient(Settings settings) {
if (closed.compareAndSet(false, true)) { if (closed.compareAndSet(false, true)) {
nettyHttpClient.shutdownGracefully(); try {
nettyHttpClient.shutdownGracefully();
} catch (IOException e) {
logger.log(Level.WARN, e.getMessage(), e);
}
} }
} }

View file

@ -24,13 +24,13 @@ public class HttpSearchClient extends AbstractSearchClient implements Elasticsea
} }
@Override @Override
public void init(Settings settings) throws IOException { public void init(Settings settings) {
super.init(settings); super.init(settings);
helper.init(settings); helper.init(settings);
} }
@Override @Override
protected ElasticsearchClient createClient(Settings settings) throws IOException { protected ElasticsearchClient createClient(Settings settings) {
return this; return this;
} }

View file

@ -2,13 +2,11 @@ package org.xbib.elx.http.test;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.elasticsearch.common.unit.TimeValue;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.api.extension.ExtendWith;
import org.xbib.elx.api.IndexDefinition; import org.xbib.elx.api.IndexDefinition;
import org.xbib.elx.common.ClientBuilder; import org.xbib.elx.common.ClientBuilder;
import org.xbib.elx.common.DefaultIndexDefinition; import org.xbib.elx.common.DefaultIndexDefinition;
import org.xbib.elx.common.Parameters;
import org.xbib.elx.http.HttpBulkClient; import org.xbib.elx.http.HttpBulkClient;
import org.xbib.elx.http.HttpBulkClientProvider; import org.xbib.elx.http.HttpBulkClientProvider;
@ -37,7 +35,7 @@ class BulkClientTest {
void testNewIndex() throws Exception { void testNewIndex() throws Exception {
try (HttpBulkClient bulkClient = ClientBuilder.builder() try (HttpBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(HttpBulkClientProvider.class) .setBulkClientProvider(HttpBulkClientProvider.class)
.put(helper.getHttpSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
@ -48,7 +46,7 @@ class BulkClientTest {
void testSingleDoc() throws Exception { void testSingleDoc() throws Exception {
try (HttpBulkClient bulkClient = ClientBuilder.builder() try (HttpBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(HttpBulkClientProvider.class) .setBulkClientProvider(HttpBulkClientProvider.class)
.put(helper.getHttpSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
@ -68,7 +66,7 @@ class BulkClientTest {
long numactions = ACTIONS; long numactions = ACTIONS;
try (HttpBulkClient bulkClient = ClientBuilder.builder() try (HttpBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(HttpBulkClientProvider.class) .setBulkClientProvider(HttpBulkClientProvider.class)
.put(helper.getHttpSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
@ -94,7 +92,7 @@ class BulkClientTest {
long timeout = 120L; long timeout = 120L;
try (HttpBulkClient bulkClient = ClientBuilder.builder() try (HttpBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(HttpBulkClientProvider.class) .setBulkClientProvider(HttpBulkClientProvider.class)
.put(helper.getHttpSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);

View file

@ -1,6 +1,7 @@
package org.xbib.elx.http.test; package org.xbib.elx.http.test;
import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.index.query.QueryBuilders;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.xbib.elx.api.IndexDefinition; import org.xbib.elx.api.IndexDefinition;
import org.xbib.elx.common.ClientBuilder; import org.xbib.elx.common.ClientBuilder;
@ -14,6 +15,7 @@ import java.nio.file.Files;
import java.nio.file.Paths; import java.nio.file.Paths;
import java.util.stream.Stream; import java.util.stream.Stream;
@Disabled
class DumpIDTest { class DumpIDTest {
@Test @Test

View file

@ -34,7 +34,7 @@ class DuplicateIDTest {
long numactions = ACTIONS; long numactions = ACTIONS;
try (HttpBulkClient bulkClient = ClientBuilder.builder() try (HttpBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(HttpBulkClientProvider.class) .setBulkClientProvider(HttpBulkClientProvider.class)
.put(helper.getHttpSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);

View file

@ -40,11 +40,11 @@ class IndexPruneTest {
void testPrune() throws IOException { void testPrune() throws IOException {
try (HttpAdminClient adminClient = ClientBuilder.builder() try (HttpAdminClient adminClient = ClientBuilder.builder()
.setAdminClientProvider(HttpAdminClientProvider.class) .setAdminClientProvider(HttpAdminClientProvider.class)
.put(helper.getHttpSettings()) .put(helper.getClientSettings())
.build(); .build();
HttpBulkClient bulkClient = ClientBuilder.builder() HttpBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(HttpBulkClientProvider.class) .setBulkClientProvider(HttpBulkClientProvider.class)
.put(helper.getHttpSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
indexDefinition.setIndex("test"); indexDefinition.setIndex("test");

View file

@ -40,11 +40,11 @@ class IndexShiftTest {
void testIndexShift() throws Exception { void testIndexShift() throws Exception {
try (HttpAdminClient adminClient = ClientBuilder.builder() try (HttpAdminClient adminClient = ClientBuilder.builder()
.setAdminClientProvider(HttpAdminClientProvider.class) .setAdminClientProvider(HttpAdminClientProvider.class)
.put(helper.getHttpSettings()) .put(helper.getClientSettings())
.build(); .build();
HttpBulkClient bulkClient = ClientBuilder.builder() HttpBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(HttpBulkClientProvider.class) .setBulkClientProvider(HttpBulkClientProvider.class)
.put(helper.getHttpSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
indexDefinition.setFullIndexName("test_shift"); indexDefinition.setFullIndexName("test_shift");

View file

@ -39,7 +39,7 @@ class SearchTest {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
try (HttpBulkClient bulkClient = ClientBuilder.builder() try (HttpBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(HttpBulkClientProvider.class) .setBulkClientProvider(HttpBulkClientProvider.class)
.put(helper.getHttpSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
for (int i = 0; i < ACTIONS; i++) { for (int i = 0; i < ACTIONS; i++) {
@ -57,7 +57,7 @@ class SearchTest {
} }
try (HttpSearchClient searchClient = ClientBuilder.builder() try (HttpSearchClient searchClient = ClientBuilder.builder()
.setSearchClientProvider(HttpSearchClientProvider.class) .setSearchClientProvider(HttpSearchClientProvider.class)
.put(helper.getHttpSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
Stream<SearchHit> stream = searchClient.search(qb -> qb Stream<SearchHit> stream = searchClient.search(qb -> qb
.setIndices(indexDefinition.getFullIndexName()) .setIndices(indexDefinition.getFullIndexName())

View file

@ -37,11 +37,11 @@ class SmokeTest {
void smokeTest() throws Exception { void smokeTest() throws Exception {
try (HttpAdminClient adminClient = ClientBuilder.builder() try (HttpAdminClient adminClient = ClientBuilder.builder()
.setAdminClientProvider(HttpAdminClientProvider.class) .setAdminClientProvider(HttpAdminClientProvider.class)
.put(helper.getHttpSettings()) .put(helper.getClientSettings())
.build(); .build();
HttpBulkClient bulkClient = ClientBuilder.builder() HttpBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(HttpBulkClientProvider.class) .setBulkClientProvider(HttpBulkClientProvider.class)
.put(helper.getHttpSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = IndexDefinition indexDefinition =
new DefaultIndexDefinition(adminClient, "test_smoke", "doc", Settings.EMPTY); new DefaultIndexDefinition(adminClient, "test_smoke", "doc", Settings.EMPTY);
@ -49,21 +49,20 @@ class SmokeTest {
assertEquals(helper.getClusterName(), adminClient.getClusterName()); assertEquals(helper.getClusterName(), adminClient.getClusterName());
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest
bulkClient.waitForResponses(30, TimeUnit.SECONDS); assertTrue(bulkClient.waitForResponses(30, TimeUnit.SECONDS));
adminClient.checkMapping(indexDefinition); adminClient.checkMapping(indexDefinition);
bulkClient.update(indexDefinition, "1", "{ \"name\" : \"Another name\"}"); bulkClient.update(indexDefinition, "1", "{ \"name\" : \"Another name\"}");
bulkClient.delete(indexDefinition, "1"); bulkClient.delete(indexDefinition, "1");
bulkClient.waitForResponses(30, TimeUnit.SECONDS); assertTrue(bulkClient.waitForResponses(30, TimeUnit.SECONDS));
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}");
bulkClient.delete(indexDefinition, "1"); bulkClient.delete(indexDefinition, "1");
bulkClient.waitForResponses(30, TimeUnit.SECONDS); assertTrue(bulkClient.waitForResponses(30, TimeUnit.SECONDS));
adminClient.deleteIndex(indexDefinition); adminClient.deleteIndex(indexDefinition);
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}");
bulkClient.waitForResponses(30, TimeUnit.SECONDS); assertTrue(bulkClient.waitForResponses(30, TimeUnit.SECONDS));
adminClient.updateReplicaLevel(indexDefinition, 2); adminClient.updateReplicaLevel(indexDefinition, 1);
int replica = adminClient.getReplicaLevel(indexDefinition); assertEquals(1, adminClient.getReplicaLevel(indexDefinition));
assertEquals(2, replica);
assertEquals(0, bulkClient.getBulkProcessor().getBulkMetric().getFailed().getCount()); assertEquals(0, bulkClient.getBulkProcessor().getBulkMetric().getFailed().getCount());
assertEquals(6, bulkClient.getBulkProcessor().getBulkMetric().getSucceeded().getCount()); assertEquals(6, bulkClient.getBulkProcessor().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkProcessor().getLastBulkError() != null) { if (bulkClient.getBulkProcessor().getLastBulkError() != null) {

View file

@ -3,21 +3,12 @@ package org.xbib.elx.http.test;
import org.apache.logging.log4j.Level; import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.elasticsearch.ElasticsearchTimeoutException;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthAction;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoAction; import org.elasticsearch.action.admin.cluster.node.info.NodesInfoAction;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoRequest; import org.elasticsearch.action.admin.cluster.node.info.NodesInfoRequest;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse; import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.client.ElasticsearchClient; import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress; import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.http.HttpInfo; import org.elasticsearch.http.HttpInfo;
import org.elasticsearch.node.Node; import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeValidationException; import org.elasticsearch.node.NodeValidationException;
@ -38,18 +29,10 @@ import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor; import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes; import java.nio.file.attribute.BasicFileAttributes;
import java.util.Collections; import java.util.Collections;
import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map;
import java.util.Random; import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
/**
* Junit 5 extension for testing Elasticsearch.
* The extension will be instantiated as a singleton.
* For parallel test method executions, for example in gradle, it requires a helper class
* to ensure different ES homes/clusters for each run.
*/
public class TestExtension implements ParameterResolver, BeforeEachCallback, AfterEachCallback { public class TestExtension implements ParameterResolver, BeforeEachCallback, AfterEachCallback {
private static final Logger logger = LogManager.getLogger("test"); private static final Logger logger = LogManager.getLogger("test");
@ -93,21 +76,6 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
helper.httpHost = address.address().getHostName(); helper.httpHost = address.address().getHostName();
helper.httpPort = address.address().getPort(); helper.httpPort = address.address().getPort();
logger.log(Level.INFO, "http host = " + helper.httpHost + " port = " + helper.httpPort); logger.log(Level.INFO, "http host = " + helper.httpHost + " port = " + helper.httpPort);
try {
ClusterHealthResponse healthResponse = helper.client().execute(ClusterHealthAction.INSTANCE,
new ClusterHealthRequest().waitForStatus(ClusterHealthStatus.GREEN)
.timeout(TimeValue.timeValueSeconds(30))).actionGet();
if (healthResponse != null && healthResponse.isTimedOut()) {
throw new IOException("cluster state is " + healthResponse.getStatus().name()
+ ", from here on, everything will fail!");
}
} catch (ElasticsearchTimeoutException e) {
throw new IOException("cluster does not respond to health request, cowardly refusing to continue");
}
ClusterStateRequest clusterStateRequest = new ClusterStateRequest().all();
ClusterStateResponse clusterStateResponse =
helper.client().execute(ClusterStateAction.INSTANCE, clusterStateRequest).actionGet();
logger.info("cluster name = {}", clusterStateResponse.getClusterName().value());
} }
@Override @Override
@ -183,12 +151,14 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
return cluster; return cluster;
} }
Settings getHttpSettings() { Settings getClientSettings() {
return Settings.builder() return Settings.builder()
.put("cluster.name", getClusterName()) .put("cluster.name", getClusterName())
.put("path.home", getHome()) .put("path.home", getHome())
.put("host", httpHost) .put("host", httpHost)
.put("port", httpPort) .put("port", httpPort)
.put("cluster.target_health", "YELLOW")
.put("cluster.target_health_timeout", "1m")
.build(); .build();
} }

View file

@ -2,15 +2,11 @@ package org.xbib.elx.node.test;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.api.extension.ExtendWith;
import org.xbib.elx.api.IndexDefinition; import org.xbib.elx.api.IndexDefinition;
import org.xbib.elx.common.ClientBuilder; import org.xbib.elx.common.ClientBuilder;
import org.xbib.elx.common.DefaultIndexDefinition; import org.xbib.elx.common.DefaultIndexDefinition;
import org.xbib.elx.node.NodeAdminClient;
import org.xbib.elx.node.NodeAdminClientProvider;
import org.xbib.elx.node.NodeBulkClient; import org.xbib.elx.node.NodeBulkClient;
import org.xbib.elx.node.NodeBulkClientProvider; import org.xbib.elx.node.NodeBulkClientProvider;
@ -39,7 +35,7 @@ class BulkClientTest {
void testNewIndex() throws Exception { void testNewIndex() throws Exception {
try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client()) try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client())
.setBulkClientProvider(NodeBulkClientProvider.class) .setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
@ -50,7 +46,7 @@ class BulkClientTest {
void testSingleDoc() throws Exception { void testSingleDoc() throws Exception {
try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client()) try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client())
.setBulkClientProvider(NodeBulkClientProvider.class) .setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
@ -69,7 +65,7 @@ class BulkClientTest {
long numactions = ACTIONS; long numactions = ACTIONS;
try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client()) try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client())
.setBulkClientProvider(NodeBulkClientProvider.class) .setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
@ -97,7 +93,7 @@ class BulkClientTest {
long timeout = 120L; long timeout = 120L;
try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client()) try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client())
.setBulkClientProvider(NodeBulkClientProvider.class) .setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
indexDefinition.setStartBulkRefreshSeconds(0); indexDefinition.setStartBulkRefreshSeconds(0);

View file

@ -37,7 +37,7 @@ class DuplicateIDTest {
long numactions = ACTIONS; long numactions = ACTIONS;
try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client()) try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client())
.setBulkClientProvider(NodeBulkClientProvider.class) .setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings()) .put(helper.getClientSettings())
.put(Parameters.BULK_MAX_ACTIONS_PER_REQUEST.getName(), MAX_ACTIONS_PER_REQUEST) .put(Parameters.BULK_MAX_ACTIONS_PER_REQUEST.getName(), MAX_ACTIONS_PER_REQUEST)
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");

View file

@ -40,11 +40,11 @@ class IndexPruneTest {
void testPrune() throws IOException { void testPrune() throws IOException {
try (NodeAdminClient adminClient = ClientBuilder.builder(helper.client()) try (NodeAdminClient adminClient = ClientBuilder.builder(helper.client())
.setAdminClientProvider(NodeAdminClientProvider.class) .setAdminClientProvider(NodeAdminClientProvider.class)
.put(helper.getNodeSettings()) .put(helper.getClientSettings())
.build(); .build();
NodeBulkClient bulkClient = ClientBuilder.builder(helper.client()) NodeBulkClient bulkClient = ClientBuilder.builder(helper.client())
.setBulkClientProvider(NodeBulkClientProvider.class) .setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
indexDefinition.setIndex("test_prune"); indexDefinition.setIndex("test_prune");

View file

@ -39,11 +39,11 @@ class IndexShiftTest {
void testIndexShift() throws Exception { void testIndexShift() throws Exception {
try (NodeAdminClient adminClient = ClientBuilder.builder(helper.client()) try (NodeAdminClient adminClient = ClientBuilder.builder(helper.client())
.setAdminClientProvider(NodeAdminClientProvider.class) .setAdminClientProvider(NodeAdminClientProvider.class)
.put(helper.getNodeSettings()) .put(helper.getClientSettings())
.build(); .build();
NodeBulkClient bulkClient = ClientBuilder.builder(helper.client()) NodeBulkClient bulkClient = ClientBuilder.builder(helper.client())
.setBulkClientProvider(NodeBulkClientProvider.class) .setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
indexDefinition.setFullIndexName("test_shift"); indexDefinition.setFullIndexName("test_shift");

View file

@ -43,7 +43,7 @@ class SearchTest {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client()) try (NodeBulkClient bulkClient = ClientBuilder.builder(helper.client())
.setBulkClientProvider(NodeBulkClientProvider.class) .setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings()) .put(helper.getClientSettings())
.put(Parameters.BULK_MAX_ACTIONS_PER_REQUEST.getName(), MAX_ACTIONS_PER_REQUEST) .put(Parameters.BULK_MAX_ACTIONS_PER_REQUEST.getName(), MAX_ACTIONS_PER_REQUEST)
.build()) { .build()) {
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
@ -64,7 +64,7 @@ class SearchTest {
} }
try (NodeSearchClient searchClient = ClientBuilder.builder(helper.client()) try (NodeSearchClient searchClient = ClientBuilder.builder(helper.client())
.setSearchClientProvider(NodeSearchClientProvider.class) .setSearchClientProvider(NodeSearchClientProvider.class)
.put(helper.getNodeSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
// test stream count // test stream count
Stream<SearchHit> stream = searchClient.search(qb -> qb Stream<SearchHit> stream = searchClient.search(qb -> qb

View file

@ -37,11 +37,11 @@ class SmokeTest {
void smokeTest() throws Exception { void smokeTest() throws Exception {
try (NodeAdminClient adminClient = ClientBuilder.builder(helper.client()) try (NodeAdminClient adminClient = ClientBuilder.builder(helper.client())
.setAdminClientProvider(NodeAdminClientProvider.class) .setAdminClientProvider(NodeAdminClientProvider.class)
.put(helper.getNodeSettings()) .put(helper.getClientSettings())
.build(); .build();
NodeBulkClient bulkClient = ClientBuilder.builder(helper.client()) NodeBulkClient bulkClient = ClientBuilder.builder(helper.client())
.setBulkClientProvider(NodeBulkClientProvider.class) .setBulkClientProvider(NodeBulkClientProvider.class)
.put(helper.getNodeSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = IndexDefinition indexDefinition =
new DefaultIndexDefinition(adminClient, "test_smoke", "doc", Settings.EMPTY); new DefaultIndexDefinition(adminClient, "test_smoke", "doc", Settings.EMPTY);
@ -52,21 +52,20 @@ class SmokeTest {
indexDefinition.setType("doc"); indexDefinition.setType("doc");
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest
bulkClient.waitForResponses(30, TimeUnit.SECONDS); assertTrue(bulkClient.waitForResponses(30, TimeUnit.SECONDS));
adminClient.checkMapping(indexDefinition); adminClient.checkMapping(indexDefinition);
bulkClient.update(indexDefinition, "1", "{ \"name\" : \"Another name\"}"); bulkClient.update(indexDefinition, "1", "{ \"name\" : \"Another name\"}");
bulkClient.delete(indexDefinition, "1"); bulkClient.delete(indexDefinition, "1");
bulkClient.waitForResponses(30, TimeUnit.SECONDS); assertTrue(bulkClient.waitForResponses(30, TimeUnit.SECONDS));
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}");
bulkClient.delete(indexDefinition, "1"); bulkClient.delete(indexDefinition, "1");
bulkClient.waitForResponses(30, TimeUnit.SECONDS); assertTrue(bulkClient.waitForResponses(30, TimeUnit.SECONDS));
adminClient.deleteIndex(indexDefinition); adminClient.deleteIndex(indexDefinition);
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}");
bulkClient.waitForResponses(30, TimeUnit.SECONDS); assertTrue(bulkClient.waitForResponses(30, TimeUnit.SECONDS));
adminClient.updateReplicaLevel(indexDefinition, 2); adminClient.updateReplicaLevel(indexDefinition, 1);
int replica = adminClient.getReplicaLevel(indexDefinition); assertEquals(1, adminClient.getReplicaLevel(indexDefinition));
assertEquals(2, replica);
assertEquals(0, bulkClient.getBulkProcessor().getBulkMetric().getFailed().getCount()); assertEquals(0, bulkClient.getBulkProcessor().getBulkMetric().getFailed().getCount());
assertEquals(6, bulkClient.getBulkProcessor().getBulkMetric().getSucceeded().getCount()); assertEquals(6, bulkClient.getBulkProcessor().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkProcessor().getLastBulkError() != null) { if (bulkClient.getBulkProcessor().getLastBulkError() != null) {

View file

@ -2,21 +2,12 @@ package org.xbib.elx.node.test;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.elasticsearch.ElasticsearchTimeoutException;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthAction;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoAction; import org.elasticsearch.action.admin.cluster.node.info.NodesInfoAction;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoRequest; import org.elasticsearch.action.admin.cluster.node.info.NodesInfoRequest;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse; import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.client.ElasticsearchClient; import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress; import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.node.Node; import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeValidationException; import org.elasticsearch.node.NodeValidationException;
import org.elasticsearch.plugins.Plugin; import org.elasticsearch.plugins.Plugin;
@ -36,18 +27,10 @@ import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor; import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes; import java.nio.file.attribute.BasicFileAttributes;
import java.util.Collections; import java.util.Collections;
import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map;
import java.util.Random; import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
/**
* Junit 5 extension for testing Elasticsearch.
* The extension will be instantiated as a singleton.
* For parallel test method executions, for example in gradle, it requires a helper class
* to ensure different ES homes/clusters for each run.
*/
public class TestExtension implements ParameterResolver, BeforeEachCallback, AfterEachCallback { public class TestExtension implements ParameterResolver, BeforeEachCallback, AfterEachCallback {
private static final Logger logger = LogManager.getLogger("test"); private static final Logger logger = LogManager.getLogger("test");
@ -89,21 +72,6 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
helper.host = address.address().getHostName(); helper.host = address.address().getHostName();
helper.port = address.address().getPort(); helper.port = address.address().getPort();
logger.info("host = " + helper.host + " port = " + helper.port); logger.info("host = " + helper.host + " port = " + helper.port);
try {
ClusterHealthResponse healthResponse = helper.client().execute(ClusterHealthAction.INSTANCE,
new ClusterHealthRequest().waitForStatus(ClusterHealthStatus.GREEN)
.timeout(TimeValue.timeValueSeconds(30))).actionGet();
if (healthResponse != null && healthResponse.isTimedOut()) {
throw new IOException("cluster state is " + healthResponse.getStatus().name()
+ ", from here on, everything will fail!");
}
} catch (ElasticsearchTimeoutException e) {
throw new IOException("cluster does not respond to health request, cowardly refusing to continue");
}
ClusterStateRequest clusterStateRequest = new ClusterStateRequest().all();
ClusterStateResponse clusterStateResponse =
helper.client().execute(ClusterStateAction.INSTANCE, clusterStateRequest).actionGet();
logger.info("cluster up, name = {}", clusterStateResponse.getClusterName().value());
} }
@Override @Override
@ -179,13 +147,15 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
return cluster; return cluster;
} }
Settings getNodeSettings() { Settings getClientSettings() {
return Settings.builder() return Settings.builder()
.put("cluster.name", getClusterName()) .put("cluster.name", getClusterName())
.put("path.home", getHome()) .put("path.home", getHome())
.put("node.max_local_storage_nodes", 2) .put("node.max_local_storage_nodes", 2)
.put("cluster.initial_master_nodes", "1") .put("cluster.initial_master_nodes", "1")
.put("discovery.seed_hosts", "127.0.0.1:9300") .put("discovery.seed_hosts", "127.0.0.1:9300")
.put("cluster.target_health", "YELLOW")
.put("cluster.target_health_timeout", "1m")
.build(); .build();
} }

View file

@ -5,8 +5,6 @@ import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
import org.xbib.elx.common.AbstractAdminClient; import org.xbib.elx.common.AbstractAdminClient;
import java.io.IOException;
/** /**
* Transport admin client. * Transport admin client.
*/ */
@ -20,12 +18,12 @@ public class TransportAdminClient extends AbstractAdminClient {
} }
@Override @Override
public ElasticsearchClient createClient(Settings settings) throws IOException { public ElasticsearchClient createClient(Settings settings) {
return helper.createClient(settings); return helper.createClient(settings);
} }
@Override @Override
public void init(Settings settings) throws IOException { public void init(Settings settings) {
super.init(settings); super.init(settings);
helper.init((TransportClient) getClient(), settings); helper.init((TransportClient) getClient(), settings);
} }

View file

@ -4,7 +4,6 @@ import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
import org.xbib.elx.common.AbstractBulkClient; import org.xbib.elx.common.AbstractBulkClient;
import java.io.IOException;
/** /**
* Transport search client with additional methods. * Transport search client with additional methods.
@ -19,12 +18,12 @@ public class TransportBulkClient extends AbstractBulkClient {
} }
@Override @Override
public ElasticsearchClient createClient(Settings settings) throws IOException { public ElasticsearchClient createClient(Settings settings) {
return helper.createClient(settings); return helper.createClient(settings);
} }
@Override @Override
public void init(Settings settings) throws IOException { public void init(Settings settings) {
super.init(settings); super.init(settings);
helper.init((TransportClient) getClient(), settings); helper.init((TransportClient) getClient(), settings);
} }

View file

@ -1,5 +1,6 @@
package org.xbib.elx.transport; package org.xbib.elx.transport;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.elasticsearch.Version; import org.elasticsearch.Version;
@ -55,7 +56,7 @@ public class TransportClientHelper {
} }
} }
public void init(TransportClient transportClient, Settings settings) throws IOException { public void init(TransportClient transportClient, Settings settings) {
Collection<TransportAddress> addrs = findAddresses(settings); Collection<TransportAddress> addrs = findAddresses(settings);
if (!connect(transportClient, addrs, settings.getAsBoolean("autodiscover", false))) { if (!connect(transportClient, addrs, settings.getAsBoolean("autodiscover", false))) {
throw new NoNodeAvailableException("no cluster nodes available, check settings = " throw new NoNodeAvailableException("no cluster nodes available, check settings = "
@ -63,7 +64,7 @@ public class TransportClientHelper {
} }
} }
private Collection<TransportAddress> findAddresses(Settings settings) throws IOException { private Collection<TransportAddress> findAddresses(Settings settings) {
final int defaultPort = settings.getAsInt("port", 9300); final int defaultPort = settings.getAsInt("port", 9300);
Collection<TransportAddress> addresses = new ArrayList<>(); Collection<TransportAddress> addresses = new ArrayList<>();
for (String hostname : settings.getAsList("host")) { for (String hostname : settings.getAsList("host")) {
@ -71,19 +72,27 @@ public class TransportClientHelper {
if (splitHost.length == 2) { if (splitHost.length == 2) {
try { try {
String host = splitHost[0]; String host = splitHost[0];
InetAddress inetAddress = NetworkUtils.resolveInetAddress(host, null); try {
int port = Integer.parseInt(splitHost[1]); InetAddress inetAddress = NetworkUtils.resolveInetAddress(host, null);
TransportAddress address = new TransportAddress(inetAddress, port); int port = Integer.parseInt(splitHost[1]);
addresses.add(address); TransportAddress address = new TransportAddress(inetAddress, port);
addresses.add(address);
} catch (IOException e) {
logger.log(Level.WARN, e.getMessage(), e);
}
} catch (NumberFormatException e) { } catch (NumberFormatException e) {
logger.warn(e.getMessage(), e); logger.warn(e.getMessage(), e);
} }
} }
if (splitHost.length == 1) { if (splitHost.length == 1) {
String host = splitHost[0]; String host = splitHost[0];
InetAddress inetAddress = NetworkUtils.resolveInetAddress(host, null); try {
TransportAddress address = new TransportAddress(inetAddress, defaultPort); InetAddress inetAddress = NetworkUtils.resolveInetAddress(host, null);
addresses.add(address); TransportAddress address = new TransportAddress(inetAddress, defaultPort);
addresses.add(address);
} catch (IOException e) {
logger.log(Level.WARN, e.getMessage(), e);
}
} }
} }
return addresses; return addresses;

View file

@ -4,7 +4,6 @@ import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
import org.xbib.elx.common.AbstractSearchClient; import org.xbib.elx.common.AbstractSearchClient;
import java.io.IOException;
/** /**
* Transport search client with additional methods. * Transport search client with additional methods.
@ -19,12 +18,12 @@ public class TransportSearchClient extends AbstractSearchClient {
} }
@Override @Override
public ElasticsearchClient createClient(Settings settings) throws IOException { public ElasticsearchClient createClient(Settings settings) {
return helper.createClient(settings); return helper.createClient(settings);
} }
@Override @Override
public void init(Settings settings) throws IOException { public void init(Settings settings) {
super.init(settings); super.init(settings);
helper.init((TransportClient) getClient(), settings); helper.init((TransportClient) getClient(), settings);
} }

View file

@ -35,7 +35,7 @@ class BulkClientTest {
void testNewIndex() throws Exception { void testNewIndex() throws Exception {
try (TransportBulkClient bulkClient = ClientBuilder.builder() try (TransportBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(TransportBulkClientProvider.class) .setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
@ -46,7 +46,7 @@ class BulkClientTest {
void testSingleDoc() throws Exception { void testSingleDoc() throws Exception {
try (TransportBulkClient bulkClient = ClientBuilder.builder() try (TransportBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(TransportBulkClientProvider.class) .setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
@ -65,7 +65,7 @@ class BulkClientTest {
long numactions = ACTIONS; long numactions = ACTIONS;
try (TransportBulkClient bulkClient = ClientBuilder.builder() try (TransportBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(TransportBulkClientProvider.class) .setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
@ -93,7 +93,7 @@ class BulkClientTest {
final long timeout = 120L; final long timeout = 120L;
try (TransportBulkClient bulkClient = ClientBuilder.builder() try (TransportBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(TransportBulkClientProvider.class) .setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);

View file

@ -1,6 +1,7 @@
package org.xbib.elx.transport.test; package org.xbib.elx.transport.test;
import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.index.query.QueryBuilders;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.xbib.elx.api.IndexDefinition; import org.xbib.elx.api.IndexDefinition;
import org.xbib.elx.common.ClientBuilder; import org.xbib.elx.common.ClientBuilder;
@ -14,6 +15,7 @@ import java.nio.file.Files;
import java.nio.file.Paths; import java.nio.file.Paths;
import java.util.stream.Stream; import java.util.stream.Stream;
@Disabled
class DumpIDTest { class DumpIDTest {
@Test @Test

View file

@ -38,7 +38,7 @@ class DuplicateIDTest {
try (TransportBulkClient bulkClient = ClientBuilder.builder() try (TransportBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(TransportBulkClientProvider.class) .setBulkClientProvider(TransportBulkClientProvider.class)
.put(Parameters.BULK_MAX_ACTIONS_PER_REQUEST.getName(), MAX_ACTIONS_PER_REQUEST) .put(Parameters.BULK_MAX_ACTIONS_PER_REQUEST.getName(), MAX_ACTIONS_PER_REQUEST)
.put(helper.getTransportSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);

View file

@ -40,11 +40,11 @@ class IndexPruneTest {
void testPrune() throws IOException { void testPrune() throws IOException {
try (TransportAdminClient adminClient = ClientBuilder.builder() try (TransportAdminClient adminClient = ClientBuilder.builder()
.setAdminClientProvider(TransportAdminClientProvider.class) .setAdminClientProvider(TransportAdminClientProvider.class)
.put(helper.getTransportSettings()) .put(helper.getClientSettings())
.build(); .build();
TransportBulkClient bulkClient = ClientBuilder.builder() TransportBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(TransportBulkClientProvider.class) .setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
indexDefinition.setIndex("test_prune"); indexDefinition.setIndex("test_prune");

View file

@ -39,11 +39,11 @@ class IndexShiftTest {
void testIndexShift() throws Exception { void testIndexShift() throws Exception {
try (TransportAdminClient adminClient = ClientBuilder.builder() try (TransportAdminClient adminClient = ClientBuilder.builder()
.setAdminClientProvider(TransportAdminClientProvider.class) .setAdminClientProvider(TransportAdminClientProvider.class)
.put(helper.getTransportSettings()) .put(helper.getClientSettings())
.build(); .build();
TransportBulkClient bulkClient = ClientBuilder.builder() TransportBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(TransportBulkClientProvider.class) .setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
indexDefinition.setFullIndexName("test_shift"); indexDefinition.setFullIndexName("test_shift");

View file

@ -40,7 +40,7 @@ class SearchTest {
IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc"); IndexDefinition indexDefinition = new DefaultIndexDefinition("test", "doc");
try (TransportBulkClient bulkClient = ClientBuilder.builder() try (TransportBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(TransportBulkClientProvider.class) .setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
bulkClient.startBulk(indexDefinition); bulkClient.startBulk(indexDefinition);
@ -59,7 +59,7 @@ class SearchTest {
} }
try (TransportSearchClient searchClient = ClientBuilder.builder() try (TransportSearchClient searchClient = ClientBuilder.builder()
.setSearchClientProvider(TransportSearchClientProvider.class) .setSearchClientProvider(TransportSearchClientProvider.class)
.put(helper.getTransportSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
// test stream count // test stream count
Stream<SearchHit> stream = searchClient.search(qb -> qb Stream<SearchHit> stream = searchClient.search(qb -> qb

View file

@ -37,11 +37,11 @@ class SmokeTest {
void smokeTest() throws Exception { void smokeTest() throws Exception {
try (TransportAdminClient adminClient = ClientBuilder.builder() try (TransportAdminClient adminClient = ClientBuilder.builder()
.setAdminClientProvider(TransportAdminClientProvider.class) .setAdminClientProvider(TransportAdminClientProvider.class)
.put(helper.getTransportSettings()) .put(helper.getClientSettings())
.build(); .build();
TransportBulkClient bulkClient = ClientBuilder.builder() TransportBulkClient bulkClient = ClientBuilder.builder()
.setBulkClientProvider(TransportBulkClientProvider.class) .setBulkClientProvider(TransportBulkClientProvider.class)
.put(helper.getTransportSettings()) .put(helper.getClientSettings())
.build()) { .build()) {
IndexDefinition indexDefinition = IndexDefinition indexDefinition =
new DefaultIndexDefinition(adminClient, "test", "doc", Settings.EMPTY); new DefaultIndexDefinition(adminClient, "test", "doc", Settings.EMPTY);
@ -52,21 +52,20 @@ class SmokeTest {
indexDefinition.setType("doc"); indexDefinition.setType("doc");
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); // single doc ingest
bulkClient.waitForResponses(30, TimeUnit.SECONDS); assertTrue(bulkClient.waitForResponses(30, TimeUnit.SECONDS));
adminClient.checkMapping(indexDefinition); adminClient.checkMapping(indexDefinition);
bulkClient.update(indexDefinition, "1", "{ \"name\" : \"Another name\"}"); bulkClient.update(indexDefinition, "1", "{ \"name\" : \"Another name\"}");
bulkClient.delete(indexDefinition, "1"); bulkClient.delete(indexDefinition, "1");
bulkClient.waitForResponses(30, TimeUnit.SECONDS); assertTrue(bulkClient.waitForResponses(30, TimeUnit.SECONDS));
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}");
bulkClient.delete(indexDefinition, "1"); bulkClient.delete(indexDefinition, "1");
bulkClient.waitForResponses(30, TimeUnit.SECONDS); assertTrue(bulkClient.waitForResponses(30, TimeUnit.SECONDS));
adminClient.deleteIndex(indexDefinition); adminClient.deleteIndex(indexDefinition);
bulkClient.newIndex(indexDefinition); bulkClient.newIndex(indexDefinition);
bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}"); bulkClient.index(indexDefinition, "1", true, "{ \"name\" : \"Hello World\"}");
bulkClient.waitForResponses(30, TimeUnit.SECONDS); assertTrue(bulkClient.waitForResponses(30, TimeUnit.SECONDS));
adminClient.updateReplicaLevel(indexDefinition, 2); adminClient.updateReplicaLevel(indexDefinition, 1);
int replica = adminClient.getReplicaLevel(indexDefinition); assertEquals(1, adminClient.getReplicaLevel(indexDefinition));
assertEquals(2, replica);
assertEquals(0, bulkClient.getBulkProcessor().getBulkMetric().getFailed().getCount()); assertEquals(0, bulkClient.getBulkProcessor().getBulkMetric().getFailed().getCount());
assertEquals(6, bulkClient.getBulkProcessor().getBulkMetric().getSucceeded().getCount()); assertEquals(6, bulkClient.getBulkProcessor().getBulkMetric().getSucceeded().getCount());
if (bulkClient.getBulkProcessor().getLastBulkError() != null) { if (bulkClient.getBulkProcessor().getLastBulkError() != null) {

View file

@ -2,21 +2,12 @@ package org.xbib.elx.transport.test;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.elasticsearch.ElasticsearchTimeoutException;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthAction;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoAction; import org.elasticsearch.action.admin.cluster.node.info.NodesInfoAction;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoRequest; import org.elasticsearch.action.admin.cluster.node.info.NodesInfoRequest;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse; import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.client.ElasticsearchClient; import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress; import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.node.Node; import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeValidationException; import org.elasticsearch.node.NodeValidationException;
import org.elasticsearch.plugins.Plugin; import org.elasticsearch.plugins.Plugin;
@ -86,21 +77,6 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
TransportAddress address = response.getNodes().get(0).getNode().getAddress(); TransportAddress address = response.getNodes().get(0).getNode().getAddress();
helper.host = address.address().getHostName(); helper.host = address.address().getHostName();
helper.port = address.address().getPort(); helper.port = address.address().getPort();
try {
ClusterHealthResponse healthResponse = helper.client().execute(ClusterHealthAction.INSTANCE,
new ClusterHealthRequest().waitForStatus(ClusterHealthStatus.GREEN)
.timeout(TimeValue.timeValueSeconds(30))).actionGet();
if (healthResponse != null && healthResponse.isTimedOut()) {
throw new IOException("cluster state is " + healthResponse.getStatus().name()
+ ", from here on, everything will fail!");
}
} catch (ElasticsearchTimeoutException e) {
throw new IOException("cluster does not respond to health request, cowardly refusing to continue");
}
ClusterStateRequest clusterStateRequest = new ClusterStateRequest().all();
ClusterStateResponse clusterStateResponse =
helper.client().execute(ClusterStateAction.INSTANCE, clusterStateRequest).actionGet();
logger.info("cluster name = {}", clusterStateResponse.getClusterName().value());
} }
@Override @Override
@ -176,12 +152,14 @@ public class TestExtension implements ParameterResolver, BeforeEachCallback, Aft
return cluster; return cluster;
} }
Settings getTransportSettings() { Settings getClientSettings() {
return Settings.builder() return Settings.builder()
.put("cluster.name", cluster) .put("cluster.name", cluster)
.put("path.home", getHome()) .put("path.home", getHome())
.put("host", host) .put("host", host)
.put("port", port) .put("port", port)
.put("cluster.target_health", "YELLOW")
.put("cluster.target_health_timeout", "1m")
.build(); .build();
} }

View file

@ -1,6 +1,6 @@
group = org.xbib group = org.xbib
name = elx name = elx
version = 7.10.2.3 version = 7.10.2.4
gradle.wrapper.version = 6.6.1 gradle.wrapper.version = 6.6.1
xbib-metrics.version = 2.1.0 xbib-metrics.version = 2.1.0