From 546ff09498c52e0176b624ddc2d952529582701a Mon Sep 17 00:00:00 2001 From: Balmukund Trivedi Date: Tue, 11 Aug 2026 14:53:25 -0700 Subject: [PATCH 1/2] Report a bulk update which failed because the document is missing (#4926) pairErrorsWithSubmittedMutation removed every bulk item whose status is 404 from the failure list, whatever operation produced it. That is correct for a delete, because deleting an absent document leaves the index in the state the deletion asked for. An update which returns 404 is a document_missing_exception: the write did not happen. Changing the value of a SINGLE cardinality indexed property produces a deletion of the old value and an addition of the new one against the same document, with isNew false. Both become update operations, and because the mutation has deletions, mutate() withholds the upsert document. If the Elasticsearch document is absent, both items return 404, both were discarded, and the property was never indexed. Every later update to that element took the same path, so the document was never recreated. The element stayed invisible to the mixed index while present in the storage backend, and nothing reported it: mutate returned normally, so even the mutate.exceptions metric stayed at zero. Retain the request type on RequestBytes, which already reads it to decide on the retry_on_conflict parameter but did not keep it, and limit the exemption to delete items. Signed-off-by: Balmukund Trivedi Co-Authored-By: Claude Opus 5 (1M context) --- .../es/rest/RestElasticSearchClient.java | 17 +- .../es/rest/RestClientBulkItemStatusTest.java | 161 ++++++++++++++++++ 2 files changed, 176 insertions(+), 2 deletions(-) create mode 100644 janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/rest/RestClientBulkItemStatusTest.java 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..6c480012df 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 @@ -398,9 +398,12 @@ public void clearStore(String indexName, String storeName) throws IOException { class RequestBytes { final byte [] requestBytes; final byte [] requestSource; + //Retained so that a failed bulk item can be interpreted against the operation which produced it + final ElasticSearchMutation.RequestType requestType; @VisibleForTesting RequestBytes(final ElasticSearchMutation request) throws JsonProcessingException { + this.requestType = request.getRequestType(); Map requestData = new HashMap<>(); if (useMappingTypes) { requestData.put("_index", request.getIndex()); @@ -474,13 +477,23 @@ private List> pairErrorsWithSubmittedMuta throw new IllegalStateException("There should only be a single item per bulk reponse item entry"); } RestBulkResponse.RestBulkItemResponse item = bulkResponseItem.iterator().next(); - if (item.getError() != null && item.getStatus() != HttpStatus.SC_NOT_FOUND) { - errors.add(Triplet.with(item.getError(), item.getStatus(), submittedBulkRequestItems.get(itemIndex))); + final RequestBytes submittedItem = submittedBulkRequestItems.get(itemIndex); + if (item.getError() != null && !isAbsentDocumentDeletion(item, submittedItem)) { + errors.add(Triplet.with(item.getError(), item.getStatus(), submittedItem)); } } return errors; } + //Deleting a document which is already absent leaves the index in the state the deletion asked for, so its 404 is + //a success. An update which returns 404 is a document_missing_exception: the write did not happen, and treating it + //as a success drops the mutation with nothing reported + private static boolean isAbsentDocumentDeletion(final RestBulkResponse.RestBulkItemResponse item, + final RequestBytes submittedItem) { + return item.getStatus() == HttpStatus.SC_NOT_FOUND + && submittedItem.requestType == ElasticSearchMutation.RequestType.DELETE; + } + @VisibleForTesting class BulkRequestChunker implements Iterator> { //By default, Elasticsearch writes are limited to 100mb, so chunk a given batch of requests so they stay under diff --git a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/rest/RestClientBulkItemStatusTest.java b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/rest/RestClientBulkItemStatusTest.java new file mode 100644 index 0000000000..953844b39d --- /dev/null +++ b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/rest/RestClientBulkItemStatusTest.java @@ -0,0 +1,161 @@ +// 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.rest; + +import com.google.common.collect.ImmutableMap; +import org.apache.http.HttpEntity; +import org.apache.http.StatusLine; +import org.apache.tinkerpop.shaded.jackson.databind.ObjectMapper; +import org.elasticsearch.client.Response; +import org.elasticsearch.client.RestClient; +import org.janusgraph.diskstorage.es.ElasticSearchMutation; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.when; + +//A bulk response reports item level failures inside an otherwise successful HTTP response. Every 404 used to be +//treated as a success regardless of the operation which produced it. That is right for a delete, which is idempotent, +//but an update returning 404 is a document_missing_exception: the write did not happen. Because a property change on a +//SINGLE cardinality key produces a deletion and an addition against the same document, mutate() withholds the upsert, +//so both items 404 when the document is absent and the element is never indexed again. +@ExtendWith(MockitoExtension.class) +public class RestClientBulkItemStatusTest { + + private static final String INDEX = "some_index"; + private static final String TYPE = "some_type"; + + @Mock + private RestClient restClientMock; + + @Mock + private StatusLine statusLine; + + private RestElasticSearchClient createClient() throws IOException { + when(restClientMock.performRequest(any())).thenThrow(new IOException()); + final RestElasticSearchClient clientUnderTest = new RestElasticSearchClient(restClientMock, 0, false, + 0, Collections.emptySet(), 0, 0, 100_000_000); + Mockito.reset(restClientMock); + return clientUnderTest; + } + + //Builds a bulk response in which each submitted item reports the given status, and a non-null error when the + //status is a failure + private Response bulkResponseWith(List operations, List statuses) throws IOException { + final RestBulkResponse bulkResponse = new RestBulkResponse(); + final List> items = new ArrayList<>(); + for (int i = 0; i < operations.size(); i++) { + final RestBulkResponse.RestBulkItemResponse item = new RestBulkResponse.RestBulkItemResponse(); + final int status = statuses.get(i); + item.setStatus(status); + if (status >= 300) { + item.setError(status == 404 ? "document_missing_exception" : "an_error"); + } + items.add(Collections.singletonMap(operations.get(i), item)); + } + bulkResponse.setItems(items); + + final HttpEntity entity = Mockito.mock(HttpEntity.class); + when(entity.getContent()).thenReturn(new ByteArrayInputStream(new ObjectMapper().writeValueAsBytes(bulkResponse))); + when(statusLine.getStatusCode()).thenReturn(200); + final Response response = Mockito.mock(Response.class); + when(response.getEntity()).thenReturn(entity); + when(response.getStatusLine()).thenReturn(statusLine); + return response; + } + + private static ElasticSearchMutation update(String id) { + return ElasticSearchMutation.createUpdateRequest(INDEX, TYPE, id, + ImmutableMap.builder().put("doc", ImmutableMap.of("name", "value")), null); + } + + private void bulkRequest(List mutations, List operations, List statuses) + throws IOException { + try (RestElasticSearchClient clientUnderTest = createClient()) { + //Built before the stubbing below, because it stubs mocks of its own + final Response bulkResponse = bulkResponseWith(operations, statuses); + when(restClientMock.performRequest(any())).thenReturn(bulkResponse); + clientUnderTest.bulkRequest(mutations, null); + } + } + + @Test + public void shouldTreatTheDeletionOfAnAbsentDocumentAsSuccess() throws IOException { + //Nothing is thrown: the index is already in the state the deletion asked for + bulkRequest(Collections.singletonList(ElasticSearchMutation.createDeleteRequest(INDEX, TYPE, "doc1")), + Collections.singletonList("delete"), Collections.singletonList(404)); + } + + @Test + public void shouldReportAnUpdateOfAMissingDocument() throws IOException { + final IOException e = assertThrows(IOException.class, + () -> bulkRequest(Collections.singletonList(update("doc1")), + Collections.singletonList("update"), Collections.singletonList(404))); + assertTrue(e.getMessage().contains("document_missing_exception"), e.getMessage()); + } + + @Test + public void shouldReportAnIndexRequestWhichReturnedNotFound() throws IOException { + //An index request creates the document, so a 404 is not something it can ask for + assertThrows(IOException.class, () -> bulkRequest( + Collections.singletonList(ElasticSearchMutation.createIndexRequest(INDEX, TYPE, "doc1", + ImmutableMap.of("name", "value"))), + Collections.singletonList("index"), Collections.singletonList(404))); + } + + @Test + public void shouldReportAMissingDocumentUpdateSubmittedAlongsideADeletion() throws IOException { + //This is the shape mutate() produces for a value change on a SINGLE cardinality key: the field deletion script + //and the addition script against the same absent document + final IOException e = assertThrows(IOException.class, () -> bulkRequest( + Arrays.asList(update("doc1"), update("doc1")), + Arrays.asList("update", "update"), Arrays.asList(404, 404))); + assertTrue(e.getMessage().contains("document_missing_exception"), e.getMessage()); + } + + @Test + public void shouldStillReportOtherFailuresAndIgnoreSuccesses() throws IOException { + final IOException e = assertThrows(IOException.class, () -> bulkRequest( + Arrays.asList(update("doc1"), update("doc2")), + Arrays.asList("update", "update"), Arrays.asList(200, 400))); + assertTrue(e.getMessage().contains("an_error"), e.getMessage()); + //Only the failed item is reported + assertEquals(1, e.getMessage().split("an_error", -1).length - 1, e.getMessage()); + } + + @Test + public void shouldTreatADeletionAlongsideAFailedUpdateAsOnlyOneFailure() throws IOException { + final IOException e = assertThrows(IOException.class, () -> bulkRequest( + Arrays.asList(ElasticSearchMutation.createDeleteRequest(INDEX, TYPE, "doc1"), update("doc1")), + Arrays.asList("delete", "update"), Arrays.asList(404, 404))); + //The delete is exempt, the update is not + assertEquals(1, e.getMessage().split("document_missing_exception", -1).length - 1, e.getMessage()); + } +} From 3053b67841ee45e7f3c601b8213b229b42b19f38 Mon Sep 17 00:00:00 2001 From: Balmukund Trivedi Date: Wed, 12 Aug 2026 10:03:15 -0700 Subject: [PATCH 2/2] Treat a missing document as success for a field deletion as well as a delete The previous commit exempted only RequestType.DELETE from the rule that a 404 in a bulk item is a lost write. mutate() sends a field deletion as an UPDATE which runs the parameterized deletion script, so that exemption is too narrow: a document with no fields left to delete is already the state the mutation asked for, and reporting it fails index maintenance which removes stale records and any transaction which deletes a field of a document another transaction removed. Carry the distinction on the mutation instead of reading it off the request type. A whole document deletion and a field deletion both only take content out of the index. An addition is what a 404 loses, and mutate() withholds the upsert from the addition once the mutation also has deletions, which is the case the previous commit set out to report. Signed-off-by: Balmukund Trivedi Co-Authored-By: Claude Opus 5 (1M context) --- .../diskstorage/es/ElasticSearchIndex.java | 4 +-- .../diskstorage/es/ElasticSearchMutation.java | 26 +++++++++++--- .../es/rest/RestElasticSearchClient.java | 20 +++++------ .../es/rest/RestClientBulkItemStatusTest.java | 35 ++++++++++++++----- 4 files changed, 59 insertions(+), 26 deletions(-) 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..c2e586494d 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 @@ -861,8 +861,8 @@ public void mutate(Map> mutations, KeyInforma mutation.getDeletions(), true); Map doc = compat.prepareStoredScript(parameterizedDeletionScriptId, params).build(); log.trace("Deletion script {} with params {}", PARAMETERIZED_DELETION_SCRIPT, params); - requestByStore.add(ElasticSearchMutation.createUpdateRequest(indexStoreName, storeName, - documentId, doc)); + requestByStore.add(ElasticSearchMutation.createFieldDeletionRequest(indexStoreName, + storeName, documentId, doc)); } } if (mutation.hasAdditions()) { diff --git a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchMutation.java b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchMutation.java index cc0f05f6ef..75b6e899c3 100644 --- a/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchMutation.java +++ b/janusgraph-es/src/main/java/org/janusgraph/diskstorage/es/ElasticSearchMutation.java @@ -42,35 +42,51 @@ public enum RequestType { private final Map source; - private ElasticSearchMutation(RequestType requestType, String index, String type, String id, Map source) { + //A mutation which only takes content out of the index asks for nothing that an absent document does not already + //satisfy. Elasticsearch reports an absent document as a 404 for such a mutation and for one which puts content in, + //so the two can only be told apart from the mutation itself + private final boolean removesContentOnly; + + private ElasticSearchMutation(RequestType requestType, String index, String type, String id, Map source, + boolean removesContentOnly) { this.requestType = requestType; this.index = index; this.type = type; this.id = id; this.source = source; + this.removesContentOnly = removesContentOnly; } public static ElasticSearchMutation createDeleteRequest(String index, String type, String id) { - return new ElasticSearchMutation(RequestType.DELETE, index, type, id, null); + return new ElasticSearchMutation(RequestType.DELETE, index, type, id, null, true); } public static ElasticSearchMutation createIndexRequest(String index, String type, String id, Map source) { - return new ElasticSearchMutation(RequestType.INDEX, index, type, id, source); + return new ElasticSearchMutation(RequestType.INDEX, index, type, id, source, false); + } + + //An update which runs a script removing fields from a document, rather than the whole document + public static ElasticSearchMutation createFieldDeletionRequest(String index, String type, String id, Map source) { + return new ElasticSearchMutation(RequestType.UPDATE, index, type, id, source, true); } public static ElasticSearchMutation createUpdateRequest(String index, String type, String id, Map source) { - return new ElasticSearchMutation(RequestType.UPDATE, index, type, id, source); + return new ElasticSearchMutation(RequestType.UPDATE, index, type, id, source, false); } public static ElasticSearchMutation createUpdateRequest(String index, String type, String id, ImmutableMap.Builder builder, Map upsert) { final Map source = upsert == null ? builder.build() : builder.put(ES_UPSERT_KEY, upsert).build(); - return new ElasticSearchMutation(RequestType.UPDATE, index, type, id, source); + return new ElasticSearchMutation(RequestType.UPDATE, index, type, id, source, false); } public RequestType getRequestType() { return requestType; } + public boolean removesContentOnly() { + return removesContentOnly; + } + public String getIndex() { return index; } 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 6c480012df..dc8e152975 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 @@ -399,11 +399,11 @@ class RequestBytes { final byte [] requestBytes; final byte [] requestSource; //Retained so that a failed bulk item can be interpreted against the operation which produced it - final ElasticSearchMutation.RequestType requestType; + final boolean removesContentOnly; @VisibleForTesting RequestBytes(final ElasticSearchMutation request) throws JsonProcessingException { - this.requestType = request.getRequestType(); + this.removesContentOnly = request.removesContentOnly(); Map requestData = new HashMap<>(); if (useMappingTypes) { requestData.put("_index", request.getIndex()); @@ -478,20 +478,20 @@ private List> pairErrorsWithSubmittedMuta } RestBulkResponse.RestBulkItemResponse item = bulkResponseItem.iterator().next(); final RequestBytes submittedItem = submittedBulkRequestItems.get(itemIndex); - if (item.getError() != null && !isAbsentDocumentDeletion(item, submittedItem)) { + if (item.getError() != null && !isAbsentDocumentRemoval(item, submittedItem)) { errors.add(Triplet.with(item.getError(), item.getStatus(), submittedItem)); } } return errors; } - //Deleting a document which is already absent leaves the index in the state the deletion asked for, so its 404 is - //a success. An update which returns 404 is a document_missing_exception: the write did not happen, and treating it - //as a success drops the mutation with nothing reported - private static boolean isAbsentDocumentDeletion(final RestBulkResponse.RestBulkItemResponse item, - final RequestBytes submittedItem) { - return item.getStatus() == HttpStatus.SC_NOT_FOUND - && submittedItem.requestType == ElasticSearchMutation.RequestType.DELETE; + //Removing content from a document which is already absent leaves the index in the state the mutation asked for, so + //the 404 Elasticsearch answers with is a success. Deleting the whole document and running a script which deletes + //fields both count. A 404 for a mutation which adds content is a document_missing_exception: the write did not + //happen, and treating it as a success drops the mutation with nothing reported + private static boolean isAbsentDocumentRemoval(final RestBulkResponse.RestBulkItemResponse item, + final RequestBytes submittedItem) { + return item.getStatus() == HttpStatus.SC_NOT_FOUND && submittedItem.removesContentOnly; } @VisibleForTesting diff --git a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/rest/RestClientBulkItemStatusTest.java b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/rest/RestClientBulkItemStatusTest.java index 953844b39d..3cbcbc1923 100644 --- a/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/rest/RestClientBulkItemStatusTest.java +++ b/janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/rest/RestClientBulkItemStatusTest.java @@ -41,11 +41,12 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.when; -//A bulk response reports item level failures inside an otherwise successful HTTP response. Every 404 used to be -//treated as a success regardless of the operation which produced it. That is right for a delete, which is idempotent, -//but an update returning 404 is a document_missing_exception: the write did not happen. Because a property change on a -//SINGLE cardinality key produces a deletion and an addition against the same document, mutate() withholds the upsert, -//so both items 404 when the document is absent and the element is never indexed again. +//A bulk response reports item level failures inside an otherwise successful HTTP response. Every 404 used to be treated +//as a success regardless of the mutation which produced it. That is right for a mutation which only removes content, +//because an absent document already satisfies it, and both a whole document deletion and a script which deletes fields +//qualify. It is wrong for a mutation which adds content: there the 404 is a document_missing_exception and the write did +//not happen. A property change on a SINGLE cardinality key produces a field deletion and an addition against the same +//document, and mutate() withholds the upsert from the addition, so the addition is the item worth reporting. @ExtendWith(MockitoExtension.class) public class RestClientBulkItemStatusTest { @@ -91,11 +92,18 @@ private Response bulkResponseWith(List operations, List statuse return response; } + //An update which adds content. Without an upsert Elasticsearch answers 404 when the document is absent private static ElasticSearchMutation update(String id) { return ElasticSearchMutation.createUpdateRequest(INDEX, TYPE, id, ImmutableMap.builder().put("doc", ImmutableMap.of("name", "value")), null); } + //An update which runs the script mutate() uses to take fields out of a document + private static ElasticSearchMutation fieldDeletion(String id) { + return ElasticSearchMutation.createFieldDeletionRequest(INDEX, TYPE, id, + ImmutableMap.of("script", ImmutableMap.of("id", "deletion_script"))); + } + private void bulkRequest(List mutations, List operations, List statuses) throws IOException { try (RestElasticSearchClient clientUnderTest = createClient()) { @@ -113,6 +121,14 @@ public void shouldTreatTheDeletionOfAnAbsentDocumentAsSuccess() throws IOExcepti Collections.singletonList("delete"), Collections.singletonList(404)); } + @Test + public void shouldTreatTheFieldDeletionOfAnAbsentDocumentAsSuccess() throws IOException { + //Nothing is thrown: a document with no fields left to delete is the state the mutation asked for. Elasticsearch + //reports this as an update of a missing document, the same way it reports an addition which was lost + bulkRequest(Collections.singletonList(fieldDeletion("doc1")), + Collections.singletonList("update"), Collections.singletonList(404)); + } + @Test public void shouldReportAnUpdateOfAMissingDocument() throws IOException { final IOException e = assertThrows(IOException.class, @@ -131,13 +147,14 @@ public void shouldReportAnIndexRequestWhichReturnedNotFound() throws IOException } @Test - public void shouldReportAMissingDocumentUpdateSubmittedAlongsideADeletion() throws IOException { + public void shouldReportOnlyTheAdditionWhenAFieldDeletionAccompaniesIt() throws IOException { //This is the shape mutate() produces for a value change on a SINGLE cardinality key: the field deletion script - //and the addition script against the same absent document + //and the addition script against the same absent document. mutate() leaves the addition without an upsert once + //the mutation has deletions, so the addition is the half which lost a write and the only half worth reporting final IOException e = assertThrows(IOException.class, () -> bulkRequest( - Arrays.asList(update("doc1"), update("doc1")), + Arrays.asList(fieldDeletion("doc1"), update("doc1")), Arrays.asList("update", "update"), Arrays.asList(404, 404))); - assertTrue(e.getMessage().contains("document_missing_exception"), e.getMessage()); + assertEquals(1, e.getMessage().split("document_missing_exception", -1).length - 1, e.getMessage()); } @Test