From 746d2076ce0301d97a3886d8a3a94a808d2bfe6a Mon Sep 17 00:00:00 2001 From: Balmukund Trivedi Date: Tue, 11 Aug 2026 13:42:01 -0700 Subject: [PATCH] Retry transient Elasticsearch failures instead of dropping the index mutation (#4925) ElasticSearchIndex reported every failure except an interrupt as a PermanentBackendException. A commit applies index mutations after the storage mutations, and cannot roll them back. A transient failure therefore dropped the mutation with one ERROR log line. Examples are a rolling restart, a saturated write queue, and a socket timeout during a GC pause. The mixed index then stayed inconsistent with the graph until a reindex or a transaction-log recovery repaired it. Classify such a failure as a TemporaryBackendException. BackendOperation already reattempts that exception with exponential backoff for up to storage.write-time. Two new options control the classification, and both are enabled by default. temporary-error-codes lists the HTTP status codes which are transient (429, 502, 503 and 504). temporary-transport-failures covers the failures which produce no HTTP response, and which no status code can match. A bulk request reports item level failures inside a successful HTTP response, so the enclosing response gives no status to classify on. Retain the item statuses in the new ElasticSearchBulkFailureException. Reattempt the request only if every item which failed did so transiently, because a batch which holds a permanently failing item cannot succeed on a reattempt. A mapping conflict is one such item. Restore the interrupt status of the thread after the InterruptedException has been consumed. BackendOperation needs that status to abort its backoff wait. Without it, a cancelled commit continued to reissue the bulk request for the whole write time budget. Log a transient failure at WARN, and keep ERROR for a mutation which is about to be dropped. Signed-off-by: Balmukund Trivedi Co-Authored-By: Claude Opus 5 (1M context) --- docs/changelog.md | 42 ++++ docs/configs/janusgraph-cfg.md | 2 + .../es/ElasticSearchBulkFailureException.java | 42 ++++ .../diskstorage/es/ElasticSearchIndex.java | 103 +++++++- .../es/rest/RestElasticSearchClient.java | 5 +- .../ElasticSearchExceptionConversionTest.java | 234 ++++++++++++++++++ .../es/rest/RestClientRetryTest.java | 39 +++ 7 files changed, 463 insertions(+), 4 deletions(-) create mode 100644 janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchBulkFailureException.java create mode 100644 janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/ElasticSearchExceptionConversionTest.java 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;