disable push-API for indexing in ingestionEngine

Signed-off-by: Yupeng Fu <[email protected]>

Author: yupeng9

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

@@ -35,6 +35,7 @@
 import org.opensearch.indices.pollingingest.StreamPoller;
 
 import java.io.IOException;
+import java.io.UnsupportedEncodingException;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -145,6 +146,16 @@ protected Set<IngestionShardPointer> fetchPersistedOffsets(DirectoryReader direc
 
     @Override
     public IndexResult index(Index index) throws IOException {
+        throw new UnsupportedEncodingException("push-based indexing is not supported in ingestion engine, use streaming source instead");
+    }
+
+    /**
+     * Indexes the document into the engine. This is used internally by the stream poller only.
+     * @param index the index request
+     * @return the index result
+     * @throws IOException if an error occurs
+     */
+    public IndexResult indexInternal(Index index) throws IOException {
         assert Objects.equals(index.uid().field(), IdFieldMapper.NAME) : index.uid().field();
         ensureOpen();
         final IndexResult indexResult;
@@ -168,7 +179,7 @@ private void addDocs(final List<ParseContext.Document> docs, final IndexWriter i
 
     @Override
     public DeleteResult delete(Delete delete) throws IOException {
-        return null;
+        throw new UnsupportedEncodingException("push-based deletion is not supported in ingestion engine, use streaming source instead");
     }
 
     @Override

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

@@ -119,7 +119,7 @@ protected void process(Message message, IngestionShardPointer pointer) {
                 Engine.Operation operation = getOperation(payload, pointer);
                 switch (operation.operationType()) {
                     case INDEX:
-                        engine.index((Engine.Index) operation);
+                        engine.indexInternal((Engine.Index) operation);
                         break;
                     case DELETE:
                         engine.delete((Engine.Delete) operation);

server/src/test/java/org/opensearch/index/engine/IngestionEngineTests.java

@@ -35,6 +35,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.mockito.Mockito;
+
 import static org.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doThrow;
@@ -128,6 +130,22 @@ public void testRecovery() throws IOException {
         waitForResults(ingestionEngine, 4);
     }
 
+    public void testPushAPIFailures() throws IOException {
+        try {
+            ingestionEngine.index(Mockito.any());
+            fail("Expected UnsupportedOperationException to be thrown");
+        } catch (Exception e) {
+            assertEquals("push-based indexing is not supported in ingestion engine, use streaming source instead", e.getMessage());
+        }
+
+        try {
+            ingestionEngine.delete(Mockito.any());
+            fail("Expected UnsupportedOperationException to be thrown");
+        } catch (Exception e) {
+            assertEquals("push-based deletion is not supported in ingestion engine, use streaming source instead", e.getMessage());
+        }
+    }
+
     public void testCreationFailure() throws IOException {
         final AtomicLong globalCheckpoint = new AtomicLong(SequenceNumbers.NO_OPS_PERFORMED);
         FakeIngestionSource.FakeIngestionConsumerFactory consumerFactory = new FakeIngestionSource.FakeIngestionConsumerFactory(messages);