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 f399863065..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 @@ -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 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 && !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> { //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..3cbcbc1923 --- /dev/null +++ b/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 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; + } + + //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()) { + //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()); + } +}