JanusGraph · batrived · Aug 11, 2026
diff --git 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
@@ -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/...aph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchBulkFailureException.java b/...aph-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<Integer> failedItemStatusCodes;
+
+    public ElasticSearchBulkFailureException(String message, Set<Integer> 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<Integer> 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
@@ -14,13 +14,17 @@
 
 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;
 import com.google.common.collect.Iterators;
 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<String[]> 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<Boolean> 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<Integer> 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<Integer> 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<Integer> 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<Integer> 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<Integer> 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<String, Map<String, IndexMutation>> 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
@@ -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<ElasticSearchMutation> requests, String inges
                         } else {
                             final List<Object> 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