Fixed the silent-batch-drop bug where OpenSearch bulk indexing fails during shard reallocation

changelog/unreleased/pr-27031.toml

@@ -0,0 +1,5 @@
+type = "fixed"
+message = "Fixed bulk indexing requests being dropped instead of retried when OpenSearch returns an incomplete response during shard reallocation."
+
+issues = ["26853"]
+pulls = ["27031"]

graylog-storage-opensearch3/src/main/java/org/graylog/storage/opensearch3/MessagesAdapterOS.java

@@ -19,7 +19,9 @@
 import com.codahale.metrics.Meter;
 import com.codahale.metrics.MetricRegistry;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Throwables;
 import jakarta.inject.Inject;
+import org.graylog2.indexer.IncompleteBulkResponseException;
 import org.graylog2.indexer.messages.ChunkedBulkIndexer;
 import org.graylog2.indexer.messages.DocumentNotFoundException;
 import org.graylog2.indexer.messages.Indexable;
@@ -36,6 +38,7 @@
 import org.opensearch.client.opensearch._types.ErrorCause;
 import org.opensearch.client.opensearch._types.ErrorResponse;
 import org.opensearch.client.opensearch._types.OpenSearchException;
+import org.opensearch.client.opensearch._types.ShardSearchFailure;
 import org.opensearch.client.opensearch.core.BulkRequest;
 import org.opensearch.client.opensearch.core.BulkResponse;
 import org.opensearch.client.opensearch.core.GetRequest;
@@ -46,6 +49,7 @@
 import org.opensearch.client.opensearch.indices.AnalyzeResponse;
 import org.opensearch.client.opensearch.indices.analyze.AnalyzeToken;
 import org.opensearch.client.transport.httpclient5.ResponseException;
+import org.opensearch.client.util.MissingRequiredPropertyException;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -144,10 +148,35 @@ private ChunkedBulkIndexer.BulkIndexResult runBulkRequest(int indexedSuccessfull
             throw new org.graylog2.indexer.ElasticsearchException(e);
         } catch (IOException e) {
             throw new org.graylog2.indexer.ElasticsearchException(e);
+        } catch (RuntimeException e) {
+            if (isIncompleteShardFailureResponse(e)) {
+                // Can happen while shards are being reallocated during a rolling restart: OpenSearch omits the
+                // "shard" field on a shard failure entry, which the opensearch-java 3.x client fails to parse.
+                // Messages#createBulkRequestRetryerBuilder() retries this exception type explicitly.
+                throw new IncompleteBulkResponseException(
+                        "Received an incomplete bulk response from OpenSearch, this can happen while shards are being reallocated",
+                        e);
+            }
+            throw e;
         }
         return new ChunkedBulkIndexer.BulkIndexResult(indexingResultsFrom(result, chunk), () -> buildFailureMessage(result), result.items().size());
     }
 
+    // The opensearch-java client throws this whenever ANY response it deserializes is missing a field it
+    // considers required, so we only treat it as our known, retryable rolling-restart scenario when the
+    // missing property is specifically ShardSearchFailure#shard. Anything else (a different missing field,
+    // or a genuinely incompatible/broken response) should keep propagating instead of being retried forever.
+    // Note: this is the same "_shards.failures[]" schema field that was named ShardFailure#shard in
+    // opensearch-java <= 3.3.0; the client renamed the generated class to ShardSearchFailure afterwards. If
+    // opensearch.client.version is bumped and this stops matching, MessagesAdapterOSTest will fail loudly.
+    private boolean isIncompleteShardFailureResponse(RuntimeException e) {
+        return Throwables.getCausalChain(e).stream()
+                .filter(MissingRequiredPropertyException.class::isInstance)
+                .map(MissingRequiredPropertyException.class::cast)
+                .anyMatch(missingProperty -> ShardSearchFailure.class.equals(missingProperty.getObjectClass())
+                        && "shard".equals(missingProperty.getPropertyName()));
+    }
+
     private OpenSearchException toOpenSearchException(ResponseException re) {
         String[] split = re.getMessage().split("\n");
         if (split.length != 2) {

graylog-storage-opensearch3/src/test/java/org/graylog/storage/opensearch3/MessagesAdapterOSTest.java

@@ -0,0 +1,144 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog.storage.opensearch3;
+
+import com.codahale.metrics.MetricRegistry;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.graylog.storage.opensearch3.testing.client.mock.ServerlessOpenSearchClient;
+import org.graylog2.indexer.IncompleteBulkResponseException;
+import org.graylog2.indexer.messages.ChunkedBulkIndexer;
+import org.graylog2.indexer.messages.Indexable;
+import org.graylog2.indexer.messages.IndexingRequest;
+import org.graylog2.indexer.results.ResultMessageFactory;
+import org.graylog2.shared.utilities.ExceptionUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.opensearch.client.util.MissingRequiredPropertyException;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class MessagesAdapterOSTest {
+
+    @Mock
+    private ResultMessageFactory resultMessageFactory;
+
+    /**
+     * Reproduces a rolling restart where a replica shard is still being promoted: OpenSearch reports a shard-level
+     * write failure whose {@code _shards.failures[]} entry omits the "shard" number. The opensearch-java client
+     * treats that as a required property and refuses to parse the response at all.
+     */
+    private static final String BULK_RESPONSE_WITH_INCOMPLETE_SHARD_FAILURE = """
+            {
+              "took": 1,
+              "errors": true,
+              "items": [
+                {
+                  "index": {
+                    "_index": "graylog_0",
+                    "_id": "message-id",
+                    "status": 200,
+                    "_shards": {
+                      "total": 2,
+                      "successful": 1,
+                      "failed": 1,
+                      "failures": [
+                        {
+                          "index": "graylog_0",
+                          "node": "node-1",
+                          "reason": {
+                            "type": "unavailable_shards_exception",
+                            "reason": "primary shard is not active"
+                          }
+                        }
+                      ]
+                    }
+                  }
+                }
+              ]
+            }
+            """;
+
+    /**
+     * A missing "status" is a required-property failure too, but it has nothing to do with shard reallocation.
+     * It must not be mistaken for the scenario above and silently retried forever.
+     */
+    private static final String BULK_RESPONSE_MISSING_UNRELATED_REQUIRED_PROPERTY = """
+            {
+              "took": 1,
+              "errors": true,
+              "items": [
+                {
+                  "index": {
+                    "_index": "graylog_0",
+                    "_id": "message-id"
+                  }
+                }
+              ]
+            }
+            """;
+
+    @Test
+    void bulkIndexTreatsIncompleteShardFailureResponseAsRetryable() {
+        final OfficialOpensearchClient officialOpensearchClient = ServerlessOpenSearchClient.builder()
+                .stubResponse("POST", "/_bulk", BULK_RESPONSE_WITH_INCOMPLETE_SHARD_FAILURE)
+                .build();
+
+        final MessagesAdapterOS messagesAdapterOS = new MessagesAdapterOS(resultMessageFactory, officialOpensearchClient,
+                new MetricRegistry(), new ChunkedBulkIndexer(), new ObjectMapper());
+
+        final Indexable message = mock(Indexable.class);
+        when(message.getId()).thenReturn("message-id");
+        when(message.toElasticSearchObject(any(), any())).thenReturn(Map.of("message", "test"));
+
+        final List<IndexingRequest> request = List.of(IndexingRequest.create("graylog_0", message));
+
+        assertThatThrownBy(() -> messagesAdapterOS.bulkIndex(request))
+                .isInstanceOf(IncompleteBulkResponseException.class)
+                .satisfies(e -> assertThat(ExceptionUtils.hasCauseOf(e, MissingRequiredPropertyException.class)).isTrue());
+    }
+
+    @Test
+    void bulkIndexDoesNotTreatUnrelatedMissingPropertyAsRetryable() {
+        final OfficialOpensearchClient officialOpensearchClient = ServerlessOpenSearchClient.builder()
+                .stubResponse("POST", "/_bulk", BULK_RESPONSE_MISSING_UNRELATED_REQUIRED_PROPERTY)
+                .build();
+
+        final MessagesAdapterOS messagesAdapterOS = new MessagesAdapterOS(resultMessageFactory, officialOpensearchClient,
+                new MetricRegistry(), new ChunkedBulkIndexer(), new ObjectMapper());
+
+        final Indexable message = mock(Indexable.class);
+        when(message.getId()).thenReturn("message-id");
+        when(message.toElasticSearchObject(any(), any())).thenReturn(Map.of("message", "test"));
+
+        final List<IndexingRequest> request = List.of(IndexingRequest.create("graylog_0", message));
+
+        // Must propagate as the original, unwrapped exception rather than being turned into a retryable one.
+        assertThatThrownBy(() -> messagesAdapterOS.bulkIndex(request))
+                .isInstanceOf(MissingRequiredPropertyException.class)
+                .hasMessageContaining("status");
+    }
+}

graylog2-server/src/main/java/org/graylog2/indexer/IncompleteBulkResponseException.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog2.indexer;
+
+/**
+ * Thrown by a {@code MessagesAdapter} when an indexer's bulk response cannot be parsed because it omitted
+ * shard-failure detail while a shard was being reallocated, e.g. during a rolling restart. Retried like other
+ * transient indexing failures, see {@code Messages#createBulkRequestRetryerBuilder()}.
+ * <p>
+ * See <a href="https://github.com/opensearch-project/opensearch-java/issues/551">opensearch-java#551</a>: the
+ * client treats the shard number on a shard-failure entry as a required field, but OpenSearch can omit it while
+ * the failing shard is still being promoted. A client-side fix is proposed in
+ * <a href="https://github.com/opensearch-project/opensearch-java/pull/2023">opensearch-java#2023</a> (unmerged
+ * as of this writing); once a released client version includes it, this workaround can likely be removed.
+ */
+public class IncompleteBulkResponseException extends ElasticsearchException {
+    public IncompleteBulkResponseException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

graylog2-server/src/main/java/org/graylog2/indexer/messages/Messages.java

@@ -26,6 +26,7 @@
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import org.graylog.failure.FailureSubmissionService;
+import org.graylog2.indexer.IncompleteBulkResponseException;
 import org.graylog2.indexer.InvalidWriteTargetException;
 import org.graylog2.indexer.MasterNotDiscoveredException;
 import org.graylog2.indexer.results.ResultMessage;
@@ -67,7 +68,8 @@ private RetryerBuilder<IndexingResults> createBulkRequestRetryerBuilder() {
         return RetryerBuilder.<IndexingResults>newBuilder()
                 .retryIfException(t -> ExceptionUtils.hasCauseOf(t, IOException.class)
                         || t instanceof InvalidWriteTargetException
-                        || t instanceof MasterNotDiscoveredException)
+                        || t instanceof MasterNotDiscoveredException
+                        || t instanceof IncompleteBulkResponseException)
                 .withWaitStrategy(WaitStrategies.exponentialWait(MAX_WAIT_TIME.getQuantity(), MAX_WAIT_TIME.getUnit()))
                 .withRetryListener(new RetryListener() {
                     @Override

graylog2-server/src/test/java/org/graylog2/indexer/messages/MessagesTest.java

@@ -18,6 +18,8 @@
 
 import com.google.common.collect.ImmutableList;
 import org.graylog.failure.FailureSubmissionService;
+import org.graylog2.indexer.ElasticsearchException;
+import org.graylog2.indexer.IncompleteBulkResponseException;
 import org.graylog2.plugin.Message;
 import org.graylog2.plugin.MessageFactory;
 import org.graylog2.plugin.TestMessageFactory;
@@ -35,6 +37,7 @@
 import org.mockito.quality.Strictness;
 
 import java.io.IOException;
+import java.net.SocketTimeoutException;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
@@ -250,6 +253,58 @@ public void bulkIndexRequests_nothingPropagatedToFailureSubmissionServiceWhenThe
         verifyNoInteractions(failureSubmissionService);
     }
 
+    @Test
+    public void bulkIndexRequests_retriesWhenAdapterThrowsExceptionWithIOExceptionCause() throws Exception {
+        // given
+        final DateTime ts = Tools.nowUTC();
+        final Message message1 = message("msg-1", ts);
+
+        final List<IndexingRequest> indexingRequest = ImmutableList.of(IndexingRequest.create("", message1));
+
+        // A generic transient transport failure (e.g. a real connection problem), as any MessagesAdapter might
+        // throw it, unrelated to the more specific IncompleteBulkResponseException case tested below.
+        final ElasticsearchException transientFailure = new ElasticsearchException(
+                "Could not reach the indexer", new SocketTimeoutException("connect timed out"));
+
+        when(messagesAdapter.bulkIndex(indexingRequest))
+                .thenThrow(transientFailure)
+                .thenReturn(IndexingResults.create(List.of(new IndexingSuccess(message1, "index_1")), List.of()));
+
+        // when
+        final IndexingResults indexingResults = messages.bulkIndexRequests(indexingRequest, false);
+
+        // then
+        verify(messagesAdapter, times(2)).bulkIndex(indexingRequest);
+        assertThat(indexingResults.errors()).isEmpty();
+        assertThat(indexingResults.successes()).hasSize(1);
+    }
+
+    @Test
+    public void bulkIndexRequests_retriesWhenAdapterThrowsIncompleteBulkResponseException() throws Exception {
+        // given
+        final DateTime ts = Tools.nowUTC();
+        final Message message1 = message("msg-1", ts);
+
+        final List<IndexingRequest> indexingRequest = ImmutableList.of(IndexingRequest.create("", message1));
+
+        // What MessagesAdapterOS throws when OpenSearch returns an incomplete bulk response (e.g. a
+        // "_shards.failures[]" entry missing its "shard" field) while shards are being reallocated.
+        final IncompleteBulkResponseException transientFailure = new IncompleteBulkResponseException(
+                "Received an incomplete bulk response from OpenSearch", new RuntimeException("missing property"));
+
+        when(messagesAdapter.bulkIndex(indexingRequest))
+                .thenThrow(transientFailure)
+                .thenReturn(IndexingResults.create(List.of(new IndexingSuccess(message1, "index_1")), List.of()));
+
+        // when
+        final IndexingResults indexingResults = messages.bulkIndexRequests(indexingRequest, false);
+
+        // then
+        verify(messagesAdapter, times(2)).bulkIndex(indexingRequest);
+        assertThat(indexingResults.errors()).isEmpty();
+        assertThat(indexingResults.successes()).hasSize(1);
+    }
+
     private List<IndexingSuccess> createSuccessFromMessages(List<MessageWithIndex> messageList) {
         return messageList.stream().map(m -> new IndexingSuccess(m.message(), "index_2")).collect(Collectors.toList());
     }