Make error strategy config type-safe

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

Author: varunbharadwaj

CHANGELOG-3.0.md

@@ -22,6 +22,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Arrow Flight RPC plugin with Flight server bootstrap logic and client for internode communication ([#16962](https://github.com/opensearch-project/OpenSearch/pull/16962))
 - Added offset management for the pull-based Ingestion ([#17354](https://github.com/opensearch-project/OpenSearch/pull/17354))
 - Add filter function for AbstractQueryBuilder, BoolQueryBuilder, ConstantScoreQueryBuilder([#17409](https://github.com/opensearch-project/OpenSearch/pull/17409))
+- Added error handling support for the pull-based ingestion ([#17427](https://github.com/opensearch-project/OpenSearch/pull/17427))
 
 ### Dependencies
 - Update Apache Lucene to 10.1.0 ([#16366](https://github.com/opensearch-project/OpenSearch/pull/16366))

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

@@ -771,28 +771,13 @@ public Iterator<Setting<?>> settings() {
         Property.Final
     );
 
-    public static final String SETTING_INGESTION_SOURCE_ERROR_STRATEGY = "index.ingestion_source.error.strategy";
-    public static final Setting<String> INGESTION_SOURCE_ERROR_STRATEGY_SETTING = Setting.simpleString(
+    public static final String SETTING_INGESTION_SOURCE_ERROR_STRATEGY = "index.ingestion_source.error_strategy";
+    public static final Setting<IngestionErrorStrategy.ErrorStrategy> INGESTION_SOURCE_ERROR_STRATEGY_SETTING = new Setting<>(
         SETTING_INGESTION_SOURCE_ERROR_STRATEGY,
         IngestionErrorStrategy.ErrorStrategy.DROP.name(),
-        new Setting.Validator<>() {
-
-            @Override
-            public void validate(final String value) {
-                try {
-                    IngestionErrorStrategy.ErrorStrategy.valueOf(value.toUpperCase(Locale.ROOT));
-                } catch (IllegalArgumentException e) {
-                    throw new IllegalArgumentException("Invalid value for " + SETTING_INGESTION_SOURCE_ERROR_STRATEGY + " [" + value + "]");
-                }
-            }
-
-            @Override
-            public void validate(final String value, final Map<Setting<?>, Object> settings) {
-                validate(value);
-            }
-        },
-        Property.IndexScope,
-        Property.Dynamic
+        IngestionErrorStrategy.ErrorStrategy::parseFromString,
+        (errorStrategy) -> {},
+        Property.IndexScope
     );
 
     public static final Setting.AffixSetting<Object> INGESTION_SOURCE_PARAMS_SETTING = Setting.prefixKeySetting(
@@ -1030,10 +1015,7 @@ public IngestionSource getIngestionSource() {
                 pointerInitResetValue
             );
 
-            final String errorStrategyString = INGESTION_SOURCE_ERROR_STRATEGY_SETTING.get(settings);
-            IngestionErrorStrategy.ErrorStrategy errorStrategy = IngestionErrorStrategy.ErrorStrategy.valueOf(
-                errorStrategyString.toUpperCase(Locale.ROOT)
-            );
+            final IngestionErrorStrategy.ErrorStrategy errorStrategy = INGESTION_SOURCE_ERROR_STRATEGY_SETTING.get(settings);
             final Map<String, Object> ingestionSourceParams = INGESTION_SOURCE_PARAMS_SETTING.getAsMap(settings);
             return new IngestionSource(ingestionSourceType, pointerInitReset, errorStrategy, ingestionSourceParams);
         }

server/src/main/java/org/opensearch/index/engine/IngestionEngine.java

@@ -29,8 +29,8 @@
 import org.opensearch.index.translog.TranslogStats;
 import org.opensearch.index.translog.listener.CompositeTranslogEventListener;
 import org.opensearch.indices.pollingingest.DefaultStreamPoller;
-import org.opensearch.indices.pollingingest.PollingIngestStats;
 import org.opensearch.indices.pollingingest.IngestionErrorStrategy;
+import org.opensearch.indices.pollingingest.PollingIngestStats;
 import org.opensearch.indices.pollingingest.StreamPoller;
 
 import java.io.IOException;

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

@@ -10,6 +10,8 @@
 
 import org.opensearch.common.annotation.ExperimentalApi;
 
+import java.util.Locale;
+
 /**
  * Defines the error handling strategy when an error is encountered either during polling records from ingestion source
  * or during processing the polled records.
@@ -43,7 +45,15 @@ static IngestionErrorStrategy create(ErrorStrategy errorStrategy, String ingesti
     @ExperimentalApi
     enum ErrorStrategy {
         DROP,
-        BLOCK
+        BLOCK;
+
+        public static ErrorStrategy parseFromString(String errorStrategy) {
+            try {
+                return ErrorStrategy.valueOf(errorStrategy.toUpperCase(Locale.ROOT));
+            } catch (IllegalArgumentException e) {
+                throw new IllegalArgumentException("Invalid ingestion errorStrategy: " + errorStrategy, e);
+            }
+        }
     }
 
     /**