Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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() {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
);
}
}
Loading