I think that embedding the on failure handling here is good. I've been thinking about how to make this class also usable on the pipeline level:

What I think we can change here is to let HandledProcessor also handle multiple processors, so that then this class can also be used at the pipeline level.

Something like this:

public class HandledProcessor implements Processor {

    // No need to use HandledProcessor as generic type, since that is an implementation detail
    private final List<Processor> processors; 
    private final List<Processor> onFailureProcessors;

    public HandledProcessor(Processor... processor) {
        this(Arrays.asList(processor), Collections.emptyList());
    }
    public HandledProcessor(List<Processor> processors, List<Processor> onFailureProcessors) {
        this.processors = Collections.unmodifiableList(processors);
        this.onFailureProcessors = Collections.unmodifiableList(onFailureProcessors);
    }

    public boolean isHandled() {
        return !onFailureProcessors.isEmpty();
    }

    public List<Processor> getProcessors() {
        return processors;
    }

    @Override
    public void execute(IngestDocument ingestDocument) throws Exception {
        try {
            for (Processor processor : processors) {
                processor.execute(ingestDocument);
            }
        } catch (Exception e) {
            if (isHandled()) {
                executeOnFailure(ingestDocument);
            } else {
                throw e;
            }
        }

    }

    public void executeOnFailure(IngestDocument ingestDocument) throws Exception {
        for (Processor processor : onFailureProcessors) {
            processor.execute(ingestDocument);
        }
    }
}

Then in Pipeline the processors field type can be changed to HandledProcessor.
The the pipeline factory can then benefit from having the same parsing logic for the on_failure on both processor and pipeline level.

However the Processor#getType() method is problematic with this approach, because it doesn't assume a processor is embedding multiple processors. This method is only really used in the simulate api and I think for that reason this method needs to be removed from the Processor interface and the functionality of figuring out what is the processor type should go somewhere else maybe the PipelineStore? We can do this in a followup pr and for now just throw an UnsupportedOperationException in HandledProcessor#getType(). The Pipeline#getProcessors() method then just needs to delegate to HandledProcessor#getProcessors so that the SimulateExecutionService doesn't fail.

If this makes sense, then we should rename HandledProcessor to CompoundProcessor.

For now, what do you say about having CompoundProcessor's getType return: "compound[processor1,processor2]". so it wraps all the underlying processor types within its type name.

I think that is okay for now. I do like to see Processor#getType() be replaced that just works for the simulate api.

maybe add a private helper method for parsing on failure processors that both readProcessor(...) and create(...) use?

yeah. good idea. I added a readProcessors helper. definitely improves code-reuse.

million dollar question: shall we catch Exception or Throwable here?

I suppose Throwable would catch more things, as we would allow processors to run arbitrary code in their execute. I will change it to that

per @martijnvg's note. I will leave this as Exception

Maybe catch Exception instead. If there is a jvm error (out of memory and such) then we should just bail and not try to execute the onfailure processors.