[Backport 2.x] implement mark() in class FilterStreamInput

CHANGELOG.md

@@ -39,6 +39,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Fixed
 - Fix bulk API ignores ingest pipeline for upsert ([#12883](https://github.com/opensearch-project/OpenSearch/pull/12883))
 - Fix issue with feature flags where default value may not be honored ([#12849](https://github.com/opensearch-project/OpenSearch/pull/12849))
+- Fix implement mark() and markSupported() in class FilterStreamInput ([#13098](https://github.com/opensearch-project/OpenSearch/pull/13098))
 
 ### Security
 

libs/core/src/main/java/org/opensearch/core/common/io/stream/FilterStreamInput.java

@@ -80,6 +80,16 @@ public void reset() throws IOException {
         delegate.reset();
     }
 
+    @Override
+    public void mark(int readlimit) {
+        delegate.mark(readlimit);
+    }
+
+    @Override
+    public boolean markSupported() {
+        return delegate.markSupported();
+    }
+
     @Override
     public int read() throws IOException {
         return delegate.read();

libs/core/src/test/java/org/opensearch/core/common/io/stream/FilterStreamInputTests.java

@@ -12,6 +12,9 @@
 import org.opensearch.core.common.bytes.BytesReference;
 
 import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import static org.hamcrest.Matchers.is;
 
 /** test the FilterStreamInput using the same BaseStreamTests */
 public class FilterStreamInputTests extends BaseStreamTests {
@@ -21,4 +24,24 @@ protected StreamInput getStreamInput(BytesReference bytesReference) throws IOExc
         return new FilterStreamInput(StreamInput.wrap(br.bytes, br.offset, br.length)) {
         };
     }
+
+    public void testMarkAndReset() throws IOException {
+        FilterStreamInputTests filterStreamInputTests = new FilterStreamInputTests();
+
+        ByteBuffer buffer = ByteBuffer.wrap(new byte[20]);
+        for (int i = 0; i < buffer.limit(); i++) {
+            buffer.put((byte) i);
+        }
+        buffer.rewind();
+        BytesReference bytesReference = BytesReference.fromByteBuffer(buffer);
+        StreamInput streamInput = filterStreamInputTests.getStreamInput(bytesReference);
+        streamInput.read();
+        assertThat(streamInput.markSupported(), is(true));
+        streamInput.mark(-1);
+        int int1 = streamInput.read();
+        int int2 = streamInput.read();
+        streamInput.reset();
+        assertEquals(int1, streamInput.read());
+        assertEquals(int2, streamInput.read());
+    }
 }