Fix SigV4 signing failure on bulk retry after partial success

TrackingBytesArray.toInputStream() returned the entire underlying byte
array including bytes of removed entries, while writeTo() correctly
skipped them. This caused a mismatch between the x-amz-content-sha256
header (computed from toInputStream()) and the actual HTTP body (sent
via writeTo()) when retrying a bulk request after partial success.

Changed toInputStream() to build its result using the same entry
traversal logic as writeTo().

Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

Author: lawofcycles

mr/src/main/java/org/opensearch/hadoop/util/TrackingBytesArray.java

@@ -29,6 +29,7 @@
 package org.opensearch.hadoop.util;
 
 import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -161,9 +162,14 @@ public void writeTo(OutputStream out) throws IOException {
     public InputStream toInputStream() {
         if (size == 0) {
             return new ByteArrayInputStream(new byte[0]);
-        } else {
-            return data.toInputStream();
         }
+        ByteArrayOutputStream out = new ByteArrayOutputStream(size);
+        try {
+            writeTo(out);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+        return new ByteArrayInputStream(out.toByteArray());
     }
 
     public void reset() {

mr/src/test/java/org/opensearch/hadoop/util/TrackingBytesArrayTest.java

@@ -123,4 +123,38 @@ public void testPopData() throws Exception {
         assertEquals(7, data.length());
         assertEquals(2, entry.length());
     }
+
+    @Test
+    public void testToInputStreamIncludesRemovedEntriesBytesAfterRemove() throws Exception {
+        TrackingBytesArray tba = new TrackingBytesArray(new BytesArray(1024));
+
+        byte[] entry1 = "{\"index\":{\"_id\":\"1\"}}\n{\"title\":\"doc1\"}\n".getBytes();
+        byte[] entry2 = "{\"index\":{\"_id\":\"2\"}}\n{\"title\":\"doc2\"}\n".getBytes();
+        byte[] entry3 = "{\"index\":{\"_id\":\"3\"}}\n{\"title\":\"doc3\"}\n".getBytes();
+
+        tba.copyFrom(new BytesArray(entry1, entry1.length));
+        tba.copyFrom(new BytesArray(entry2, entry2.length));
+        tba.copyFrom(new BytesArray(entry3, entry3.length));
+
+        // Simulate partial bulk success: first entry succeeded, remove it before retry
+        tba.remove(0);
+
+        // writeTo: what gets sent over HTTP
+        ByteArrayOutputStream httpBody = new ByteArrayOutputStream();
+        tba.writeTo(httpBody);
+
+        // toInputStream: what SigV4 uses to compute x-amz-content-sha256
+        byte[] signedBody = tba.toInputStream().readAllBytes();
+
+        // These SHOULD be equal for SigV4 signature to match the actual HTTP body.
+        // If they differ, the signature computed from toInputStream() will not match
+        // the body sent via writeTo(), causing AOSS to reject with
+        // "x-amz-content-sha256 invalid"
+        assertArrayEquals(
+            "toInputStream() must return the same bytes as writeTo() after remove(). " +
+            "Mismatch causes SigV4 signature failure on AOSS bulk retry.",
+            httpBody.toByteArray(),
+            signedBody
+        );
+    }
 }