Add tests and disable multi-thread writes till feature is complete

Signed-off-by: Varun Bharadwaj <[email protected]>

Author: varunbharadwaj

plugins/ingestion-kafka/src/internalClusterTest/java/org/opensearch/plugin/kafka/IngestFromKafkaIT.java

@@ -138,30 +138,11 @@ public void testCloseIndex() throws Exception {
         client().admin().indices().close(Requests.closeIndexRequest(indexName)).get();
     }
 
-    public void testKafkaIngestionWithMultipleProcessorThreads() throws Exception {
-        for (int i = 1; i <= 2000; i++) {
-            produceData(String.valueOf(i), "name" + i, "25");
-        }
-        createIndexWithDefaultSettings(indexName, 1, 0, 3);
-        RangeQueryBuilder query = new RangeQueryBuilder("age").gte(21);
-        waitForState(() -> {
-            refresh(indexName);
-            SearchResponse response = client().prepareSearch(indexName).setQuery(query).get();
-            assertThat(response.getHits().getTotalHits().value(), is(2000L));
-            PollingIngestStats stats = client().admin().indices().prepareStats(indexName).get().getIndex(indexName).getShards()[0]
-                .getPollingIngestStats();
-            assertNotNull(stats);
-            assertThat(stats.getMessageProcessorStats().getTotalProcessedCount(), is(2000L));
-            assertThat(stats.getConsumerStats().getTotalPolledCount(), is(2000L));
-            return true;
-        });
-    }
-
     public void testUpdateAndDelete() throws Exception {
         // Step 1: Produce message and wait for it to be searchable
 
         produceData("1", "name", "25", defaultMessageTimestamp, "index");
-        createIndexWithDefaultSettings(indexName, 1, 0, 3);
+        createIndexWithDefaultSettings(1, 0);
         ensureGreen(indexName);
         waitForState(() -> {
             BoolQueryBuilder query = new BoolQueryBuilder().must(new TermQueryBuilder("_id", "1"));

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

@@ -787,13 +787,15 @@ public Iterator<Setting<?>> settings() {
     );
 
     /**
-     * Defines the number of processor threads that will write to the lucene index.
+     * Defines the number of processor threads that will write to the lucene index. This setting is currently disabled
+     * and will be allowed once the feature is ready. A default value of 1 will be used.
      */
     public static final String SETTING_INGESTION_SOURCE_NUM_PROCESSOR_THREADS = "index.ingestion_source.num_processor_threads";
     public static final Setting<Integer> INGESTION_SOURCE_NUM_PROCESSOR_THREADS_SETTING = Setting.intSetting(
         SETTING_INGESTION_SOURCE_NUM_PROCESSOR_THREADS,
         1,
         1,
+        1,
         Setting.Property.IndexScope,
         Setting.Property.Final
     );

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

@@ -232,7 +232,7 @@ protected void startPoll() {
                 includeBatchStartPointer = false;
             } catch (Throwable e) {
                 encounteredError = true;
-                logger.error("Error in polling the shard {}: {}", consumer.getShardId(), e);
+                logger.error("Error in polling the shard {}, lastProcessedPointer {}: {}", consumer.getShardId(), lastProcessedPointer, e);
                 errorStrategy.handleError(e, IngestionErrorStrategy.ErrorStage.POLLING);
 
                 if (!errorStrategy.shouldIgnoreError(e, IngestionErrorStrategy.ErrorStage.POLLING)) {

server/src/test/java/org/opensearch/indices/pollingingest/DefaultStreamPollerTests.java

@@ -372,6 +372,37 @@ public void testProcessingErrorWithBlockErrorIngestionStrategy() throws TimeoutE
         assertEquals(DefaultStreamPoller.State.POLLING, poller.getState());
     }
 
+    public void testInitialConsumerReadTransientError() throws TimeoutException, InterruptedException {
+        messages.add("{\"_id\":\"3\",\"_source\":{\"name\":\"bob\", \"age\": 24}}".getBytes(StandardCharsets.UTF_8));
+        messages.add("{\"_id\":\"4\",\"_source\":{\"name\":\"alice\", \"age\": 21}}".getBytes(StandardCharsets.UTF_8));
+
+        DropIngestionErrorStrategy mockErrorStrategy = spy(new DropIngestionErrorStrategy("ingestion_source"));
+        processorRunnable = new MessageProcessorRunnable(new ArrayBlockingQueue<>(5), processor, mockErrorStrategy);
+        PartitionedBlockingQueueContainer blockingQueueContainer = new PartitionedBlockingQueueContainer(processorRunnable, 0);
+        FakeIngestionSource.FakeIngestionConsumer consumerSpy = spy(fakeConsumer);
+
+        // fail consumer's first poll attempt
+        doThrow(new RuntimeException("failed to poll messages")).doCallRealMethod()
+            .when(consumerSpy)
+            .readNext(any(), anyBoolean(), anyLong(), anyInt());
+
+        poller = new DefaultStreamPoller(
+            new FakeIngestionSource.FakeIngestionShardPointer(0),
+            persistedPointers,
+            consumerSpy,
+            blockingQueueContainer,
+            StreamPoller.ResetState.NONE,
+            "",
+            mockErrorStrategy,
+            StreamPoller.State.NONE
+        );
+        poller.start();
+        Thread.sleep(sleepTime);
+
+        verify(mockErrorStrategy, times(1)).handleError(any(), eq(IngestionErrorStrategy.ErrorStage.POLLING));
+        assertEquals(4, blockingQueueContainer.getTotalProcessedCount());
+    }
+
     public void testUpdateErrorStrategy() {
         assertTrue(poller.getErrorStrategy() instanceof DropIngestionErrorStrategy);
         assertTrue(processorRunnable.getErrorStrategy() instanceof DropIngestionErrorStrategy);