ingest: expose reroute inquiry/reset via Elastic-internal API bridge

Logstash's Integration filter works directly with the processors, but cannot
use the IngestService that is tightly-coupled with cluster state and must
therefore emulate the behavior introduced in elastic#94000.

To do so, the additional methods for inquiring about and resetting the reroute
state need to be externally-accessible. Exposing them through a clearly-named
bridge allows us to avoid making these Elastic-internal bits a part of the
public APIs that are subject to years-long stability and deprecation notice
policies.

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

@@ -0,0 +1,39 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.ingest;
+
+/**
+ * This bridge class exposes package-private components of Ingest in a way
+ * that can be consumed by Logstash's Elastic Integration Filter without
+ * expanding our externally-consumable API.
+ *
+ * @apiNote this is an Elastic-internal API bridge intended for exclusive use by
+ *          Logstash and its Elastic Integration Filter.
+ */
+public class LogstashInternalBridge {
+
+    private LogstashInternalBridge() {}
+
+    /**
+     * The document has been redirected to another target.
+     * This implies that the default pipeline of the new target needs to be invoked.
+     *
+     * @return whether the document is redirected to another target
+     */
+    public static boolean isReroute(final IngestDocument ingestDocument) {
+        return ingestDocument.isReroute();
+    }
+
+    /**
+     * Set the reroute flag of the provided {@link IngestDocument} to {@code false}.
+     */
+    public static void resetReroute(final IngestDocument ingestDocument) {
+        ingestDocument.resetReroute();
+    }
+}

server/src/test/java/org/elasticsearch/ingest/LogstashInternalBridgeTests.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.ingest;
+
+import org.elasticsearch.test.ESTestCase;
+
+import static org.elasticsearch.ingest.TestIngestDocument.emptyIngestDocument;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+
+public class LogstashInternalBridgeTests extends ESTestCase {
+    public void testIngestDocumentRerouteBridge() {
+        final IngestDocument ingestDocument = emptyIngestDocument();
+        ingestDocument.setFieldValue("_index", "nowhere");
+        assertThat(ingestDocument.getFieldValue("_index", String.class), is(equalTo("nowhere")));
+        assertThat(LogstashInternalBridge.isReroute(ingestDocument), is(false));
+
+        ingestDocument.reroute("somewhere");
+        assertThat(ingestDocument.getFieldValue("_index", String.class), is(equalTo("somewhere")));
+        assertThat(LogstashInternalBridge.isReroute(ingestDocument), is(true));
+
+        LogstashInternalBridge.resetReroute(ingestDocument);
+        assertThat(ingestDocument.getFieldValue("_index", String.class), is(equalTo("somewhere")));
+        assertThat(LogstashInternalBridge.isReroute(ingestDocument), is(false));
+    }
+}