diff --git a/CHANGELOG.md b/CHANGELOG.md index 4f96378bb..e1a1f00bc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) - Fixed serverless mode SaveMode.Overwrite failing when document count exceeds scroll size ([#693](https://github.com/opensearch-project/opensearch-hadoop/pull/693)) - Fixed RowSerializationEventConverterTest for Spark 3.4+ StructType.toString() format change ([#702](https://github.com/opensearch-project/opensearch-hadoop/pull/702)) - Fixed object fields with `enabled: false` returning empty structs or throwing exceptions when read via the connector ([#715](https://github.com/opensearch-project/opensearch-hadoop/pull/715)) +- Fixed SigV4 signing failure on bulk retry after partial success causing `x-amz-content-sha256 invalid` ([#759](https://github.com/opensearch-project/opensearch-hadoop/pull/759)) ### Security diff --git a/mr/src/main/java/org/opensearch/hadoop/util/TrackingBytesArray.java b/mr/src/main/java/org/opensearch/hadoop/util/TrackingBytesArray.java index e7de55ad2..dad42bc49 100644 --- a/mr/src/main/java/org/opensearch/hadoop/util/TrackingBytesArray.java +++ b/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() { diff --git a/mr/src/test/java/org/opensearch/hadoop/util/TrackingBytesArrayTest.java b/mr/src/test/java/org/opensearch/hadoop/util/TrackingBytesArrayTest.java index e34830979..ec28c26f7 100644 --- a/mr/src/test/java/org/opensearch/hadoop/util/TrackingBytesArrayTest.java +++ b/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 + ); + } } \ No newline at end of file