update ingestion status in index metadata xcontent and avoid retry for parsing error

Signed-off-by: Varun Bharadwaj <varunbharadwaj1995@gmail.com>

Author: varunbharadwaj

CHANGELOG.md

@@ -51,6 +51,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Fix QueryPhaseResultConsumer incomplete callback loops ([#19231](https://github.com/opensearch-project/OpenSearch/pull/19231))
 - Fix the `scaled_float` precision issue ([#19188](https://github.com/opensearch-project/OpenSearch/pull/19188))
 - Fix Using an excessively large reindex slice can lead to a JVM OutOfMemoryError on coordinator.([#18964](https://github.com/opensearch-project/OpenSearch/pull/18964))
+- Fix ingestion state xcontent serialization in IndexMetadata ([#19320](https://github.com/opensearch-project/OpenSearch/pull/19320))
 
 ### Dependencies
 - Bump `com.netflix.nebula.ospackage-base` from 12.0.0 to 12.1.0 ([#19019](https://github.com/opensearch-project/OpenSearch/pull/19019))

server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java

@@ -943,6 +943,7 @@ public Iterator<Setting<?>> settings() {
     public static final String TRANSLOG_METADATA_KEY = "translog_metadata";
     public static final String CONTEXT_KEY = "context";
     public static final String INGESTION_SOURCE_KEY = "ingestion_source";
+    public static final String INGESTION_STATUS_KEY = "ingestion_status";
 
     public static final String INDEX_STATE_FILE_PREFIX = "state-";
 
@@ -2264,6 +2265,11 @@ public static void toXContent(IndexMetadata indexMetadata, XContentBuilder build
                 indexMetadata.context.toXContent(builder, params);
             }
 
+            if (indexMetadata.ingestionStatus != null) {
+                builder.field(INGESTION_STATUS_KEY);
+                indexMetadata.ingestionStatus.toXContent(builder, params);
+            }
+
             builder.endObject();
         }
 
@@ -2347,6 +2353,8 @@ public static IndexMetadata fromXContent(XContentParser parser) throws IOExcepti
                         parser.skipChildren();
                     } else if (CONTEXT_KEY.equals(currentFieldName)) {
                         builder.context(Context.fromXContent(parser));
+                    } else if (INGESTION_STATUS_KEY.equals(currentFieldName)) {
+                        builder.ingestionStatus(IngestionStatus.fromXContent(parser));
                     } else {
                         // assume it's custom index metadata
                         builder.putCustom(currentFieldName, parser.mapStrings());

server/src/main/java/org/opensearch/cluster/metadata/IngestionStatus.java

@@ -12,14 +12,18 @@
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
 import org.opensearch.core.common.io.stream.Writeable;
+import org.opensearch.core.xcontent.ToXContent;
+import org.opensearch.core.xcontent.XContentBuilder;
+import org.opensearch.core.xcontent.XContentParser;
 
 import java.io.IOException;
 
 /**
  * Indicates pull-based ingestion status.
  */
 @ExperimentalApi
-public record IngestionStatus(boolean isPaused) implements Writeable {
+public record IngestionStatus(boolean isPaused) implements Writeable, ToXContent {
+    public static final String IS_PAUSED = "is_paused";
 
     public IngestionStatus(StreamInput in) throws IOException {
         this(in.readBoolean());
@@ -30,6 +34,37 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeBoolean(isPaused);
     }
 
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        builder.field(IS_PAUSED, isPaused);
+        builder.endObject();
+        return builder;
+    }
+
+    public static IngestionStatus fromXContent(XContentParser parser) throws IOException {
+        boolean isPaused = false;
+
+        XContentParser.Token token = parser.currentToken();
+        if (token == null) {
+            token = parser.nextToken();
+        }
+
+        if (token == XContentParser.Token.START_OBJECT) {
+            while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
+                if (token == XContentParser.Token.FIELD_NAME) {
+                    String fieldName = parser.currentName();
+                    if (IS_PAUSED.equals(fieldName)) {
+                        parser.nextToken();
+                        isPaused = parser.booleanValue();
+                    }
+                }
+            }
+        }
+
+        return new IngestionStatus(isPaused);
+    }
+
     public static IngestionStatus getDefaultValue() {
         return new IngestionStatus(false);
     }

server/src/main/java/org/opensearch/indices/pollingingest/MessageProcessorRunnable.java

@@ -26,6 +26,7 @@
 import org.opensearch.index.engine.IngestionEngine;
 import org.opensearch.index.engine.VersionConflictEngineException;
 import org.opensearch.index.mapper.IdFieldMapper;
+import org.opensearch.index.mapper.MapperParsingException;
 import org.opensearch.index.mapper.ParseContext;
 import org.opensearch.index.mapper.ParsedDocument;
 import org.opensearch.index.mapper.SourceToParse;
@@ -311,7 +312,7 @@ public void run() {
                 } catch (Exception e) {
                     messageProcessorMetrics.failedMessageCounter.inc();
                     errorStrategy.handleError(e, IngestionErrorStrategy.ErrorStage.PROCESSING);
-                    boolean retriesExhausted = retryCount >= MIN_RETRY_COUNT || e instanceof IllegalArgumentException;
+                    boolean retriesExhausted = hasExhaustedRetries(e, retryCount);
                     if (retriesExhausted && errorStrategy.shouldIgnoreError(e, IngestionErrorStrategy.ErrorStage.PROCESSING)) {
                         logDroppedMessage(shardUpdateMessage);
                         shardUpdateMessage = null;
@@ -336,6 +337,15 @@ private void waitBeforeRetry() {
         }
     }
 
+    private boolean hasExhaustedRetries(Exception e, int retryCount) {
+        if (retryCount >= MIN_RETRY_COUNT) {
+            return true;
+        }
+
+        // Don't retry validation/parsing errors
+        return e instanceof IllegalArgumentException || e instanceof MapperParsingException;
+    }
+
     public MessageProcessorMetrics getMessageProcessorMetrics() {
         return messageProcessorMetrics;
     }