Elasticsearch: transient failures classified as PermanentBackendException, so index mutations are silently dropped instead of retried

[batrived]

• Version: master (ac0eb23) — also all released versions; convert() has had this shape since the Titan era
• Storage Backend: any (behaviour is not backend-specific)
• Mixed Index Backend: elasticsearch
• Expected Behavior: transient Elasticsearch failures (HTTP 429/502/503/504, connection reset, socket timeout) should be classified as TemporaryBackendException, so that the retry loop already present in BackendOperation absorbs them within the storage.write-time budget.
• Current Behavior: every exception except InterruptedException is classified as PermanentBackendException, so transient failures are never retried. The index mutation is dropped, and because index mutations are applied after storage in StandardJanusGraph.commit(), the failure cannot be rolled back — it is only logged, leaving the mixed index permanently inconsistent with the graph.

Details

ElasticSearchIndex.convert() collapses every failure mode into one bucket:

janusgraph/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchIndex.java

Lines 543 to 549 in ac0eb23

	private BackendException convert(Exception esException) {
	if (esException instanceof InterruptedException) {
	return new TemporaryBackendException("Interrupted while waiting for response", esException);
	} else {
	return new PermanentBackendException("Unknown exception while executing index operation", esException);
	}
	}

private BackendException convert(Exception esException) {
    if (esException instanceof InterruptedException) {
        return new TemporaryBackendException("Interrupted while waiting for response", esException);
    } else {
        return new PermanentBackendException("Unknown exception while executing index operation", esException);
    }
}

The retry machinery already exists and is already wired up, so this is a classification fix rather than new functionality. IndexTransaction.flushInternal() runs the mutation through BackendOperation.execute(..., maxWriteTime):

janusgraph/janusgraph-core/src/main/java/org/janusgraph/diskstorage/indexing/IndexTransaction.java

Lines 141 to 159 in ac0eb23

	private void flushInternal() throws BackendException {
	if (mutations!=null && !mutations.isEmpty()) {
	//Consolidate all mutations prior to persistence to ensure that no addition accidentally gets swallowed by a delete
	for (Map<String, IndexMutation> store : mutations.values()) {
	for (IndexMutation mut : store.values()) mut.consolidate();
	}
	BackendOperation.execute(new Callable<Boolean>() {
	@Override
	public Boolean call() throws Exception {
	index.mutate(mutations, keyInformation, indexTx);
	return true;
	}
	@Override
	public String toString() {
	return "IndexMutation";
	}
	}, maxWriteTime);

and BackendOperation.executeDirect implements jittered exponential backoff — but only when the innermost BackendException is a TemporaryBackendException:

janusgraph/janusgraph-core/src/main/java/org/janusgraph/diskstorage/util/BackendOperation.java

Lines 59 to 98 in ac0eb23

	public static <V> V executeDirect(Callable<V> exe, Duration totalWaitTime) throws BackendException {
	Preconditions.checkArgument(!totalWaitTime.isZero(),"Need to specify a positive waitTime: %s",totalWaitTime);
	long maxTime = System.currentTimeMillis()+totalWaitTime.toMillis();
	Duration waitTime = pertubTime(BASE_REATTEMPT_TIME);
	BackendException lastException;
	while (true) {
	try {
	return exe.call();
	} catch (final Throwable e) {
	//Find inner-most StorageException
	Throwable ex = e;
	BackendException storeEx = null;
	do {
	if (ex instanceof BackendException) storeEx = (BackendException)ex;
	} while ((ex=ex.getCause())!=null);
	if (storeEx!=null && storeEx instanceof TemporaryBackendException) {
	lastException = storeEx;
	} else if (e instanceof BackendException) {
	throw (BackendException)e;
	} else {
	throw new PermanentBackendException("Permanent exception while executing backend operation "+exe.toString(),e);
	}
	}
	//Wait and retry
	assert lastException!=null;
	if (System.currentTimeMillis()+waitTime.toMillis()<maxTime) {
	log.info("Temporary exception during backend operation ["+exe.toString()+"]. Attempting backoff retry.",lastException);
	try {
	Thread.sleep(waitTime.toMillis());
	} catch (InterruptedException r) {
	// added thread interrupt signal to support traversal interruption
	Thread.currentThread().interrupt();
	throw new PermanentBackendException("Interrupted while waiting to retry failed backend operation", r);
	}
	} else {
	break;
	}
	waitTime = pertubTime(waitTime.multipliedBy(2));
	}
	throw new TemporaryBackendException("Could not successfully complete backend operation due to repeated temporary exceptions after "+totalWaitTime,lastException);

So today:

• ES returns 429 es_rejected_execution_exception (write threadpool queue full) → PermanentBackendException → no retry → mutation dropped.
• ES returns 502/503/504 (shard unavailable, gateway restart, rolling upgrade) → same.
• Socket timeout or connection reset mid-bulk → same.

Because commit() commits storage first and then collects index failures rather than aborting:

janusgraph/janusgraph-core/src/main/java/org/janusgraph/graphdb/database/StandardJanusGraph.java

Lines 1047 to 1054 in ac0eb23

	//2. Commit indexes - [FAILURE] all exceptions are collected and logged but nothing is aborted
	indexFailures = mutator.commitIndexes();
	if (!indexFailures.isEmpty()) {
	status = LogTxStatus.SECONDARY_FAILURE;
	for (Map.Entry<String,Throwable> entry : indexFailures.entrySet()) {
	log.error("Error while committing index mutations for transaction ["+transactionId+"] on index: " +entry.getKey(),entry.getValue());
	}
	}

the dropped mutation cannot be undone. Unless log-tx is enabled (default false) and a transaction-recovery processor is running, the divergence is permanent and the only signal is a single ERROR log line.

The net effect is that a brief and entirely routine Elasticsearch event — a rolling restart, a hot shard rejecting a bulk, a GC pause causing a socket timeout — permanently desynchronizes a mixed index, in a situation the existing retry loop was designed to absorb.

Steps to Reproduce

1. Configure a graph with an Elasticsearch mixed index.
2. Make ES return a retryable error for a bulk request — e.g. saturate the write threadpool queue so bulk items fail with es_rejected_execution_exception, or interpose a proxy that returns 503 for one request.
3. Commit a transaction that writes to that mixed index.
4. Observe: Error while committing index mutations for transaction [...] is logged once, no retry is attempted, and the document is absent from ES while the element is present in the storage backend.

Suggested Fix

Classify on transport / HTTP status rather than a single instanceof:

• TemporaryBackendException: HTTP 429, 502, 503, 504; ConnectException, SocketTimeoutException, ConnectionClosedException, NoHttpResponseException; InterruptedException (unchanged)
• PermanentBackendException: HTTP 400 (including mapper_parsing_exception), 401/403, 404, and anything unrecognised

The ES REST client exposes the status via ResponseException.getResponse().getStatusLine().getStatusCode(), so this can be handled entirely inside convert() without touching call sites. Making the retryable status set configurable would additionally let operators adapt to a new failure mode without waiting for a release.

Glad to open a PR for this if the approach sounds reasonable.

[batrived]

Clarifying the scope of this, since master already has retry machinery that I should have accounted for up front.

master does have Elasticsearch-client-level retries: performRequestWithRetry, plus per-item bulk retry in bulkRequest via pairErrorsWithSubmittedMutation, governed by retry-limit, retry-error-codes, retry-initial-wait and retry-max-wait. The request in this issue is still worth making, for three reasons:

1. It is off by default. retry-limit defaults to 0 and retry-error-codes defaults to an empty array, so a default deployment performs no retries at all and every transient failure is still classified permanent and dropped.

2. It cannot cover transport-level failures. performRequestWithRetry only catches ResponseException, which requires an HTTP response to exist. ConnectException, SocketTimeoutException, ConnectionClosedException and NoHttpResponseException never produce one, so no value of retry-error-codes can make them retryable. Those are precisely the cases that need classifying in convert().

3. The two layers do not compose. Client-level waits are min(retry-initial-wait * 10^attempt, retry-max-wait) — with the defaults, at most 1s per attempt. BackendOperation retries up to storage.write-time, default 100s. But once client retries are exhausted the resulting exception still passes through convert() and becomes a PermanentBackendException, so the much longer outer budget never engages. Classifying correctly in convert() is what makes the outer loop usable as a second, longer-horizon layer — which is the one that matters for events like a rolling restart.

So the suggested change is complementary to retry-error-codes rather than a duplicate of it: keep the fast client-level retries for known status codes, and additionally let genuinely transient failures reach BackendOperation as TemporaryBackendException.