Fix file source infinite re-read in non-tail mode with codec (#6934) (#6937)

* Stop file source from rescheduling readers in non-tail mode (#6934)

FileReaderPool.onReaderComplete inferred the "should I reschedule?"
decision from the completed reader's RotationType, which has no
terminal value. In non-tail mode any path that did not result in
DELETED or CREATE_RENAME (notably NO_ROTATION and the codec one-shot
path that never updates lastRotationType) was rescheduled every
500 ms, producing duplicate events.

Make tail mode the single source of truth for rescheduling: when the
reader completes in non-tail mode, mark the checkpoint completed,
promote pending files, and exit. This restores the documented
"non-tail = read once, stop" contract for the modern path and matches
the behavior of the legacy ClassicFileStrategy.

Resolves #6934

Signed-off-by: Srikanth Padakanti <srikanth_padakanti@apple.com>

* Persist read offset after non-tail codec one-shot read (#6934)

readFileWithCodecOneShot returned without updating the checkpoint
entry, so a successful one-shot read advanced no offset. After the
pool-side fix (no reschedule in non-tail mode), an in-process loop
no longer occurs, but a restart would still re-read the file from
offset 0 and produce duplicate events.

After parseWithCodec returns true, advance readOffset, the
checkpoint's read offset, and the committed offset to the file
size. On parse failure the readErrors counter is incremented and
the offsets stay at zero, matching the pre-fix semantics for the
error path.

Resolves #6934

Signed-off-by: Srikanth Padakanti <srikanth_padakanti@apple.com>

---------

Signed-off-by: Srikanth Padakanti <srikanth_padakanti@apple.com>

Author: srikanthpadakanti

data-prepper-plugins/file-source/src/main/java/org/opensearch/dataprepper/plugins/source/file/FileReader.java

@@ -197,7 +197,12 @@ private void readFileWithCodecOneShot() {
         try (final InputStream rawStream = Files.newInputStream(path);
              final InputStream decompressedStream = decompressionEngine.createInputStream(rawStream)) {
             metrics.getFilesOpened().increment();
-            parseWithCodec(decompressedStream);
+            if (parseWithCodec(decompressedStream)) {
+                final long fileSize = fileOps.size(path);
+                readOffset.set(fileSize);
+                checkpointEntry.setReadOffset(fileSize);
+                checkpointEntry.setCommittedOffset(fileSize);
+            }
         } catch (final IOException e) {
             LOG.error("Error reading file with codec: {}", path, e);
             metrics.getReadErrors().increment();

data-prepper-plugins/file-source/src/main/java/org/opensearch/dataprepper/plugins/source/file/FileReaderPool.java

@@ -119,6 +119,12 @@ private synchronized void onReaderComplete(final FileIdentity fileIdentity, fina
         }
         metrics.getActiveFileCount().decrementAndGet();
 
+        if (!readerContext.isTailMode()) {
+            checkpointRegistry.markCompleted(fileIdentity.toString());
+            promotePendingFiles();
+            return;
+        }
+
         if (completedReader.getLastRotationType() == RotationType.CREATE_RENAME) {
             LOG.info("Re-adding path {} after create/rename rotation", path);
             final FileIdentity newIdentity = FileIdentity.from(path, readerContext.getFileOps(),

data-prepper-plugins/file-source/src/test/java/org/opensearch/dataprepper/plugins/source/file/FileReaderPoolTest.java

@@ -49,6 +49,8 @@
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 @ExtendWith(MockitoExtension.class)
@@ -620,4 +622,69 @@ void onReaderComplete_with_deleted_rotation_marks_completed_and_processes_pendin
 
         limitedPool.shutdown();
     }
+
+    private FileReaderContext createNonTailReaderContext() {
+        return new FileReaderContext(
+                buffer, eventFactory, fileOps, metrics, rotationDetector,
+                acknowledgementSetManager, false, StandardCharsets.UTF_8,
+                4096, 1048576, 5000, Duration.ofSeconds(5),
+                Duration.ofSeconds(30), StartPosition.BEGINNING, false,
+                Duration.ofSeconds(30), 1000,
+                Duration.ofSeconds(5), 3, null, false, in -> in);
+    }
+
+    @Test
+    void onReaderComplete_in_non_tail_mode_does_not_reschedule() throws Exception {
+        Path testFile = tempDir.resolve("non-tail.log");
+        Files.writeString(testFile, "line1\nline2\nline3\n");
+
+        Counter filesOpened = mock(Counter.class);
+        Counter filesClosed = mock(Counter.class);
+        Counter bytesRead = mock(Counter.class);
+        Counter linesRead = mock(Counter.class);
+        Counter eventsEmitted = mock(Counter.class);
+        Timer backpressureTimer = mock(Timer.class);
+        lenient().when(metrics.getFilesOpened()).thenReturn(filesOpened);
+        lenient().when(metrics.getFilesClosed()).thenReturn(filesClosed);
+        lenient().when(metrics.getBytesRead()).thenReturn(bytesRead);
+        lenient().when(metrics.getLinesRead()).thenReturn(linesRead);
+        lenient().when(metrics.getEventsEmitted()).thenReturn(eventsEmitted);
+        lenient().when(metrics.getBackpressureTimer()).thenReturn(backpressureTimer);
+        lenient().when(metrics.getFileLagBytes()).thenReturn(new AtomicLong(0));
+        when(metrics.getActiveFileCount()).thenReturn(new AtomicLong(0));
+
+        BasicFileAttributes attrs = mock(BasicFileAttributes.class);
+        when(attrs.fileKey()).thenReturn("inode-non-tail");
+        when(attrs.creationTime()).thenReturn(FileTime.from(Instant.EPOCH));
+        when(fileOps.readAttributes(testFile)).thenReturn(attrs);
+        when(fileOps.size(testFile)).thenReturn(Files.size(testFile));
+        FileChannel realChannel = FileChannel.open(testFile, StandardOpenOption.READ);
+        when(fileOps.openReadChannel(testFile)).thenReturn(realChannel);
+
+        when(rotationDetector.checkRotation(any(), any(), any(long.class)))
+                .thenReturn(RotationResult.NO_ROTATION);
+
+        EventBuilder mockBuilder = mock(EventBuilder.class);
+        Event mockEvent = mock(Event.class);
+        lenient().when(eventFactory.eventBuilder(EventBuilder.class)).thenReturn(mockBuilder);
+        lenient().when(mockBuilder.withEventType(any())).thenReturn(mockBuilder);
+        lenient().when(mockBuilder.withData(any(Map.class))).thenReturn(mockBuilder);
+        lenient().when(mockBuilder.build()).thenReturn(mockEvent);
+
+        FileReaderPool pool = new FileReaderPool(
+                checkpointRegistry, metrics, 10, 2,
+                Duration.ofMinutes(30), createNonTailReaderContext());
+        when(checkpointRegistry.getOrCreate(anyString())).thenReturn(new CheckpointEntry());
+
+        FileIdentity identity = FileIdentity.from(testFile, fileOps, 1024);
+        pool.addFile(identity, testFile);
+
+        verify(checkpointRegistry, timeout(5000)).markCompleted(anyString());
+
+        await().pollDelay(1500, TimeUnit.MILLISECONDS)
+                .atMost(3, TimeUnit.SECONDS)
+                .until(() -> pool.getActiveReaderCount() == 0);
+
+        pool.shutdown();
+    }
 }

data-prepper-plugins/file-source/src/test/java/org/opensearch/dataprepper/plugins/source/file/FileReaderTest.java

@@ -911,6 +911,62 @@ void run_one_shot_codec_with_acknowledgements_creates_ack_set() throws Exception
         verify(ackSet, atLeastOnce()).add(any(Event.class));
     }
 
+    @Test
+    void run_one_shot_codec_persists_read_offset_after_successful_parse() throws Exception {
+        Path testFile = tempDir.resolve("one-shot-offset.log");
+        final String contents = "line1\nline2\nline3\n";
+        Files.writeString(testFile, contents);
+        final long fileSize = Files.size(testFile);
+
+        Counter filesOpened = mock(Counter.class);
+        Counter filesClosed = mock(Counter.class);
+        Counter linesRead = mock(Counter.class);
+        Counter eventsEmitted = mock(Counter.class);
+        Timer backpressureTimer = mock(Timer.class);
+        when(metrics.getFilesOpened()).thenReturn(filesOpened);
+        when(metrics.getFilesClosed()).thenReturn(filesClosed);
+        lenient().when(metrics.getLinesRead()).thenReturn(linesRead);
+        lenient().when(metrics.getEventsEmitted()).thenReturn(eventsEmitted);
+        lenient().when(metrics.getBackpressureTimer()).thenReturn(backpressureTimer);
+
+        when(fileOps.size(testFile)).thenReturn(fileSize);
+
+        InputCodec mockCodec = mock(InputCodec.class);
+        doAnswer(inv -> null).when(mockCodec).parse(any(), any());
+
+        FileReaderContext context = createOneShotContextWithCodecAndAcknowledgements(mockCodec);
+        final FileReader reader = createReaderWithContext(testFile, context);
+        reader.run();
+
+        assertThat(checkpointEntry.getReadOffset(), equalTo(fileSize));
+        assertThat(checkpointEntry.getCommittedOffset(), equalTo(fileSize));
+        assertThat(reader.getReadOffset(), equalTo(fileSize));
+    }
+
+    @Test
+    void run_one_shot_codec_does_not_persist_offset_when_parse_throws() throws Exception {
+        Path testFile = tempDir.resolve("one-shot-offset-fail.log");
+        Files.writeString(testFile, "line1\n");
+
+        Counter filesOpened = mock(Counter.class);
+        Counter filesClosed = mock(Counter.class);
+        Counter readErrors = mock(Counter.class);
+        when(metrics.getFilesOpened()).thenReturn(filesOpened);
+        when(metrics.getFilesClosed()).thenReturn(filesClosed);
+        when(metrics.getReadErrors()).thenReturn(readErrors);
+
+        InputCodec mockCodec = mock(InputCodec.class);
+        doThrow(new IOException("parse failed")).when(mockCodec).parse(any(), any());
+
+        FileReaderContext context = createOneShotContextWithCodecAndAcknowledgements(mockCodec);
+        final FileReader reader = createReaderWithContext(testFile, context);
+        reader.run();
+
+        assertThat(checkpointEntry.getReadOffset(), equalTo(0L));
+        assertThat(checkpointEntry.getCommittedOffset(), equalTo(0L));
+        verify(readErrors).increment();
+    }
+
     @Test
     void run_with_acknowledgements_batch_timeout_triggers_complete() throws Exception {
         Path testFile = tempDir.resolve("ack-timeout.log");