Report a bulk update which failed because the document is missing (#4926)

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

@@ -861,8 +861,8 @@ public void mutate(Map<String, Map<String, IndexMutation>> 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()) {

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;
     }

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 boolean removesContentOnly;
 
         @VisibleForTesting
         RequestBytes(final ElasticSearchMutation request) throws JsonProcessingException {
+            this.removesContentOnly = request.removesContentOnly();
             Map<String, Object> requestData = new HashMap<>();
             if (useMappingTypes) {
                 requestData.put("_index", request.getIndex());
@@ -474,13 +477,23 @@ private List<Triplet<Object, Integer, RequestBytes>> 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 && !isAbsentDocumentRemoval(item, submittedItem)) {
+                errors.add(Triplet.with(item.getError(), item.getStatus(), submittedItem));
             }
         }
         return errors;
     }
 
+    //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
     class BulkRequestChunker implements Iterator<List<RequestBytes>> {
         //By default, Elasticsearch writes are limited to 100mb, so chunk a given batch of requests so they stay under

janusgraph-es/src/test/java/org/janusgraph/diskstorage/es/rest/RestClientBulkItemStatusTest.java

@@ -0,0 +1,178 @@
+// 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 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 {
+
+    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<String> operations, List<Integer> statuses) throws IOException {
+        final RestBulkResponse bulkResponse = new RestBulkResponse();
+        final List<Map<String, RestBulkResponse.RestBulkItemResponse>> 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;
+    }
+
+    //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<ElasticSearchMutation> mutations, List<String> operations, List<Integer> 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 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,
+            () -> 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 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. 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(fieldDeletion("doc1"), update("doc1")),
+            Arrays.asList("update", "update"), Arrays.asList(404, 404)));
+        assertEquals(1, e.getMessage().split("document_missing_exception", -1).length - 1, 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());
+    }
+}