Handle any exception thrown while generating source for an IngestDocument (elastic#91981)

Author: joegallo

docs/changelog/91981.yaml

@@ -0,0 +1,5 @@
+pr: 91981
+summary: Handle any exception thrown while generating source for an `IngestDocument`
+area: Ingest Node
+type: bug
+issues: []

server/src/main/java/org/elasticsearch/ingest/IngestService.java

@@ -947,10 +947,8 @@ private void innerExecute(
                         boolean ensureNoSelfReferences = ingestDocument.doNoSelfReferencesCheck();
                         indexRequest.source(ingestDocument.getSource(), indexRequest.getContentType(), ensureNoSelfReferences);
                     } catch (IllegalArgumentException ex) {
-                        // An IllegalArgumentException can be thrown when an ingest
-                        // processor creates a source map that is self-referencing.
-                        // In that case, we catch and wrap the exception so we can
-                        // include which pipeline failed.
+                        // An IllegalArgumentException can be thrown when an ingest processor creates a source map that is self-referencing.
+                        // In that case, we catch and wrap the exception, so we can include which pipeline failed.
                         totalMetrics.ingestFailed();
                         handler.accept(
                             new IllegalArgumentException(
@@ -959,6 +957,19 @@ private void innerExecute(
                             )
                         );
                         return;
+                    } catch (Exception ex) {
+                        // If anything goes wrong here, we want to know, and cannot proceed with normal execution. For example,
+                        // *rarely*, a ConcurrentModificationException could be thrown if a pipeline leaks a reference to a shared mutable
+                        // collection, and another indexing thread modifies the shared reference while we're trying to ensure it has
+                        // no self references.
+                        totalMetrics.ingestFailed();
+                        handler.accept(
+                            new RuntimeException(
+                                "Failed to generate the source document for ingest pipeline [" + pipeline.getId() + "]",
+                                ex
+                            )
+                        );
+                        return;
                     }
                     Map<String, String> map;
                     if ((map = metadata.getDynamicTemplates()) != null) {