diff --git a/.github/workflows/master-build.yml b/.github/workflows/master-build.yml index d73bb3a693..5f525ab718 100644 --- a/.github/workflows/master-build.yml +++ b/.github/workflows/master-build.yml @@ -19,7 +19,7 @@ on: branches: [master] pull_request: types: [opened, synchronize, reopened] - branches: [master] + branches: [master, nci-develop] jobs: javadoc: strategy: @@ -104,6 +104,8 @@ jobs: - 'src/testresources/**' plugins: - 'src/plugin/**' + indexer_plugins: + - 'src/plugin/indexer-*/**' buildconf: - 'build.xml' - 'ivy/ivy.xml' @@ -120,6 +122,10 @@ jobs: - name: test plugins if: ${{ steps.filter.outputs.plugins == 'true' && steps.filter.outputs.core == 'false' && steps.filter.outputs.buildconf == 'false' }} run: ant clean test-plugins -buildfile build.xml + # run indexer integration tests when indexer plugin files change (Docker required, ubuntu-latest only) + - name: test indexer integration + if: ${{ steps.filter.outputs.indexer_plugins == 'true' && matrix.os == 'ubuntu-latest' }} + run: ant clean test-indexer-integration -buildfile build.xml - name: Check for test results id: check_tests if: always() && matrix.os == 'ubuntu-latest' @@ -139,4 +145,4 @@ jobs: path: | ./build/test/TEST-*.xml ./build/**/test/TEST-*.xml - retention-days: 1 \ No newline at end of file + retention-days: 1 diff --git a/LICENSE-binary b/LICENSE-binary index addc4a2824..23fbec535a 100644 --- a/LICENSE-binary +++ b/LICENSE-binary @@ -208,6 +208,7 @@ This product bundles some components that are also licensed under the Apache License Version 2.0: +co.elastic.clients:elasticsearch-java com.amazonaws:aws-java-sdk-cloudsearch com.amazonaws:aws-java-sdk-core com.beust:jcommander @@ -494,20 +495,7 @@ org.eclipse.jetty.toolchain:jetty-servlet-api org.eclipse.jetty.websocket:websocket-api org.eclipse.jetty.websocket:websocket-client org.eclipse.jetty.websocket:websocket-common -org.elasticsearch:elasticsearch -org.elasticsearch:elasticsearch-cli -org.elasticsearch:elasticsearch-core -org.elasticsearch:elasticsearch-geo -org.elasticsearch:elasticsearch-secure-sm -org.elasticsearch:elasticsearch-x-content -org.elasticsearch:jna org.elasticsearch.client:elasticsearch-rest-client -org.elasticsearch.client:elasticsearch-rest-high-level-client -org.elasticsearch.plugin:aggs-matrix-stats-client -org.elasticsearch.plugin:lang-mustache-client -org.elasticsearch.plugin:mapper-extras-client -org.elasticsearch.plugin:parent-join-client -org.elasticsearch.plugin:rank-eval-client org.gagravarr:vorbis-java-core org.gagravarr:vorbis-java-tika org.jetbrains:annotations @@ -721,6 +709,8 @@ Eclipse Public License - Version 2.0 (licenses-binary/LICENSE-eclipse-public-license---version-2.0.txt) +jakarta.json:jakarta.json-api +org.eclipse.parsson:parsson org.eclipse.jetty:jetty-http org.eclipse.jetty:jetty-io org.eclipse.jetty:jetty-security diff --git a/NOTICE-binary b/NOTICE-binary index 412ce7d38e..11af6751da 100644 --- a/NOTICE-binary +++ b/NOTICE-binary @@ -341,6 +341,10 @@ AOP alliance (http://aopalliance.sourceforge.net) - license: Public Domain (licenses-binary/LICENSE-public-domain.txt) +# co.elastic.clients:elasticsearch-java +Elasticsearch Java API Client (https://github.com/elastic/elasticsearch-java/) +- license: The Apache Software License, Version 2.0 + # com.adobe.xmp:xmpcore Adobe XMPCore (https://www.adobe.com/devnet/xmp/library/eula-xmp-library-java.html) - license: The BSD 3-Clause License (BSD3) @@ -985,6 +989,11 @@ Jakarta Annotations API (https://projects.eclipse.org/projects/ee4j.ca) - license: EPL 2.0 (licenses-binary/LICENSE-epl-2.0.txt) +# jakarta.json:jakarta.json-api +Jakarta JSON Processing API (https://github.com/eclipse-ee4j/jsonp) +- license: Eclipse Public License 2.0 + (licenses-binary/LICENSE-epl-2.0.txt) + # jakarta.jws:jakarta.jws-api Jakarta Web Services Metadata API (https://github.com/eclipse-ee4j/jws-api) - license: Eclipse Distribution License - v 1.0 @@ -1254,50 +1263,14 @@ Jetty :: Websocket :: Client Jetty :: Websocket :: Common - license: Apache Software License - Version 2.0 -# org.elasticsearch:elasticsearch -The Apache Software License, Version 2.0 (https://github.com/elastic/elasticsearch) -- license: The Apache Software License, Version 2.0 -# org.elasticsearch:elasticsearch-cli -The Apache Software License, Version 2.0 (https://github.com/elastic/elasticsearch) -- license: The Apache Software License, Version 2.0 -# org.elasticsearch:elasticsearch-core -The Apache Software License, Version 2.0 (https://github.com/elastic/elasticsearch) -- license: The Apache Software License, Version 2.0 -# org.elasticsearch:elasticsearch-geo -The Apache Software License, Version 2.0 (https://github.com/elastic/elasticsearch) -- license: The Apache Software License, Version 2.0 -# org.elasticsearch:elasticsearch-secure-sm -The Apache Software License, Version 2.0 (https://github.com/elastic/elasticsearch) -- license: The Apache Software License, Version 2.0 -# org.elasticsearch:elasticsearch-x-content -The Apache Software License, Version 2.0 (https://github.com/elastic/elasticsearch) -- license: The Apache Software License, Version 2.0 -# org.elasticsearch:jna -Elastic JNA Distribution (https://github.com/java-native-access/jna) -- license: Apache License, Version 2.0 +# org.eclipse.parsson:parsson +Eclipse Parsson +- license: Eclipse Public License 2.0 + (licenses-binary/LICENSE-epl-2.0.txt) # org.elasticsearch.client:elasticsearch-rest-client The Apache Software License, Version 2.0 (https://github.com/elastic/elasticsearch) - license: The Apache Software License, Version 2.0 -# org.elasticsearch.client:elasticsearch-rest-high-level-client -The Apache Software License, Version 2.0 (https://github.com/elastic/elasticsearch) -- license: The Apache Software License, Version 2.0 - -# org.elasticsearch.plugin:aggs-matrix-stats-client -The Apache Software License, Version 2.0 (https://github.com/elastic/elasticsearch) -- license: The Apache Software License, Version 2.0 -# org.elasticsearch.plugin:lang-mustache-client -The Apache Software License, Version 2.0 (https://github.com/elastic/elasticsearch) -- license: The Apache Software License, Version 2.0 -# org.elasticsearch.plugin:mapper-extras-client -The Apache Software License, Version 2.0 (https://github.com/elastic/elasticsearch) -- license: The Apache Software License, Version 2.0 -# org.elasticsearch.plugin:parent-join-client -The Apache Software License, Version 2.0 (https://github.com/elastic/elasticsearch) -- license: The Apache Software License, Version 2.0 -# org.elasticsearch.plugin:rank-eval-client -The Apache Software License, Version 2.0 (https://github.com/elastic/elasticsearch) -- license: The Apache Software License, Version 2.0 # org.fusesource.leveldbjni:leveldbjni-all ${project.artifactId} diff --git a/build.xml b/build.xml index d97918aa28..a342e94c3e 100644 --- a/build.xml +++ b/build.xml @@ -534,6 +534,10 @@ + + + + diff --git a/docker/Dockerfile b/docker/Dockerfile index 2eb218bade..aec5943282 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -42,16 +42,19 @@ WORKDIR /root/ # Install dependencies RUN apk update -RUN apk --no-cache add apache-ant bash git openjdk11 supervisor +RUN apk --no-cache add apache-ant bash git openjdk17 supervisor # Establish environment variables -RUN echo 'export JAVA_HOME=/usr/lib/jvm/java-11-openjdk' >> $HOME/.bashrc -RUN echo 'export JAVA_HOME=/usr/lib/jvm/java-11-openjdk' >> $HOME/.ashrc -ENV JAVA_HOME='/usr/lib/jvm/java-11-openjdk' +RUN echo 'export JAVA_HOME=/usr/lib/jvm/java-17-openjdk' >> $HOME/.bashrc +RUN echo 'export JAVA_HOME=/usr/lib/jvm/java-17-openjdk' >> $HOME/.ashrc +RUN echo 'export NUTCH_JAVA_HOME=/usr/lib/jvm/java-17-openjdk' >> $HOME/.bashrc +RUN echo 'export NUTCH_JAVA_HOME=/usr/lib/jvm/java-17-openjdk' >> $HOME/.ashrc +ENV JAVA_HOME='/usr/lib/jvm/java-17-openjdk' +ENV NUTCH_JAVA_HOME='/usr/lib/jvm/java-17-openjdk' ENV NUTCH_HOME='/root/nutch_source/runtime/local' -# Checkout and build the Nutch master branch (1.x) -RUN git clone https://github.com/apache/nutch.git nutch_source && \ +# Clone from our fork. +RUN git clone --single-branch --branch v1.22-with-es8 https://github.com/nciocpl/nutch.git nutch_source && \ cd nutch_source && \ ant runtime && \ rm -rf build/ && \ @@ -78,7 +81,7 @@ ENV SERVER_HOST=$SERVER_HOST RUN mkdir -p /var/log/supervisord COPY ./config/supervisord_startserver.conf /etc/supervisord.conf -# Expose port for server which can only be accessed if +# Expose port for server which can only be accessed if # the same port is published when the container is run. EXPOSE $SERVER_PORT @@ -105,7 +108,7 @@ RUN git clone https://github.com/apache/nutch-webapp.git nutch_webapp && \ RUN mkdir -p /var/log/supervisord COPY ./config/supervisord_startserver_webapp.conf /etc/supervisord.conf -# Expose ports for server and webapp, these can only be accessed if +# Expose ports for server and webapp, these can only be accessed if # the same ports are published when the container is run. EXPOSE $SERVER_PORT EXPOSE $WEBAPP_PORT diff --git a/ivy/ivy.xml b/ivy/ivy.xml index 9b38d2fa96..4beba78c9f 100644 --- a/ivy/ivy.xml +++ b/ivy/ivy.xml @@ -38,9 +38,9 @@ - - - + + + @@ -74,7 +74,14 @@ - + + + + + + + + @@ -90,15 +97,16 @@ - - - - - - - - - + + + + + + + + + + @@ -126,6 +134,10 @@ + + + + @@ -136,6 +148,38 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/plugin/build-plugin.xml b/src/plugin/build-plugin.xml index b0aca71038..114143a3b3 100755 --- a/src/plugin/build-plugin.xml +++ b/src/plugin/build-plugin.xml @@ -84,6 +84,7 @@ + @@ -189,7 +190,7 @@ - + Tests failed! + + + + + + + + + + + + + + + + + + + + + + + + + Indexer integration tests failed! + + diff --git a/src/plugin/build.xml b/src/plugin/build.xml index 498259a950..37c8d73ba1 100755 --- a/src/plugin/build.xml +++ b/src/plugin/build.xml @@ -172,6 +172,13 @@ + + + + + + + diff --git a/src/plugin/indexer-elastic/README.md b/src/plugin/indexer-elastic/README.md index 3dfd888ff8..07b22d71f2 100644 --- a/src/plugin/indexer-elastic/README.md +++ b/src/plugin/indexer-elastic/README.md @@ -47,8 +47,8 @@ Each parameter has the form `` and the par Parameter Name | Description | Default value --|--|-- -host | Comma-separated list of hostnames to send documents to using [TransportClient](https://static.javadoc.io/org.elasticsearch/elasticsearch/5.3.0/org/elasticsearch/client/transport/TransportClient.html). Either host and port must be defined. | -port | The port to connect to using [TransportClient](https://static.javadoc.io/org.elasticsearch/elasticsearch/5.3.0/org/elasticsearch/client/transport/TransportClient.html). | 9300 +host | Comma-separated list of hostnames to send documents to using the Elasticsearch Java API Client and low-level REST client. Either host and port must be defined. | +port | The REST port to connect to the Elasticsearch server. | 9200 scheme | The scheme (http or https) to connect to elastic server. | http index | Default index to send documents to. | nutch username | Username for auth credentials | elastic @@ -56,6 +56,6 @@ password | Password for auth credentials | "" auth | Whether to enable HTTP basic authentication with elastic. Use `username` and `password` properties to configure your credentials. | false max.bulk.docs | Maximum size of the bulk in number of documents. | 250 max.bulk.size | Maximum size of the bulk in bytes. | 2500500 -exponential.backoff.millis | Initial delay for the [BulkProcessor](https://static.javadoc.io/org.elasticsearch/elasticsearch/5.3.0/org/elasticsearch/action/bulk/BulkProcessor.html) exponential backoff policy. | 100 -exponential.backoff.retries | Number of times the [BulkProcessor](https://static.javadoc.io/org.elasticsearch/elasticsearch/5.3.0/org/elasticsearch/action/bulk/BulkProcessor.html) exponential backoff policy should retry bulk operations. | 10 -bulk.close.timeout | Number of seconds allowed for the [BulkProcessor](https://static.javadoc.io/org.elasticsearch/elasticsearch/5.3.0/org/elasticsearch/action/bulk/BulkProcessor.html) to complete its last operation. | 600 +exponential.backoff.millis | Initial delay for the [BulkIngester](https://artifacts.elastic.co/javadoc/co/elastic/clients/elasticsearch-java/8.16.6/co/elastic/clients/elasticsearch/_helpers/bulk/BulkIngester.html) retry policy. | 100 +exponential.backoff.retries | Number of times the [BulkIngester](https://artifacts.elastic.co/javadoc/co/elastic/clients/elasticsearch-java/8.16.6/co/elastic/clients/elasticsearch/_helpers/bulk/BulkIngester.html) retry policy should retry retryable bulk operations. | 10 +bulk.close.timeout | Number of seconds allowed for the [BulkIngester](https://artifacts.elastic.co/javadoc/co/elastic/clients/elasticsearch-java/8.16.6/co/elastic/clients/elasticsearch/_helpers/bulk/BulkIngester.html) to complete its last operation. | 600 diff --git a/src/plugin/indexer-elastic/howto_upgrade_es.md b/src/plugin/indexer-elastic/howto_upgrade_es.md index ca58639d19..5aedb4ef5c 100644 --- a/src/plugin/indexer-elastic/howto_upgrade_es.md +++ b/src/plugin/indexer-elastic/howto_upgrade_es.md @@ -15,7 +15,8 @@ limitations under the License. --> -1. Upgrade Elasticsearch dependency in src/plugin/indexer-elastic/ivy.xml +1. Upgrade the Elasticsearch Java API Client and low-level REST client + dependencies in src/plugin/indexer-elastic/ivy.xml 2. Upgrade the Elasticsearch specific dependencies in src/plugin/indexer-elastic/plugin.xml To get the list of dependencies and their versions execute: @@ -24,12 +25,12 @@ $ ls lib | sed 's/^/ /g' In the plugin.xml replace all lines between - + and - + with the output of the command above. -4. (Optionally) remove overlapping dependencies between indexer-elastic and Nutch core dependencies: +3. (Optionally) remove overlapping dependencies between indexer-elastic and Nutch core dependencies: - check for libs present both in build/lib and @@ -40,11 +41,11 @@ - but it should be made sure that the library versions in ivy/ivy.xml correspond to those required by Tika -5. Remove the locally "installed" dependencies in src/plugin/indexer-elastic/lib/: +4. Remove the locally "installed" dependencies in src/plugin/indexer-elastic/lib/: $ rm -rf lib/ -6. Build Nutch and run all unit tests: +5. Build Nutch and run all unit tests: $ cd ../../../ $ ant clean runtime test diff --git a/src/plugin/indexer-elastic/ivy.xml b/src/plugin/indexer-elastic/ivy.xml index ee812a225c..075d4e9ac2 100644 --- a/src/plugin/indexer-elastic/ivy.xml +++ b/src/plugin/indexer-elastic/ivy.xml @@ -1,5 +1,5 @@ - - @@ -36,21 +36,15 @@ - - - - - - - - - - - - - - - + + + + + + + + + - + diff --git a/src/plugin/indexer-elastic/plugin.xml b/src/plugin/indexer-elastic/plugin.xml index ab7b3c2b00..b2f978033a 100644 --- a/src/plugin/indexer-elastic/plugin.xml +++ b/src/plugin/indexer-elastic/plugin.xml @@ -21,55 +21,20 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + diff --git a/src/plugin/indexer-elastic/src/java/org/apache/nutch/indexwriter/elastic/ElasticIndexWriter.java b/src/plugin/indexer-elastic/src/java/org/apache/nutch/indexwriter/elastic/ElasticIndexWriter.java index 84978dbc22..a0517fbaf2 100644 --- a/src/plugin/indexer-elastic/src/java/org/apache/nutch/indexwriter/elastic/ElasticIndexWriter.java +++ b/src/plugin/indexer-elastic/src/java/org/apache/nutch/indexwriter/elastic/ElasticIndexWriter.java @@ -16,17 +16,32 @@ */ package org.apache.nutch.indexwriter.elastic; +import java.io.IOException; import java.lang.invoke.MethodHandles; import java.time.format.DateTimeFormatter; -import java.io.IOException; import java.util.AbstractMap; +import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import javax.net.ssl.SSLContext; +import co.elastic.clients.elasticsearch.ElasticsearchClient; +import co.elastic.clients.elasticsearch._helpers.bulk.BulkIngester; +import co.elastic.clients.elasticsearch._helpers.bulk.BulkListener; +import co.elastic.clients.elasticsearch.core.BulkRequest; +import co.elastic.clients.elasticsearch.core.BulkResponse; +import co.elastic.clients.elasticsearch.core.bulk.BulkOperation; +import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem; +import co.elastic.clients.json.jackson.JacksonJsonpMapper; +import co.elastic.clients.transport.ElasticsearchTransport; +import co.elastic.clients.transport.rest_client.RestClientTransport; import org.apache.commons.lang3.StringUtils; import org.apache.hadoop.conf.Configuration; import org.apache.http.HttpHost; @@ -44,24 +59,9 @@ import org.apache.nutch.indexer.NutchDocument; import org.apache.nutch.indexer.NutchField; import org.apache.nutch.util.StringUtil; -import org.elasticsearch.action.bulk.BulkResponse; -import org.elasticsearch.action.DocWriteRequest; -import org.elasticsearch.action.bulk.BackoffPolicy; -import org.elasticsearch.action.bulk.BulkProcessor; -import org.elasticsearch.action.bulk.BulkRequest; -import org.elasticsearch.action.delete.DeleteRequest; -import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestClientBuilder; -import org.elasticsearch.client.RestHighLevelClient; -import org.elasticsearch.common.unit.ByteSizeUnit; -import org.elasticsearch.common.unit.ByteSizeValue; -import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.client.RestClientBuilder.HttpClientConfigCallback; -import org.elasticsearch.common.xcontent.XContentBuilder; -import org.elasticsearch.common.xcontent.XContentFactory; -import org.elasticsearch.client.RequestOptions; - import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -72,7 +72,7 @@ public class ElasticIndexWriter implements IndexWriter { private static final Logger LOG = LoggerFactory .getLogger(MethodHandles.lookup().lookupClass()); - private static final int DEFAULT_PORT = 9300; + private static final int DEFAULT_PORT = 9200; private static final int DEFAULT_MAX_BULK_DOCS = 250; private static final int DEFAULT_MAX_BULK_LENGTH = 2500500; private static final int DEFAULT_EXP_BACKOFF_MILLIS = 100; @@ -80,6 +80,8 @@ public class ElasticIndexWriter implements IndexWriter { private static final int DEFAULT_BULK_CLOSE_TIMEOUT = 600; private static final String DEFAULT_INDEX = "nutch"; private static final String DEFAULT_USER = "elastic"; + private static final int MAX_CONCURRENT_BULK_REQUESTS = 1; + private static final int DRAIN_POLL_MILLIS = 100; private String[] hosts; private int port; @@ -94,8 +96,12 @@ public class ElasticIndexWriter implements IndexWriter { private int expBackoffRetries; private String defaultIndex; - private RestHighLevelClient client; - private BulkProcessor bulkProcessor; + private ElasticsearchClient client; + private ElasticsearchTransport transport; + private BulkIngester bulkIngester; + private ScheduledExecutorService retryScheduler; + private final AtomicInteger scheduledRetries = new AtomicInteger(); + private volatile boolean closing; private long bulkCloseTimeout; @@ -142,30 +148,41 @@ public void open(IndexWriterParams parameters) throws IOException { DEFAULT_EXP_BACKOFF_RETRIES); client = makeClient(parameters); + retryScheduler = Executors.newScheduledThreadPool(2, + new ElasticBulkThreadFactory()); - LOG.debug("Creating BulkProcessor with maxBulkDocs={}, maxBulkLength={}", + LOG.debug("Creating BulkIngester with maxBulkDocs={}, maxBulkLength={}", maxBulkDocs, maxBulkLength); - bulkProcessor = BulkProcessor - .builder( - (request, bulkListener) -> client.bulkAsync(request, - RequestOptions.DEFAULT, bulkListener), - bulkProcessorListener()) - .setBulkActions(maxBulkDocs) - .setBulkSize(new ByteSizeValue(maxBulkLength, ByteSizeUnit.BYTES)) - .setConcurrentRequests(1) - .setBackoffPolicy(BackoffPolicy.exponentialBackoff( - TimeValue.timeValueMillis(expBackoffMillis), expBackoffRetries)) - .build(); + bulkIngester = BulkIngester.of(b -> b.client(client) + .maxOperations(maxBulkDocs) + .maxSize(maxBulkLength) + .maxConcurrentRequests(MAX_CONCURRENT_BULK_REQUESTS) + .scheduler(retryScheduler) + .listener(bulkListener())); } /** - * Generates a RestHighLevelClient with the hosts given - * @param parameters implementation specific {@link org.apache.nutch.indexer.IndexWriterParams} - * @return an initialized {@link org.elasticsearch.client.RestHighLevelClient} - * @throws IOException if there is an error reading the - * {@link org.apache.nutch.indexer.IndexWriterParams} + * Generates an ElasticsearchClient with the hosts given. + * + * @param parameters implementation specific {@link IndexWriterParams} + * @return an initialized {@link ElasticsearchClient} + * @throws IOException if there is an error reading the {@link IndexWriterParams} */ - protected RestHighLevelClient makeClient(IndexWriterParams parameters) + protected ElasticsearchClient makeClient(IndexWriterParams parameters) + throws IOException { + RestClient restClient = makeRestClient(parameters); + transport = new RestClientTransport(restClient, new JacksonJsonpMapper()); + return new ElasticsearchClient(transport); + } + + /** + * Generates a low-level RestClient with the hosts given. + * + * @param parameters implementation specific {@link IndexWriterParams} + * @return an initialized {@link RestClient} + * @throws IOException if there is an error reading the {@link IndexWriterParams} + */ + protected RestClient makeRestClient(IndexWriterParams parameters) throws IOException { hosts = parameters.getStrings(ElasticConstants.HOSTS); port = parameters.getInt(ElasticConstants.PORT, DEFAULT_PORT); @@ -178,82 +195,103 @@ protected RestHighLevelClient makeClient(IndexWriterParams parameters) credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(user, password)); - RestHighLevelClient client = null; - - if (hosts != null && port > 1) { - HttpHost[] hostsList = new HttpHost[hosts.length]; - int i = 0; - for (String host : hosts) { - hostsList[i++] = new HttpHost(host, port, scheme); - } - RestClientBuilder restClientBuilder = RestClient.builder(hostsList); - - if (auth) { - restClientBuilder - .setHttpClientConfigCallback(new HttpClientConfigCallback() { - @Override - public HttpAsyncClientBuilder customizeHttpClient( - HttpAsyncClientBuilder arg0) { - return arg0.setDefaultCredentialsProvider(credentialsProvider); - } - }); - } + if (hosts == null || port <= 1) { + throw new IOException( + "ElasticRestClient initialization Failed!!!\\n\\nPlease Provide the hosts"); + } - // In case of HTTPS, set the client up for ignoring problems with self-signed - // certificates and stuff - if ("https".equals(scheme)) { - try { - SSLContextBuilder sslBuilder = SSLContexts.custom(); - sslBuilder.loadTrustMaterial(null, new TrustSelfSignedStrategy()); - final SSLContext sslContext = sslBuilder.build(); + HttpHost[] hostsList = new HttpHost[hosts.length]; + int i = 0; + for (String host : hosts) { + hostsList[i++] = new HttpHost(host, port, scheme); + } + RestClientBuilder restClientBuilder = RestClient.builder(hostsList); - restClientBuilder.setHttpClientConfigCallback(new HttpClientConfigCallback() { + if (auth) { + restClientBuilder + .setHttpClientConfigCallback(new HttpClientConfigCallback() { @Override - public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) { - if (auth) { - httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider); - } - // ignore issues with self-signed certificates - httpClientBuilder.setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE); - return httpClientBuilder.setSSLContext(sslContext); + public HttpAsyncClientBuilder customizeHttpClient( + HttpAsyncClientBuilder arg0) { + return arg0.setDefaultCredentialsProvider(credentialsProvider); } }); - } catch (Exception e) { - LOG.error("Error setting up SSLContext because: {}", e.getMessage(), e); - } - } + } - client = new RestHighLevelClient(restClientBuilder); - } else { - throw new IOException( - "ElasticRestClient initialization Failed!!!\\n\\nPlease Provide the hosts"); + // In case of HTTPS, set the client up for ignoring problems with self-signed + // certificates and stuff. + if ("https".equals(scheme)) { + try { + SSLContextBuilder sslBuilder = SSLContexts.custom(); + sslBuilder.loadTrustMaterial(null, new TrustSelfSignedStrategy()); + final SSLContext sslContext = sslBuilder.build(); + + restClientBuilder.setHttpClientConfigCallback(new HttpClientConfigCallback() { + @Override + public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) { + if (auth) { + httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider); + } + // ignore issues with self-signed certificates + httpClientBuilder.setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE); + return httpClientBuilder.setSSLContext(sslContext); + } + }); + } catch (Exception e) { + LOG.error("Error setting up SSLContext because: {}", e.getMessage(), e); + } } - return client; + return restClientBuilder.build(); } /** - * Generates a default BulkProcessor.Listener - * @return {@link BulkProcessor.Listener} + * Generates a default BulkIngester listener. + * + * @return {@link BulkListener} */ - protected BulkProcessor.Listener bulkProcessorListener() { - return new BulkProcessor.Listener() { + protected BulkListener bulkListener() { + return new BulkListener() { @Override - public void beforeBulk(long executionId, BulkRequest request) { + public void beforeBulk(long executionId, BulkRequest request, + List contexts) { } @Override public void afterBulk(long executionId, BulkRequest request, - Throwable failure) { + List contexts, Throwable failure) { LOG.error("Elasticsearch indexing failed:", failure); + for (RetryContext context : contexts) { + scheduleRetry(context, "bulk request failure: " + failure.getMessage()); + } } @Override public void afterBulk(long executionId, BulkRequest request, - BulkResponse response) { - if (response.hasFailures()) { - LOG.warn("Failures occurred during bulk request: {}", - response.buildFailureMessage()); + List contexts, BulkResponse response) { + if (!response.errors()) { + return; + } + int loggedFailures = 0; + List items = response.items(); + for (int i = 0; i < items.size() && i < contexts.size(); i++) { + BulkResponseItem item = items.get(i); + if (item.error() == null) { + continue; + } + RetryContext context = contexts.get(i); + String reason = item.error().reason(); + if (isRetryableStatus(item.status())) { + scheduleRetry(context, "status " + item.status() + ": " + reason); + } else { + loggedFailures++; + LOG.warn("Permanent Elasticsearch bulk failure for {}: status={}, type={}, reason={}", + context.description, item.status(), item.error().type(), reason); + } + } + if (loggedFailures > 0) { + LOG.warn("Permanent failures occurred during bulk request: {}", + loggedFailures); } } }; @@ -262,39 +300,14 @@ public void afterBulk(long executionId, BulkRequest request, @Override public void write(NutchDocument doc) throws IOException { String id = (String) doc.getFieldValue("id"); - String type = doc.getDocumentMeta().get("type"); - if (type == null) - type = "doc"; - - // Add each field of this doc to the index builder - XContentBuilder builder = XContentFactory.jsonBuilder().startObject(); - for (final Map.Entry e : doc) { - final List values = e.getValue().getValues(); - - if (values.size() > 1) { - builder.array(e.getKey(), values); - } else { - Object value = values.get(0); - if (value instanceof java.util.Date) { - value = DateTimeFormatter.ISO_INSTANT - .format(((java.util.Date) value).toInstant()); - } - builder.field(e.getKey(), value); - } - } - builder.endObject(); - - IndexRequest request = new IndexRequest(defaultIndex).id(id) - .source(builder); - request.opType(DocWriteRequest.OpType.INDEX); - - bulkProcessor.add(request); + BulkOperation operation = buildIndexOperation(id, doc); + addOperation(operation, "index document " + id); } @Override public void delete(String key) throws IOException { - DeleteRequest request = new DeleteRequest(defaultIndex, key); - bulkProcessor.add(request); + BulkOperation operation = buildDeleteOperation(key); + addOperation(operation, "delete document " + key); } @Override @@ -304,20 +317,174 @@ public void update(NutchDocument doc) throws IOException { @Override public void commit() throws IOException { - bulkProcessor.flush(); + bulkIngester.flush(); } @Override public void close() throws IOException { - // Close BulkProcessor (automatically flushes) + IOException closeException = null; try { - bulkProcessor.awaitClose(bulkCloseTimeout, TimeUnit.SECONDS); + if (bulkIngester != null) { + bulkIngester.flush(); + awaitBulkCompletion(bulkCloseTimeout, TimeUnit.SECONDS); + bulkIngester.close(); + } } catch (InterruptedException e) { - LOG.warn("interrupted while waiting for BulkProcessor to complete ({})", + Thread.currentThread().interrupt(); + LOG.warn("interrupted while waiting for BulkIngester to complete ({})", e.getMessage()); + } catch (RuntimeException e) { + closeException = new IOException("Error closing BulkIngester", e); + } finally { + closing = true; + shutdownRetryScheduler(); + } + + try { + if (transport != null) { + transport.close(); + } + } catch (IOException e) { + if (closeException == null) { + closeException = e; + } else { + closeException.addSuppressed(e); + } } - client.close(); + if (closeException != null) { + throw closeException; + } + } + + BulkOperation buildIndexOperation(String id, NutchDocument doc) { + Map source = createSourceMap(doc); + return BulkOperation.of(op -> op.index(index -> index + .index(defaultIndex) + .id(id) + .document(source))); + } + + BulkOperation buildDeleteOperation(String key) { + return BulkOperation.of(op -> op.delete(delete -> delete + .index(defaultIndex) + .id(key))); + } + + Map createSourceMap(NutchDocument doc) { + Map source = new LinkedHashMap<>(); + for (final Map.Entry e : doc) { + final List values = e.getValue().getValues(); + + if (values.size() > 1) { + List normalizedValues = new ArrayList<>(values.size()); + for (Object value : values) { + normalizedValues.add(normalizeValue(value)); + } + source.put(e.getKey(), normalizedValues); + } else { + source.put(e.getKey(), normalizeValue(values.get(0))); + } + } + return source; + } + + private static Object normalizeValue(Object value) { + if (value instanceof java.util.Date) { + return DateTimeFormatter.ISO_INSTANT + .format(((java.util.Date) value).toInstant()); + } + return value; + } + + private void addOperation(BulkOperation operation, String description) { + bulkIngester.add(operation, new RetryContext(operation, description, 0)); + } + + private void scheduleRetry(RetryContext context, String reason) { + if (closing) { + LOG.warn("Skipping Elasticsearch retry for {} because writer is closing: {}", + context.description, reason); + return; + } + if (context.attempt >= expBackoffRetries) { + LOG.warn("Exhausted Elasticsearch bulk retries for {} after {} attempts: {}", + context.description, context.attempt, reason); + return; + } + + long delayMillis = computeExponentialBackoffMillis(expBackoffMillis, + context.attempt); + RetryContext retryContext = context.nextAttempt(); + scheduledRetries.incrementAndGet(); + retryScheduler.schedule(() -> { + try { + bulkIngester.add(retryContext.operation, retryContext); + } catch (RuntimeException e) { + LOG.error("Elasticsearch retry failed for {}:", retryContext.description, e); + } finally { + scheduledRetries.decrementAndGet(); + } + }, delayMillis, TimeUnit.MILLISECONDS); + LOG.debug("Scheduled Elasticsearch retry for {} in {} ms after {}", + context.description, delayMillis, reason); + } + + private void awaitBulkCompletion(long timeout, TimeUnit unit) + throws InterruptedException { + long timeoutNanos = unit.toNanos(timeout); + long deadline = System.nanoTime() + timeoutNanos; + while (hasPendingBulkWork() && System.nanoTime() < deadline) { + TimeUnit.MILLISECONDS.sleep(DRAIN_POLL_MILLIS); + } + + if (hasPendingBulkWork()) { + LOG.warn( + "Timed out waiting for BulkIngester to complete. pendingOperations={}, pendingRequests={}, scheduledRetries={}", + bulkIngester.pendingOperations(), bulkIngester.pendingRequests(), + scheduledRetries.get()); + } + } + + private boolean hasPendingBulkWork() { + return bulkIngester != null + && (bulkIngester.pendingOperations() > 0 + || bulkIngester.pendingRequests() > 0 + || scheduledRetries.get() > 0); + } + + private void shutdownRetryScheduler() { + if (retryScheduler == null) { + return; + } + retryScheduler.shutdown(); + try { + if (!retryScheduler.awaitTermination(5, TimeUnit.SECONDS)) { + retryScheduler.shutdownNow(); + } + } catch (InterruptedException e) { + retryScheduler.shutdownNow(); + Thread.currentThread().interrupt(); + } + } + + static boolean isRetryableStatus(int status) { + return status == 429 || status == 502 || status == 503 || status == 504; + } + + static long computeExponentialBackoffMillis(int initialDelayMillis, + int completedAttempts) { + if (initialDelayMillis <= 0) { + return 0; + } + long delay = initialDelayMillis; + for (int i = 0; i < completedAttempts; i++) { + if (delay > Long.MAX_VALUE / 2) { + return Long.MAX_VALUE; + } + delay *= 2; + } + return delay; } /** @@ -353,15 +520,15 @@ public Map> describe() { this.maxBulkLength)); properties.put(ElasticConstants.EXPONENTIAL_BACKOFF_MILLIS, new AbstractMap.SimpleEntry<>( - "Initial delay for the BulkProcessor exponential backoff policy.", + "Initial delay for the BulkIngester retry policy.", this.expBackoffMillis)); properties.put(ElasticConstants.EXPONENTIAL_BACKOFF_RETRIES, new AbstractMap.SimpleEntry<>( - "Number of times the BulkProcessor exponential backoff policy should retry bulk operations.", + "Number of times the BulkIngester retry policy should retry retryable bulk operations.", this.expBackoffRetries)); properties.put(ElasticConstants.BULK_CLOSE_TIMEOUT, new AbstractMap.SimpleEntry<>( - "Number of seconds allowed for the BulkProcessor to complete its last operation.", + "Number of seconds allowed for the BulkIngester to complete its last operation.", this.bulkCloseTimeout)); return properties; @@ -376,4 +543,32 @@ public void setConf(Configuration conf) { public Configuration getConf() { return config; } + + static final class RetryContext { + private final BulkOperation operation; + private final String description; + private final int attempt; + + RetryContext(BulkOperation operation, String description, int attempt) { + this.operation = operation; + this.description = description; + this.attempt = attempt; + } + + RetryContext nextAttempt() { + return new RetryContext(operation, description, attempt + 1); + } + } + + private static final class ElasticBulkThreadFactory implements ThreadFactory { + private final AtomicInteger threadNumber = new AtomicInteger(); + + @Override + public Thread newThread(Runnable runnable) { + Thread thread = new Thread(runnable, + "nutch-elastic-bulk-" + threadNumber.incrementAndGet()); + thread.setDaemon(true); + return thread; + } + } } diff --git a/src/plugin/indexer-elastic/src/test/org/apache/nutch/indexwriter/elastic/ElasticIndexWriterIT.java b/src/plugin/indexer-elastic/src/test/org/apache/nutch/indexwriter/elastic/ElasticIndexWriterIT.java new file mode 100644 index 0000000000..419cf7e1a4 --- /dev/null +++ b/src/plugin/indexer-elastic/src/test/org/apache/nutch/indexwriter/elastic/ElasticIndexWriterIT.java @@ -0,0 +1,157 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nutch.indexwriter.elastic; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import co.elastic.clients.elasticsearch.ElasticsearchClient; +import co.elastic.clients.elasticsearch.core.GetResponse; +import co.elastic.clients.json.jackson.JacksonJsonpMapper; +import co.elastic.clients.transport.ElasticsearchTransport; +import co.elastic.clients.transport.rest_client.RestClientTransport; +import org.apache.hadoop.conf.Configuration; +import org.apache.http.HttpHost; +import org.apache.nutch.indexer.AbstractIndexWriterIT; +import org.apache.nutch.indexer.IndexWriter; +import org.apache.nutch.indexer.IndexWriterParams; +import org.apache.nutch.indexer.NutchDocument; +import org.apache.nutch.util.NutchConfiguration; +import org.elasticsearch.client.RestClient; +import org.junit.jupiter.api.Test; +import org.testcontainers.elasticsearch.ElasticsearchContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Integration tests for ElasticIndexWriter using Testcontainers. + */ +@Testcontainers(disabledWithoutDocker = true) +public class ElasticIndexWriterIT extends AbstractIndexWriterIT { + + private static final String ELASTICSEARCH_IMAGE = + "docker.elastic.co/elasticsearch/elasticsearch:8.16.6"; + private static final String TEST_INDEX = "test-index"; + + @Container + private static final ElasticsearchContainer elasticsearchContainer = + new ElasticsearchContainer(ELASTICSEARCH_IMAGE) + .withEnv("discovery.type", "single-node") + .withEnv("xpack.security.enabled", "false") + .withEnv("ES_JAVA_OPTS", "-Xms512m -Xmx512m"); + + private ElasticIndexWriter indexWriter; + private Configuration conf; + + @Override + public void setUpIndexWriter() throws Exception { + conf = NutchConfiguration.create(); + indexWriter = new ElasticIndexWriter(); + indexWriter.setConf(conf); + + Map params = new HashMap<>(); + params.put(ElasticConstants.HOSTS, elasticsearchContainer.getHost()); + params.put(ElasticConstants.PORT, String.valueOf(elasticsearchContainer.getMappedPort(9200))); + params.put(ElasticConstants.INDEX, TEST_INDEX); + params.put(ElasticConstants.SCHEME, "http"); + + IndexWriterParams writerParams = new IndexWriterParams(params); + indexWriter.open(writerParams); + } + + @Override + public void tearDownIndexWriter() throws Exception { + if (indexWriter != null) { + try { + indexWriter.close(); + } catch (Exception e) { + // Ignore if open() failed and close state is invalid + } + indexWriter = null; + } + } + + @Override + public IndexWriter getIndexWriter() { + return indexWriter; + } + + @Override + public boolean supportsDelete() { + return true; + } + + @Override + public void verifyDocumentWritten(String docId, String expectedTitle) throws Exception { + GetResponse getResponse = getDocument(docId); + assertTrue(getResponse.found(), "Document should exist in index"); + assertNotNull(getResponse.source()); + assertEquals(expectedTitle, getResponse.source().get("title")); + } + + @Test + void testWriteMultiValueField() throws Exception { + NutchDocument doc = createTestDocument("test-doc-multi-value", + "Multi Value Document", ""); + doc.add("tag", "one"); + doc.add("tag", "two"); + + indexWriter.write(doc); + indexWriter.commit(); + tearDownIndexWriter(); + + GetResponse getResponse = getDocument("test-doc-multi-value"); + assertTrue(getResponse.found(), "Document should exist in index"); + Object tags = getResponse.source().get("tag"); + assertInstanceOf(List.class, tags); + assertEquals(List.of("one", "two"), tags); + } + + @Test + void testDeleteRemovesDocument() throws Exception { + String docId = "test-doc-elastic-delete"; + NutchDocument doc = createTestDocument(docId, "Document to Delete", ""); + + indexWriter.write(doc); + indexWriter.commit(); + indexWriter.delete(docId); + indexWriter.commit(); + tearDownIndexWriter(); + + GetResponse getResponse = getDocument(docId); + assertFalse(getResponse.found(), "Document should be deleted from index"); + } + + private GetResponse getDocument(String docId) throws Exception { + try (ElasticsearchTransport transport = new RestClientTransport( + RestClient.builder( + new HttpHost(elasticsearchContainer.getHost(), + elasticsearchContainer.getMappedPort(9200), + "http")).build(), + new JacksonJsonpMapper())) { + ElasticsearchClient client = new ElasticsearchClient(transport); + return client.get(get -> get.index(TEST_INDEX).id(docId), Map.class); + } + } +} diff --git a/src/plugin/indexer-elastic/src/test/org/apache/nutch/indexwriter/elastic/TestElasticIndexWriter.java b/src/plugin/indexer-elastic/src/test/org/apache/nutch/indexwriter/elastic/TestElasticIndexWriter.java new file mode 100644 index 0000000000..de06503dac --- /dev/null +++ b/src/plugin/indexer-elastic/src/test/org/apache/nutch/indexwriter/elastic/TestElasticIndexWriter.java @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nutch.indexwriter.elastic; + +import java.time.Instant; +import java.util.Date; +import java.util.List; +import java.util.Map; + +import org.apache.nutch.indexer.NutchDocument; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class TestElasticIndexWriter { + + @Test + void testCreateSourceMapPreservesCurrentFieldShape() { + ElasticIndexWriter writer = new ElasticIndexWriter(); + NutchDocument doc = new NutchDocument(); + Date fetchTime = Date.from(Instant.parse("2024-01-02T03:04:05Z")); + + doc.add("id", "doc-1"); + doc.add("title", "Test Document"); + doc.add("tag", "one"); + doc.add("tag", "two"); + doc.add("fetchTime", fetchTime); + + Map source = writer.createSourceMap(doc); + + assertEquals("doc-1", source.get("id")); + assertEquals("Test Document", source.get("title")); + assertEquals(List.of("one", "two"), source.get("tag")); + assertEquals("2024-01-02T03:04:05Z", source.get("fetchTime")); + } + + @Test + void testRetryableStatuses() { + assertTrue(ElasticIndexWriter.isRetryableStatus(429)); + assertTrue(ElasticIndexWriter.isRetryableStatus(502)); + assertTrue(ElasticIndexWriter.isRetryableStatus(503)); + assertTrue(ElasticIndexWriter.isRetryableStatus(504)); + + assertFalse(ElasticIndexWriter.isRetryableStatus(400)); + assertFalse(ElasticIndexWriter.isRetryableStatus(401)); + assertFalse(ElasticIndexWriter.isRetryableStatus(404)); + assertFalse(ElasticIndexWriter.isRetryableStatus(409)); + } + + @Test + void testExponentialBackoffMillis() { + assertEquals(100L, + ElasticIndexWriter.computeExponentialBackoffMillis(100, 0)); + assertEquals(200L, + ElasticIndexWriter.computeExponentialBackoffMillis(100, 1)); + assertEquals(400L, + ElasticIndexWriter.computeExponentialBackoffMillis(100, 2)); + assertEquals(0L, + ElasticIndexWriter.computeExponentialBackoffMillis(0, 2)); + } +} diff --git a/src/plugin/indexer-kafka/ivy.xml b/src/plugin/indexer-kafka/ivy.xml index d6157d953e..1fcdad8ed3 100644 --- a/src/plugin/indexer-kafka/ivy.xml +++ b/src/plugin/indexer-kafka/ivy.xml @@ -39,6 +39,15 @@ + + + + + + + + + diff --git a/src/plugin/indexer-kafka/plugin.xml b/src/plugin/indexer-kafka/plugin.xml index 1f9e4d1d97..f9fb3f7d61 100644 --- a/src/plugin/indexer-kafka/plugin.xml +++ b/src/plugin/indexer-kafka/plugin.xml @@ -58,15 +58,15 @@ - - - - - - - - - + + + + + + + + + diff --git a/src/plugin/indexer-solr/ivy.xml b/src/plugin/indexer-solr/ivy.xml index 99a713c18b..017abcc2f3 100644 --- a/src/plugin/indexer-solr/ivy.xml +++ b/src/plugin/indexer-solr/ivy.xml @@ -62,6 +62,15 @@ + + + + + + + + + diff --git a/src/plugin/indexer-solr/plugin.xml b/src/plugin/indexer-solr/plugin.xml index be98131414..dc3d61c17c 100644 --- a/src/plugin/indexer-solr/plugin.xml +++ b/src/plugin/indexer-solr/plugin.xml @@ -40,15 +40,15 @@ - - - - - - - - - + + + + + + + + + diff --git a/src/plugin/language-identifier/ivy.xml b/src/plugin/language-identifier/ivy.xml index 448619ac11..b588b19797 100644 --- a/src/plugin/language-identifier/ivy.xml +++ b/src/plugin/language-identifier/ivy.xml @@ -37,7 +37,7 @@ - + diff --git a/src/plugin/language-identifier/plugin.xml b/src/plugin/language-identifier/plugin.xml index 7fc3d555cb..b29feef631 100644 --- a/src/plugin/language-identifier/plugin.xml +++ b/src/plugin/language-identifier/plugin.xml @@ -26,15 +26,15 @@ - - + + - + - + diff --git a/src/plugin/lib-htmlunit/ivy.xml b/src/plugin/lib-htmlunit/ivy.xml index 63ae1e5d6c..895ddab047 100644 --- a/src/plugin/lib-htmlunit/ivy.xml +++ b/src/plugin/lib-htmlunit/ivy.xml @@ -40,6 +40,20 @@ + + + + + + + + + + + + + + diff --git a/src/plugin/lib-htmlunit/plugin.xml b/src/plugin/lib-htmlunit/plugin.xml index 6f14209af4..321452c467 100644 --- a/src/plugin/lib-htmlunit/plugin.xml +++ b/src/plugin/lib-htmlunit/plugin.xml @@ -143,49 +143,49 @@ - + - + - + - + - + - + - + - + - + - + - + - + - + - + diff --git a/src/plugin/parse-tika/ivy.xml b/src/plugin/parse-tika/ivy.xml index 23af88eec6..a81c37294b 100644 --- a/src/plugin/parse-tika/ivy.xml +++ b/src/plugin/parse-tika/ivy.xml @@ -37,13 +37,13 @@ - + - + diff --git a/src/plugin/parse-tika/plugin.xml b/src/plugin/parse-tika/plugin.xml index 1b9252cc69..eafa71709e 100644 --- a/src/plugin/parse-tika/plugin.xml +++ b/src/plugin/parse-tika/plugin.xml @@ -27,94 +27,94 @@ - - - - - - - - + + + + + + + + - + - - - - + + + + - + - + - + - - - + + + - - + + - - - - + + + + - + - + - - - - + + + + - - - - - - - - + + + + + + + + - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + + - + - + diff --git a/src/plugin/protocol-http/ivy.xml b/src/plugin/protocol-http/ivy.xml index 29871da120..1853f47c68 100644 --- a/src/plugin/protocol-http/ivy.xml +++ b/src/plugin/protocol-http/ivy.xml @@ -37,7 +37,7 @@ - + diff --git a/src/test/org/apache/nutch/indexer/AbstractIndexWriterIT.java b/src/test/org/apache/nutch/indexer/AbstractIndexWriterIT.java new file mode 100644 index 0000000000..8971d1a807 --- /dev/null +++ b/src/test/org/apache/nutch/indexer/AbstractIndexWriterIT.java @@ -0,0 +1,91 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nutch.indexer; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.testcontainers.junit.jupiter.Testcontainers; + +/** + * Abstract base for IndexWriter integration tests. Provides common test logic + * for write/commit and delete operations. + */ +@Testcontainers(disabledWithoutDocker = true) +public abstract class AbstractIndexWriterIT + implements IndexWriterIntegrationTest { + + @BeforeEach + void setUp() throws Exception { + setUpIndexWriter(); + } + + @AfterEach + void tearDown() throws Exception { + tearDownIndexWriter(); + } + + @Test + void testWriteAndCommitDocument() throws Exception { + NutchDocument doc = createTestDocument("test-doc-1", "Test Document", + "This is a test document for integration testing."); + assertDoesNotThrow(() -> getIndexWriter().write(doc)); + assertDoesNotThrow(() -> getIndexWriter().commit()); + tearDownIndexWriter(); + verifyDocumentWritten("test-doc-1", "Test Document"); + } + + @Test + void testDeleteDocument() throws Exception { + if (!supportsDelete()) { + return; + } + String docId = "test-doc-to-delete"; + NutchDocument doc = createTestDocument(docId, "Document to Delete", ""); + + IndexWriter writer = getIndexWriter(); + writer.write(doc); + writer.commit(); + + IndexWriter deleteWriter = prepareWriterForDeleteTest(); + if (deleteWriter == null) { + deleteWriter = writer; + } + final IndexWriter writerForDelete = deleteWriter; + assertDoesNotThrow(() -> writerForDelete.delete(docId)); + assertDoesNotThrow(() -> writerForDelete.commit()); + if (deleteWriter != writer) { + try { + deleteWriter.close(); + } catch (Exception e) { + // Ignore + } + } + } + + /** Create a NutchDocument with id, title, and content. */ + protected NutchDocument createTestDocument(String id, String title, + String content) { + NutchDocument doc = new NutchDocument(); + doc.add("id", id); + doc.add("title", title); + doc.add("content", content); + return doc; + } +} diff --git a/src/test/org/apache/nutch/indexer/IndexWriterIntegrationTest.java b/src/test/org/apache/nutch/indexer/IndexWriterIntegrationTest.java new file mode 100644 index 0000000000..45d8f37b91 --- /dev/null +++ b/src/test/org/apache/nutch/indexer/IndexWriterIntegrationTest.java @@ -0,0 +1,50 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nutch.indexer; + +/** + * Contract for IndexWriter integration tests. Implementations run against + * real backends via Testcontainers. + */ +public interface IndexWriterIntegrationTest { + + /** Open the index writer before tests. */ + void setUpIndexWriter() throws Exception; + + /** Close the index writer after tests. */ + void tearDownIndexWriter() throws Exception; + + /** The IndexWriter under test. */ + IndexWriter getIndexWriter(); + + /** Whether this writer supports document deletion. */ + boolean supportsDelete(); + + /** Optional verification that a document was indexed. */ + default void verifyDocumentWritten(String docId, String expectedTitle) + throws Exception { + // no-op + } + + /** + * Optional writer configured for delete operations. Default returns null to + * use {@link #getIndexWriter()}. + */ + default IndexWriter prepareWriterForDeleteTest() throws Exception { + return null; + } +}