diff --git a/docs/changelog.md b/docs/changelog.md index 6ecffcb382..f22698827b 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -347,6 +347,48 @@ If custom vertex ids are used, avoid deleting and re-creating vertices under the (run `REINDEX` afterwards to restore them). With automatically assigned ids this race cannot occur because ids are never reused. +##### Transient Elasticsearch failures are retried instead of dropping the index mutation + +Previously every Elasticsearch failure except an interrupt was reported as a `PermanentBackendException`. Because +index mutations are applied after the storage mutations in a commit they cannot be rolled back, so a transient +failure — a rolling restart, a saturated write queue, a socket timeout during a GC pause — dropped the mutation with +a single ERROR log line and left the mixed index inconsistent with the graph until a reindex or a transaction-log +recovery repaired it. + +Such a failure is now reported as a `TemporaryBackendException`, which the retry loop already present in +`BackendOperation` reattempts with exponential backoff for up to `storage.write-time` (default 100s). This applies +after the `index.[X].elasticsearch.retry-limit` attempts made by the Elasticsearch client itself, and unlike those it +also covers failures which never produced an HTTP response and therefore cannot be matched by +`index.[X].elasticsearch.retry-error-codes`. + +This behavior is **enabled by default**. Two new configuration options control it: +``` +index.[X].elasticsearch.temporary-error-codes=429,502,503,504 +index.[X].elasticsearch.temporary-transport-failures=true +``` +`temporary-error-codes` lists the HTTP status codes considered transient, whether reported by the request itself or +by an individual bulk item. A bulk request is reattempted only when *every* item which failed did so with one of +these codes, since a batch containing a permanently failing item — a mapping conflict, for instance — cannot succeed +on a reattempt. `temporary-transport-failures` covers the failures that produce no HTTP response at all: connection +refused, connection reset, socket timeout and a prematurely closed connection. + +Be aware of the following while either option is enabled: + +- A reattempt resubmits the whole mutation, including the bulk items which had already succeeded. This applies to + both options, not only to `temporary-transport-failures`: a bulk item rejected with 429 means the remaining items + of that batch were applied, a 502 or 504 from a proxy leaves it unknown whether Elasticsearch applied the request, + and a transport failure leaves the same question open. +- Resubmission is idempotent for whole-document writes, deletions and `SET` cardinality properties, but the values + of a `LIST` cardinality property are appended, so a resubmitted item can duplicate them in the index document. +- Set `index.[X].elasticsearch.temporary-error-codes` to an empty list **and** + `index.[X].elasticsearch.temporary-transport-failures=false` if duplicated `LIST` values in a mixed index are less + acceptable than a dropped mutation. Disabling only one of the two still leaves the other able to reattempt. + +Clearing both options restores the previous behavior for every failure which reached Elasticsearch. An interrupt is +not covered by either option: it was already reported as a `TemporaryBackendException` before this change, and still +is. It now aborts the reattempt loop immediately rather than letting it run for the whole write time budget, because +the interrupt status of the thread is restored once the `InterruptedException` has been consumed. + ### Version 1.1.0 (Release Date: November 7, 2024) /// tab | Maven diff --git a/docs/configs/janusgraph-cfg.md b/docs/configs/janusgraph-cfg.md index e98d979523..f38df5ffab 100644 --- a/docs/configs/janusgraph-cfg.md +++ b/docs/configs/janusgraph-cfg.md @@ -167,6 +167,8 @@ Elasticsearch index configuration | index.[X].elasticsearch.scroll-keep-alive | How long (in seconds) elasticsearch should keep alive the scroll context. | Integer | 60 | GLOBAL_OFFLINE | | index.[X].elasticsearch.setup-max-open-scroll-contexts | Whether JanusGraph should setup max_open_scroll_context to maximum value for the cluster or not. | Boolean | true | MASKABLE | | index.[X].elasticsearch.socket-timeout | Sets the maximum socket timeout (in milliseconds). | Integer | 30000 | MASKABLE | +| index.[X].elasticsearch.temporary-error-codes | Comma separated list of Elasticsearch HTTP status codes which are considered transient. An index operation failing with one of these codes is reported as a temporary rather than a permanent backend exception, so JanusGraph reattempts it with exponential backoff for up to `storage.write-time` instead of dropping the mutation and leaving the mixed index inconsistent with the graph. This is applied after the `retry-limit` attempts made by the Elasticsearch client itself have been exhausted. A reattempt resubmits the whole mutation, including the bulk items which already succeeded, so it can duplicate the values of LIST cardinality properties as described by `temporary-transport-failures`. Set to an empty list to consider every status code permanent. E.g. "429,502,503,504" | String[] | 429,502,503,504 | LOCAL | +| index.[X].elasticsearch.temporary-transport-failures | Whether Elasticsearch failures which never produced an HTTP response - connection refused, connection reset, socket timeout, prematurely closed connection - are considered transient, and are therefore reattempted for up to `storage.write-time` as described by `temporary-error-codes`. Such a failure leaves it unknown whether Elasticsearch applied the request, so a reattempt may resubmit items which already succeeded. Resubmission is idempotent except for LIST cardinality properties, whose values are appended. Disabling this alone does not remove that risk, because a reattempt driven by `temporary-error-codes` resubmits the whole mutation as well: clear both options if duplicated values in a mixed index are less acceptable than a dropped mutation. | Boolean | true | LOCAL | | index.[X].elasticsearch.use-all-field | Whether JanusGraph should add an "all" field mapping. When enabled field mappings will include a "copy_to" parameter referencing the "all" field. This is supported since Elasticsearch 6.x and is required when using wildcard fields starting in Elasticsearch 6.x. | Boolean | true | GLOBAL_OFFLINE | | index.[X].elasticsearch.use-mapping-for-es7 | Mapping types are deprecated in ElasticSearch 7 and JanusGraph will not use mapping types by default for ElasticSearch 7 but if you want to preserve mapping types, you can setup this parameter to true. If you are updating ElasticSearch from 6 to 7 and you don't want to reindex your indexes, you may setup this parameter to true but we do recommend to reindex your indexes and don't use this parameter. | Boolean | false | MASKABLE | diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchBulkFailureException.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchBulkFailureException.java new file mode 100644 index 0000000000..b37bd986ac --- /dev/null +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchBulkFailureException.java @@ -0,0 +1,42 @@ +// Copyright 2026 JanusGraph Authors +// +// Licensed 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.janusgraph.diskstorage.es; + +import java.io.IOException; +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; + +//Thrown when items of an Elasticsearch bulk request failed and could not be retried by the client itself. A bulk +//request reports item level failures inside an otherwise successful HTTP response, so the status of a failed item is +//not available from the enclosing response. Retaining those statuses here allows a caller to tell a transient failure +//- a rejected execution while a write queue is saturated, for instance - apart from a permanent one such as a +//mapping conflict +public class ElasticSearchBulkFailureException extends IOException { + + private static final long serialVersionUID = 5060142174725161541L; + + private final Set failedItemStatusCodes; + + public ElasticSearchBulkFailureException(String message, Set failedItemStatusCodes) { + super(message); + this.failedItemStatusCodes = Collections.unmodifiableSet(new HashSet<>(failedItemStatusCodes)); + } + + //The distinct HTTP status codes Elasticsearch reported for the bulk items which failed + public Set getFailedItemStatusCodes() { + return failedItemStatusCodes; + } +} diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchIndex.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchIndex.java index 4e3e5a0ff8..b86c302de3 100644 --- a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchIndex.java +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchIndex.java @@ -14,6 +14,7 @@ package org.janusgraph.diskstorage.es; +import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; @@ -21,6 +22,9 @@ import com.google.common.collect.LinkedListMultimap; import com.google.common.collect.Multimap; import com.google.common.collect.Sets; +import org.apache.http.ConnectionClosedException; +import org.apache.http.NoHttpResponseException; +import org.elasticsearch.client.ResponseException; import org.elasticsearch.client.RestClientBuilder; import org.janusgraph.core.Cardinality; import org.janusgraph.core.JanusGraphException; @@ -71,7 +75,9 @@ import org.slf4j.LoggerFactory; import java.io.IOException; +import java.io.InterruptedIOException; import java.io.UncheckedIOException; +import java.net.SocketException; import java.time.Instant; import java.util.ArrayList; import java.util.Arrays; @@ -324,6 +330,31 @@ public class ElasticSearchIndex implements IndexProvider { "Comma separated list of Elasticsearch REST client ResponseException error codes to retry. " + "E.g. \"408,429\"", ConfigOption.Type.LOCAL, String[].class, new String[0]); + public static final ConfigOption TEMPORARY_ERROR_CODES = + new ConfigOption<>(ELASTICSEARCH_NS, "temporary-error-codes", + "Comma separated list of Elasticsearch HTTP status codes which are considered transient. An index " + + "operation failing with one of these codes is reported as a temporary rather than a permanent " + + "backend exception, so JanusGraph reattempts it with exponential backoff for up to " + + "`storage.write-time` instead of dropping the mutation and leaving the mixed index inconsistent " + + "with the graph. This is applied after the `retry-limit` attempts made by the Elasticsearch client " + + "itself have been exhausted. A reattempt resubmits the whole mutation, including the bulk items " + + "which already succeeded, so it can duplicate the values of LIST cardinality properties as " + + "described by `temporary-transport-failures`. Set to an empty list to consider every status code " + + "permanent. E.g. \"429,502,503,504\"", ConfigOption.Type.LOCAL, String[].class, + new String[]{"429", "502", "503", "504"}); + + public static final ConfigOption TEMPORARY_TRANSPORT_FAILURES = + new ConfigOption<>(ELASTICSEARCH_NS, "temporary-transport-failures", + "Whether Elasticsearch failures which never produced an HTTP response - connection refused, connection " + + "reset, socket timeout, prematurely closed connection - are considered transient, and are therefore " + + "reattempted for up to `storage.write-time` as described by `temporary-error-codes`. Such a failure " + + "leaves it unknown whether Elasticsearch applied the request, so a reattempt may resubmit items " + + "which already succeeded. Resubmission is idempotent except for LIST cardinality properties, whose " + + "values are appended. Disabling this alone does not remove that risk, because a reattempt driven by " + + "`temporary-error-codes` resubmits the whole mutation as well: clear both options if duplicated " + + "values in a mixed index are less acceptable than a dropped mutation.", + ConfigOption.Type.LOCAL, true); + public static final ConfigOption BULK_CHUNK_SIZE_LIMIT_BYTES = new ConfigOption<>(ELASTICSEARCH_NS, "bulk-chunk-size-limit-bytes", "The total size limit in bytes of a bulk request. Mutation batches in excess of this limit will be " + @@ -398,6 +429,8 @@ public class ElasticSearchIndex implements IndexProvider { private final String parameterizedDeletionScriptId; private final boolean supportsGeoShapePrefixTree; private final CircleProcessor bdbCircleProcessor; + private final Set temporaryErrorCodes; + private final boolean temporaryTransportFailures; public ElasticSearchIndex(Configuration config) throws BackendException { indexName = determineIndexName(config); @@ -413,6 +446,9 @@ public ElasticSearchIndex(Configuration config) throws BackendException { batchSize = config.get(INDEX_MAX_RESULT_SET_SIZE); log.debug("Configured ES query nb result by query to {}", batchSize); bdbCircleProcessor = MixedIndexUtilsConfigOptions.buildBKDCircleProcessor(config); + temporaryErrorCodes = Arrays.stream(config.get(TEMPORARY_ERROR_CODES)) + .mapToInt(Integer::parseInt).boxed().collect(Collectors.toSet()); + temporaryTransportFailures = config.get(TEMPORARY_TRANSPORT_FAILURES); client = interfaceConfiguration(config).getClient(); supportsGeoShapePrefixTree = client.getMajorVersion().getValue() <= 7; @@ -541,13 +577,67 @@ private ElasticSearchSetup.Connection interfaceConfiguration(Configuration confi } private BackendException convert(Exception esException) { - if (esException instanceof InterruptedException) { + return convert(esException, temporaryErrorCodes, temporaryTransportFailures); + } + + //Only a TemporaryBackendException is reattempted by BackendOperation, so a transient failure classified as + //permanent means the index mutation is dropped rather than reattempted + @VisibleForTesting + static BackendException convert(Exception esException, Set temporaryErrorCodes, + boolean temporaryTransportFailures) { + final Throwable temporaryCause = findTemporaryCause(esException, temporaryErrorCodes, + temporaryTransportFailures); + if (temporaryCause instanceof InterruptedException) { + //Throwing the InterruptedException cleared the interrupt status of the thread, and the exception itself + //is consumed here, so restore the status. BackendOperation reattempts a temporary failure and relies on + //the status to abort that wait, without which a cancelled operation keeps reattempting the mutation for + //the whole write time budget + Thread.currentThread().interrupt(); return new TemporaryBackendException("Interrupted while waiting for response", esException); + } else if (temporaryCause != null) { + return new TemporaryBackendException("Temporary exception while executing index operation", esException); } else { return new PermanentBackendException("Unknown exception while executing index operation", esException); } } + //Returns the first cause indicating that the operation may succeed if reattempted, or null if the failure is not + //recognised as transient. The whole chain is inspected because the Elasticsearch client wraps the failure, and + //wraps an interrupt during a client side retry wait in a RuntimeException + private static Throwable findTemporaryCause(Throwable throwable, Set temporaryErrorCodes, + boolean temporaryTransportFailures) { + for (Throwable cause = throwable; cause != null; cause = cause.getCause()) { + if (cause instanceof InterruptedException) { + return cause; + } else if (cause instanceof ResponseException) { + final int statusCode = ((ResponseException) cause).getResponse().getStatusLine().getStatusCode(); + if (temporaryErrorCodes.contains(statusCode)) { + return cause; + } + } else if (cause instanceof ElasticSearchBulkFailureException) { + //A bulk request is only worth reattempting if every item which failed did so transiently + final Set statusCodes = ((ElasticSearchBulkFailureException) cause).getFailedItemStatusCodes(); + if (!statusCodes.isEmpty() && temporaryErrorCodes.containsAll(statusCodes)) { + return cause; + } + } else if (temporaryTransportFailures && isTransportFailure(cause)) { + return cause; + } + } + return null; + } + + //Whether the failure occurred before Elasticsearch could produce an HTTP response, and so has no status code to + //classify on + private static boolean isTransportFailure(Throwable cause) { + //SocketException covers ConnectException and a connection reset by the peer, while InterruptedIOException + //covers SocketTimeoutException and ConnectTimeoutException + return cause instanceof SocketException + || cause instanceof InterruptedIOException + || cause instanceof NoHttpResponseException + || cause instanceof ConnectionClosedException; + } + private static String getDualMappingName(String key) { return key + STRING_MAPPING_SUFFIX; } @@ -909,8 +999,15 @@ public void mutate(Map> mutations, KeyInforma client.bulkRequest(requests, null); } } catch (final Exception e) { - log.error("Failed to execute bulk Elasticsearch mutation", e); - throw convert(e); + final BackendException converted = convert(e); + //Reserve the error level for a mutation which is about to be dropped: a temporary failure is reattempted + //by BackendOperation, and is only lost if the write time budget runs out, which commit reports itself + if (converted instanceof TemporaryBackendException) { + log.warn("Transient failure while executing bulk Elasticsearch mutation", e); + } else { + log.error("Failed to execute bulk Elasticsearch mutation", e); + } + throw converted; } } diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/rest/RestElasticSearchClient.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/rest/RestElasticSearchClient.java index f399863065..51e5a7473a 100644 --- a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/rest/RestElasticSearchClient.java +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/rest/RestElasticSearchClient.java @@ -39,6 +39,7 @@ import org.elasticsearch.client.RestClient; import org.janusgraph.core.attribute.Geoshape; import org.janusgraph.diskstorage.es.ElasticMajorVersion; +import org.janusgraph.diskstorage.es.ElasticSearchBulkFailureException; import org.janusgraph.diskstorage.es.ElasticSearchClient; import org.janusgraph.diskstorage.es.ElasticSearchMutation; import org.janusgraph.diskstorage.es.mapping.IndexMapping; @@ -569,7 +570,9 @@ public void bulkRequest(final List requests, String inges } else { final List errorItems = bulkItemsThatFailed.stream().map(Triplet::getValue0).collect(Collectors.toList()); errorItems.forEach(error -> log.error("Failed to execute ES query: {}", error)); - throw new IOException("Failure(s) in Elasticsearch bulk request: " + errorItems); + //Retain the item statuses so callers can classify the failure as transient or permanent + throw new ElasticSearchBulkFailureException( + "Failure(s) in Elasticsearch bulk request: " + errorItems, errorCodes); } } else { //The entire bulk request was successful, leave the loop diff --git a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticSearchExceptionConversionTest.java b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticSearchExceptionConversionTest.java new file mode 100644 index 0000000000..2d1a3262a7 --- /dev/null +++ b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticSearchExceptionConversionTest.java @@ -0,0 +1,234 @@ +// Copyright 2026 JanusGraph Authors +// +// Licensed 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.janusgraph.diskstorage.es; + +import com.google.common.collect.ImmutableSet; +import org.apache.http.ConnectionClosedException; +import org.apache.http.NoHttpResponseException; +import org.apache.http.StatusLine; +import org.apache.http.conn.ConnectTimeoutException; +import org.elasticsearch.client.Response; +import org.elasticsearch.client.ResponseException; +import org.janusgraph.diskstorage.BackendException; +import org.janusgraph.diskstorage.PermanentBackendException; +import org.janusgraph.diskstorage.TemporaryBackendException; +import org.janusgraph.diskstorage.util.BackendOperation; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.IOException; +import java.net.ConnectException; +import java.net.SocketException; +import java.net.SocketTimeoutException; +import java.time.Duration; +import java.util.Collections; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Stream; + +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.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +//Tests how ElasticSearchIndex classifies a failed index operation. Only a TemporaryBackendException is reattempted by +//BackendOperation, so a transient failure classified as permanent means the index mutation is dropped and the mixed +//index is left inconsistent with the graph +public class ElasticSearchExceptionConversionTest { + + private static final Set DEFAULT_TEMPORARY_ERROR_CODES = ImmutableSet.of(429, 502, 503, 504); + + @AfterEach + public void clearInterruptStatus() { + //convert restores the interrupt status of an interrupted failure, which must not leak into the next test + Thread.interrupted(); + } + + private static BackendException convert(Exception esException) { + return ElasticSearchIndex.convert(esException, DEFAULT_TEMPORARY_ERROR_CODES, true); + } + + private static ResponseException responseException(int statusCode) { + final StatusLine statusLine = mock(StatusLine.class); + when(statusLine.getStatusCode()).thenReturn(statusCode); + final Response response = mock(Response.class); + when(response.getStatusLine()).thenReturn(statusLine); + final ResponseException responseException = mock(ResponseException.class); + when(responseException.getResponse()).thenReturn(response); + return responseException; + } + + private static ElasticSearchBulkFailureException bulkFailure(Integer... failedItemStatusCodes) { + return new ElasticSearchBulkFailureException("Failure(s) in Elasticsearch bulk request: [an error]", + Stream.of(failedItemStatusCodes).collect(ImmutableSet.toImmutableSet())); + } + + @ParameterizedTest + @ValueSource(ints = {429, 502, 503, 504}) + public void shouldConvertTransientResponseStatusToTemporary(int statusCode) { + assertInstanceOf(TemporaryBackendException.class, convert(responseException(statusCode))); + } + + @ParameterizedTest + @ValueSource(ints = {400, 401, 403, 404, 409}) + public void shouldConvertPermanentResponseStatusToPermanent(int statusCode) { + assertInstanceOf(PermanentBackendException.class, convert(responseException(statusCode))); + } + + @Test + public void shouldConvertConfiguredResponseStatusToTemporary() { + final ResponseException requestTimeout = responseException(408); + assertInstanceOf(PermanentBackendException.class, convert(requestTimeout)); + assertInstanceOf(TemporaryBackendException.class, + ElasticSearchIndex.convert(requestTimeout, ImmutableSet.of(408), true)); + } + + @Test + public void shouldConvertResponseStatusToPermanentWhenNoTemporaryCodesConfigured() { + assertInstanceOf(PermanentBackendException.class, + ElasticSearchIndex.convert(responseException(503), Collections.emptySet(), true)); + } + + @Test + public void shouldInspectWholeCauseChain() { + //RestClient.extractAndWrapCause rewraps the failure of a node attempt, so the exception carrying the status + //is reached only through the cause chain + assertInstanceOf(TemporaryBackendException.class, convert(new IOException( + "method [POST], host [http://localhost:9200], status line [HTTP/1.1 503 Service Unavailable]", + responseException(503)))); + } + + @Test + public void shouldConvertBulkFailureOfEntirelyTransientItemsToTemporary() { + assertInstanceOf(TemporaryBackendException.class, convert(bulkFailure(429, 503))); + } + + @Test + public void shouldConvertPartiallyPermanentBulkFailureToPermanent() { + //A reattempt of the whole batch cannot succeed while one of its items fails permanently + assertInstanceOf(PermanentBackendException.class, convert(bulkFailure(429, 400))); + } + + @Test + public void shouldConvertBulkFailureWithoutStatusCodesToPermanent() { + assertInstanceOf(PermanentBackendException.class, convert(bulkFailure())); + } + + @Test + public void shouldConvertTransportFailuresToTemporary() { + //None of these produce an HTTP response, so no status code is available to classify on + assertInstanceOf(TemporaryBackendException.class, convert(new ConnectException("Connection refused"))); + assertInstanceOf(TemporaryBackendException.class, convert(new SocketException("Connection reset"))); + assertInstanceOf(TemporaryBackendException.class, convert(new SocketTimeoutException("30,000 milliseconds timeout on connection"))); + assertInstanceOf(TemporaryBackendException.class, convert(new ConnectTimeoutException("connect timed out"))); + assertInstanceOf(TemporaryBackendException.class, convert(new NoHttpResponseException("failed to respond"))); + assertInstanceOf(TemporaryBackendException.class, convert(new ConnectionClosedException("Connection closed"))); + } + + @Test + public void shouldConvertTransportFailuresToPermanentWhenDisabled() { + assertInstanceOf(PermanentBackendException.class, + ElasticSearchIndex.convert(new SocketTimeoutException("read timed out"), + DEFAULT_TEMPORARY_ERROR_CODES, false)); + //Disabling transport failure classification must not affect classification by status code + assertInstanceOf(TemporaryBackendException.class, + ElasticSearchIndex.convert(responseException(503), DEFAULT_TEMPORARY_ERROR_CODES, false)); + } + + @Test + public void shouldConvertInterruptToTemporary() { + final InterruptedException interrupted = new InterruptedException(); + BackendException converted = convert(interrupted); + assertInstanceOf(TemporaryBackendException.class, converted); + assertEquals("Interrupted while waiting for response", converted.getMessage()); + assertSame(interrupted, converted.getCause()); + + //An interrupt during a client side retry wait reaches convert wrapped in a RuntimeException + converted = convert(new RuntimeException("Thread interrupted while waiting for retry attempt 1 of 3", + interrupted)); + assertInstanceOf(TemporaryBackendException.class, converted); + assertEquals("Interrupted while waiting for response", converted.getMessage()); + } + + @Test + public void shouldRestoreInterruptStatusOfInterruptedFailure() { + //Throwing the InterruptedException cleared the status, and convert consumes the exception itself, so the + //status is the only remaining signal that the operation was cancelled + assertFalse(Thread.currentThread().isInterrupted()); + convert(new RuntimeException("Thread interrupted while waiting for retry attempt 1 of 3", + new InterruptedException())); + assertTrue(Thread.currentThread().isInterrupted()); + } + + @Test + public void shouldNotReattemptInterruptedFailure() { + //BackendOperation aborts its backoff wait only while the interrupt status is set. Without it a cancelled + //commit would keep reissuing the bulk request for the whole write time budget + final AtomicInteger attempts = new AtomicInteger(); + assertThrows(PermanentBackendException.class, () -> BackendOperation.executeDirect(() -> { + attempts.incrementAndGet(); + throw convert(new RuntimeException("Thread interrupted while waiting for retry attempt 1 of 3", + new InterruptedException())); + }, Duration.ofSeconds(30))); + assertEquals(1, attempts.get()); + } + + @Test + public void shouldConvertUnrecognisedFailureToPermanent() { + final IllegalArgumentException tooLarge = new IllegalArgumentException( + "Bulk request item(s) larger than permitted chunk limit."); + final BackendException converted = convert(tooLarge); + assertInstanceOf(PermanentBackendException.class, converted); + assertSame(tooLarge, converted.getCause()); + } + + @Test + public void shouldRetainCauseOfTemporaryFailure() { + final ResponseException responseException = responseException(503); + final IOException wrapper = new IOException(responseException); + assertSame(wrapper, convert(wrapper).getCause()); + } + + @Test + public void shouldReattemptTransientFailureUntilItSucceeds() throws BackendException { + //IndexTransaction submits the mutation through BackendOperation, which reattempts it only while it fails + //temporarily + final AtomicInteger attempts = new AtomicInteger(); + final boolean mutated = BackendOperation.executeDirect(() -> { + if (attempts.incrementAndGet() < 3) { + throw convert(responseException(503)); + } + return true; + }, Duration.ofSeconds(30)); + assertTrue(mutated); + assertEquals(3, attempts.get()); + } + + @Test + public void shouldNotReattemptPermanentFailure() { + final AtomicInteger attempts = new AtomicInteger(); + assertThrows(PermanentBackendException.class, () -> BackendOperation.executeDirect(() -> { + attempts.incrementAndGet(); + throw convert(responseException(400)); + }, Duration.ofSeconds(30))); + assertEquals(1, attempts.get()); + } +} diff --git a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/rest/RestClientRetryTest.java b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/rest/RestClientRetryTest.java index 9c9e101977..ee62c48348 100644 --- a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/rest/RestClientRetryTest.java +++ b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/rest/RestClientRetryTest.java @@ -22,6 +22,7 @@ import org.elasticsearch.client.Response; import org.elasticsearch.client.ResponseException; import org.elasticsearch.client.RestClient; +import org.janusgraph.diskstorage.es.ElasticSearchBulkFailureException; import org.janusgraph.diskstorage.es.ElasticSearchMutation; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -126,6 +127,44 @@ public void testRetryOfIndividuallyFailedBulkItems() throws IOException { } } + @Test + public void testFailedBulkItemsRetainTheirStatusCodes() throws IOException { + ObjectMapper mapper = new ObjectMapper(); + //A bulk response returns a success despite underlying items having failed + when(statusLine.getStatusCode()).thenReturn(200); + + RestBulkResponse.RestBulkItemResponse rejectedItem = new RestBulkResponse.RestBulkItemResponse(); + rejectedItem.setError("es_rejected_execution_exception"); + rejectedItem.setStatus(429); + RestBulkResponse.RestBulkItemResponse unmappedItem = new RestBulkResponse.RestBulkItemResponse(); + unmappedItem.setError("mapper_parsing_exception"); + unmappedItem.setStatus(400); + RestBulkResponse bulkResponse = new RestBulkResponse(); + bulkResponse.setItems( + Stream.of( + Collections.singletonMap("index", rejectedItem), + Collections.singletonMap("index", unmappedItem) + ).collect(Collectors.toList()) + ); + HttpEntity httpEntityMock = mock(HttpEntity.class); + when(httpEntityMock.getContent()).thenReturn(new ByteArrayInputStream(mapper.writeValueAsBytes(bulkResponse))); + Response responseMock = mock(Response.class); + when(responseMock.getEntity()).thenReturn(httpEntityMock); + when(responseMock.getStatusLine()).thenReturn(statusLine); + + //No retries are configured, so the item failures are reported immediately + try (RestElasticSearchClient restClientUnderTest = createClient(0, Collections.emptySet())) { + when(restClientMock.performRequest(any())).thenReturn(responseMock); + restClientUnderTest.bulkRequest(Arrays.asList( + ElasticSearchMutation.createDeleteRequest("some_index", "some_type", "some_doc_id1"), + ElasticSearchMutation.createDeleteRequest("some_index", "some_type", "some_doc_id2") + ), null); + Assertions.fail("Should have thrown for the failed bulk items"); + } catch (ElasticSearchBulkFailureException e) { + Assertions.assertEquals(Sets.newHashSet(429, 400), e.getFailedItemStatusCodes()); + } + } + @Test public void testRetryOnConfiguredErrorStatus() throws IOException { Integer retryCode = 429;