diff --git a/.github/workflows/build-main.yml b/.github/workflows/build-main.yml
index cd627b2..b1eb9f1 100644
--- a/.github/workflows/build-main.yml
+++ b/.github/workflows/build-main.yml
@@ -4,8 +4,13 @@ on:
push:
branches:
- master
+ - development
tags:
- "*-[0-9]+.*"
+ pull_request:
+ branches:
+ - master
+ - development
jobs:
build:
@@ -35,4 +40,4 @@ jobs:
MAVEN_PASS: ${{ secrets.MAVEN_PASS }}
OSSRH_PASS: ${{ secrets.OSSRH_PASS }}
SIGNING_ASC: ${{ secrets.SIGNING_ASC }}
- AWS_REGION: us-east-1
+ AWS_REGION: us-east-1
\ No newline at end of file
diff --git a/.github/workflows/build-pr.yml b/.github/workflows/build-pr.yml
deleted file mode 100644
index 45882a4..0000000
--- a/.github/workflows/build-pr.yml
+++ /dev/null
@@ -1,30 +0,0 @@
-name: build PR
-
-on:
- pull_request:
- branches:
- - master
-
-jobs:
- build:
- runs-on: ubuntu-latest
-
- steps:
- - uses: actions/checkout@v4
- - name: Start MinIO Server
- uses: comfuture/minio-action@v1
- with:
- access_key: minioadmin
- secret_key: minioadmin
- - name: Set up Java
- uses: actions/setup-java@v5
- with:
- java-version: '8'
- distribution: 'zulu'
- cache: 'maven'
- - name: Set up CI environment
- run: .github/setup.sh
- - name: Execute the build
- run: .github/build.sh
- env:
- AWS_REGION: us-east-1
diff --git a/pom.xml b/pom.xml
index 1f6befc..2cf8cd2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -11,7 +11,7 @@
org.janelia.saalfeldlab
n5-aws-s3
- 4.4.0-alpha-6-SNAPSHOT
+ 4.4.0-alpha-7-SNAPSHOT
N5 AWS S3
N5 library implementation using Amazon Web Services S3 backend.
@@ -117,7 +117,7 @@
sign,deploy-to-scijava
- 4.0.0-alpha-9
+ 4.0.0-alpha-10
2.2.2
diff --git a/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java b/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java
index a14abf7..b97c9d9 100644
--- a/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java
+++ b/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java
@@ -25,61 +25,33 @@
*/
package org.janelia.saalfeldlab.n5.s3;
-import java.io.ByteArrayOutputStream;
-import java.io.Closeable;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
-import java.io.OutputStream;
-import java.io.OutputStreamWriter;
-import java.io.Reader;
-import java.io.Writer;
+import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;
-import java.nio.channels.NonReadableChannelException;
-import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Supplier;
-import java.util.stream.Collectors;
import org.janelia.saalfeldlab.n5.KeyValueAccess;
-import org.janelia.saalfeldlab.n5.LockedChannel;
import org.janelia.saalfeldlab.n5.N5Exception;
import org.janelia.saalfeldlab.n5.N5Exception.N5IOException;
import org.janelia.saalfeldlab.n5.N5Exception.N5NoSuchKeyException;
import org.janelia.saalfeldlab.n5.N5URI;
-import org.janelia.saalfeldlab.n5.readdata.LazyRead;
import org.janelia.saalfeldlab.n5.readdata.ReadData;
import org.janelia.saalfeldlab.n5.readdata.VolatileReadData;
import software.amazon.awssdk.awscore.exception.AwsServiceException;
-import software.amazon.awssdk.core.ResponseBytes;
import software.amazon.awssdk.core.sync.RequestBody;
-import software.amazon.awssdk.core.sync.ResponseTransformer;
import software.amazon.awssdk.services.s3.S3Client;
-import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
-import software.amazon.awssdk.services.s3.model.Delete;
-import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
-import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest;
-import software.amazon.awssdk.services.s3.model.GetObjectRequest;
-import software.amazon.awssdk.services.s3.model.GetObjectResponse;
-import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
-import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
-import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
-import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
-import software.amazon.awssdk.services.s3.model.NoSuchBucketException;
-import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
-import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
-import software.amazon.awssdk.services.s3.model.PutObjectRequest;
-import software.amazon.awssdk.services.s3.model.S3Exception;
+import software.amazon.awssdk.services.s3.model.*;
public class AmazonS3KeyValueAccess implements KeyValueAccess {
private final S3Client s3;
private final URI containerURI;
private final String bucketName;
+ private S3IoPolicy ioPolicy;
private final boolean createBucket;
private Boolean bucketCheckedAndExists = null;
@@ -123,6 +95,8 @@ public AmazonS3KeyValueAccess(final S3Client s3, final URI containerURI, final b
this.bucketName = AmazonS3Utils.getS3Bucket(containerURI);
this.createBucket = createBucket;
+ this.ioPolicy = setIoPolicy();
+
if (!bucketExists()) {
if (createBucket) {
s3.createBucket(CreateBucketRequest.builder().bucket(bucketName).build());
@@ -134,6 +108,26 @@ public AmazonS3KeyValueAccess(final S3Client s3, final URI containerURI, final b
}
}
+ public void setIoPolicy(S3IoPolicy ioPolicy) {
+ this.ioPolicy = ioPolicy;
+ }
+
+ private S3IoPolicy setIoPolicy() {
+
+ String ioPolicy = System.getProperty("n5.ioPolicy");
+ if (ioPolicy == null)
+ return new S3IoPolicy.EtagMatch(s3, bucketName);
+
+ switch (ioPolicy) {
+ case "unsafe":
+ return new S3IoPolicy.Unsafe(s3, bucketName);
+ case "permissive": // For S3, this is equivalent ot just strict
+ case "strict":
+ default:
+ return new S3IoPolicy.EtagMatch(s3, bucketName);
+ }
+ }
+
private boolean bucketExists() {
return bucketCheckedAndExists = bucketCheckedAndExists != null
@@ -276,13 +270,10 @@ private boolean keyExists(final String key) {
try {
// TODO needs testing.
- // the exception thrown may depend on permissions
- @SuppressWarnings("unused")
- HeadObjectResponse headObjectResponse = s3.headObject(
- HeadObjectRequest.builder().key(key).bucket(bucketName).build());
-
+ // the exception thrown may depend on permissions
+ headObjectRequest(s3, bucketName, key, null);
return true;
- } catch( NoSuchKeyException e ) {
+ } catch( N5NoSuchKeyException | NoSuchKeyException e ) {
return false;
} catch (Throwable e) {
throw new N5Exception(e);
@@ -312,7 +303,7 @@ private boolean prefixExists(final String prefix) {
* @param path the path
* @return the path with a trailing slash
*/
- private static String addTrailingSlash(final String path) {
+ static String addTrailingSlash(final String path) {
return path.endsWith("/") ? path : path + "/";
}
@@ -324,7 +315,7 @@ private static String addTrailingSlash(final String path) {
* @param path the path
* @return the path without the leading slash
*/
- private static String removeLeadingSlash(final String path) {
+ static String removeLeadingSlash(final String path) {
return path.startsWith("/") ? path.substring(1) : path;
}
@@ -371,108 +362,31 @@ public boolean isFile(final String normalPath) {
public long size(String normalPath) throws N5NoSuchKeyException {
final String key = removeLeadingSlash(AmazonS3Utils.getS3Key(normalPath));
-
- final HeadObjectRequest headObjectRequest = HeadObjectRequest.builder()
- .bucket(bucketName)
- .key(key)
- .build();
-
- final HeadObjectResponse response = rethrowS3Exceptions(() -> s3.headObject(headObjectRequest));
- return response.contentLength();
+ return headObjectRequest(s3, bucketName, key, null).contentLength();
}
@Override
public VolatileReadData createReadData(String normalPath) throws N5Exception.N5IOException {
- // TODO locking
- return VolatileReadData.from(new S3LazyRead(normalPath));
- }
-
- @Override
- public void write(final String normalPath, final ReadData data) throws N5IOException {
-
- // TODO locking
final String key = AmazonS3Utils.getS3Key(normalPath);
- try (final S3ObjectChannel ch = new S3ObjectChannel(removeLeadingSlash(key), false);
- final OutputStream os = ch.newOutputStream();) {
- data.writeTo(os);
- } catch (IOException e) {
- throw new N5Exception.N5IOException(e);
- }
-
- }
-
- private T rethrowS3Exceptions(Supplier action) {
- try {
- return action.get();
- } catch (final NoSuchKeyException e) {
- throw new N5Exception.N5NoSuchKeyException("No such key", e);
- } catch (final AwsServiceException e) {
- final int statusCode = e.awsErrorDetails().sdkHttpResponse().statusCode();
- if (statusCode == 404 || statusCode == 403)
- throw new N5Exception.N5NoSuchKeyException("No such key", e);
- throw new N5Exception.N5IOException("S3 Exception", e);
- }
- }
-
- private class S3LazyRead implements LazyRead {
-
- private final String key;
-
- S3LazyRead(final String normalizedKey) {
- this.key = normalizedKey;
- }
-
- private GetObjectRequest createObjectRequest(final String s3Key, long offset, long length) {
-
- final GetObjectRequest.Builder requestBuilder = GetObjectRequest.builder()
- .key(s3Key)
- .bucket(bucketName);
-
- // Only add range header if we're doing a partial read
- if (offset > 0 || length > 0) {
- // HTTP Range header format: "bytes=start-end"
- // If length is 0 or negative, read from offset to end of file
- final String range = length > 0
- ? String.format("bytes=%d-%d", offset, offset + length - 1)
- : String.format("bytes=%d-", offset);
- requestBuilder.range(range);
- }
-
- return requestBuilder.build();
- }
-
- @Override public ReadData materialize(long offset, long length) throws N5Exception.N5IOException {
-
- final String s3Key = AmazonS3Utils.getS3Key(key);
- final GetObjectRequest request = createObjectRequest(s3Key, offset, length);
- final ResponseBytes response = rethrowS3Exceptions(() -> s3.getObject(request, ResponseTransformer.toBytes()));
- return ReadData.from(response.asByteArray());
- }
-
- @Override public long size() throws N5Exception.N5IOException {
-
- return AmazonS3KeyValueAccess.this.size(key);
- }
-
- @Override
- public void close() throws IOException {
- // TODO ioPolicy?
- }
+ try {
+ return ioPolicy.read(key);
+ } catch (IOException e) {
+ throw new N5IOException(e);
+ }
}
@Override
- public LockedChannel lockForReading(final String normalPath) {
+ public void write(final String normalPath, final ReadData data) throws N5IOException {
final String key = AmazonS3Utils.getS3Key(normalPath);
- return new S3ObjectChannel(key, true);
- }
+ final String normalizedKey = removeLeadingSlash(key);
- @Override
- public LockedChannel lockForWriting(final String normalPath) {
-
- final String key = AmazonS3Utils.getS3Key(normalPath);
- return new S3ObjectChannel(removeLeadingSlash(key), false);
+ try {
+ ioPolicy.write(normalizedKey, data);
+ } catch (IOException e) {
+ throw new N5IOException(e);
+ }
}
@Override
@@ -559,163 +473,38 @@ public void delete(final String normalPath) {
}
final String key = removeLeadingSlash(AmazonS3Utils.getS3Key(normalPath));
- if (!key.endsWith("/")) {
-
- final DeleteObjectRequest deleteRequest = DeleteObjectRequest.builder()
- .bucket(bucketName)
- .key(key)
- .build();
-
- try {
- s3.deleteObject(deleteRequest);
- } catch (S3Exception e) {
- }
- }
-
- final String prefix = addTrailingSlash(key);
+ try {
+ ioPolicy.delete(key);
+ } catch (IOException e) {
+ throw new N5IOException(e);
+ }
+ }
- ListObjectsV2Request listObjectsRequest;
- ListObjectsV2Response objectsListing;
- listObjectsRequest = ListObjectsV2Request.builder()
+ public static HeadObjectResponse headObjectRequest(final S3Client s3, final String bucketName, final String key, final String matchEtag) {
+ HeadObjectRequest.Builder requestBuilder = HeadObjectRequest.builder()
.bucket(bucketName)
- .prefix(prefix)
- .build();
+ .key(key);
- do {
- objectsListing = s3.listObjectsV2(listObjectsRequest);
- final List objectsToDelete = objectsListing.contents().stream().map( x -> {
- return ObjectIdentifier.builder().key(x.key()).build();
- }).collect(Collectors.toList());
+ if (matchEtag != null)
+ requestBuilder.ifMatch(matchEtag);
- if (!objectsToDelete.isEmpty()) {
- final DeleteObjectsRequest deleteRequest = DeleteObjectsRequest.builder()
- .bucket(bucketName)
- .delete( Delete.builder().objects(objectsToDelete).build())
- .build();
+ final HeadObjectRequest request = requestBuilder.build();
- s3.deleteObjects(deleteRequest);
- }
-
- // TODO what about continuation token?
-
- } while (objectsListing.isTruncated());
+ return rethrowS3Exceptions(() -> s3.headObject(request));
}
- private class S3ObjectChannel implements LockedChannel {
-
- protected final String path;
- final boolean readOnly;
- private final ArrayList resources = new ArrayList<>();
-
- protected S3ObjectChannel(final String path, final boolean readOnly) {
-
- this.path = path;
- this.readOnly = readOnly;
- }
-
- private void checkWritable() {
-
- if (readOnly) {
- throw new NonReadableChannelException();
- }
- }
-
- @Override
- public InputStream newInputStream() {
-
- try {
- final GetObjectRequest objectRequest = GetObjectRequest.builder().key(path).bucket(bucketName).build();
- // TODO consider using ResponseTransformer.toBytes
- return s3.getObject(objectRequest, ResponseTransformer.toInputStream());
- } catch (final S3Exception e) {
- // TODO figure out how to determine if the error is because the key does not exist
- if (e.statusCode() == 404 || e.statusCode() == 403)
- throw new N5Exception.N5NoSuchKeyException("No such key", e);
- else
- throw new N5Exception.N5IOException(e);
- }
- }
-
- @Override
- public Reader newReader() {
-
- final InputStreamReader reader = new InputStreamReader(newInputStream(), StandardCharsets.UTF_8);
- synchronized (resources) {
- resources.add(reader);
- }
- return reader;
- }
-
- @Override
- public OutputStream newOutputStream() {
-
- checkWritable();
- final S3OutputStream s3Out = new S3OutputStream();
- synchronized (resources) {
- resources.add(s3Out);
- }
- return s3Out;
- }
-
- @Override
- public Writer newWriter() {
-
- checkWritable();
- final OutputStreamWriter writer = new OutputStreamWriter(newOutputStream(), StandardCharsets.UTF_8);
- synchronized (resources) {
- resources.add(writer);
- }
- return writer;
- }
-
- @Override
- public void close() throws IOException {
-
- synchronized (resources) {
- for (final Closeable resource : resources)
- resource.close();
- resources.clear();
- }
- }
-
- final class S3OutputStream extends OutputStream {
-
- private final ByteArrayOutputStream buf = new ByteArrayOutputStream();
-
- private boolean closed = false;
-
- @Override
- public void write(final byte[] b, final int off, final int len) {
-
- buf.write(b, off, len);
- }
-
- @Override
- public void write(final int b) {
-
- buf.write(b);
- }
-
- @Override
- public synchronized void close() throws IOException {
-
- if (!closed) {
- closed = true;
- PutObjectRequest putOb = PutObjectRequest.builder()
- .bucket(bucketName)
- .key(path)
- .build();
-
- try {
- s3.putObject(putOb, RequestBody.fromBytes(buf.toByteArray()));
- } catch (S3Exception e) {
- e.printStackTrace();
- }
- buf.close();
- }
- }
+ public static T rethrowS3Exceptions(Supplier action) {
+ try {
+ return action.get();
+ } catch (final NoSuchKeyException e) {
+ throw new N5Exception.N5NoSuchKeyException("No such key", e);
+ } catch (final AwsServiceException e) {
+ final int statusCode = e.awsErrorDetails().sdkHttpResponse().statusCode();
+ if (statusCode == 404 || statusCode == 403)
+ throw new N5Exception.N5NoSuchKeyException("No such key", e);
+ if (statusCode == 412)
+ throw new N5Exception.N5ConcurrentModificationException("eTag has changed since initial request", e);
+ throw new N5Exception.N5IOException("S3 Exception", e);
}
}
-
-
}
diff --git a/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3Utils.java b/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3Utils.java
index cf393e3..16004b9 100644
--- a/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3Utils.java
+++ b/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3Utils.java
@@ -16,12 +16,7 @@
import software.amazon.awssdk.services.s3.S3ClientBuilder;
import software.amazon.awssdk.services.s3.S3Uri;
import software.amazon.awssdk.services.s3.S3Utilities;
-import software.amazon.awssdk.services.s3.model.DeleteBucketRequest;
-import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
-import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
-import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
-import software.amazon.awssdk.services.s3.model.S3Exception;
-import software.amazon.awssdk.services.s3.model.S3Object;
+import software.amazon.awssdk.services.s3.model.*;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
import software.amazon.awssdk.auth.credentials.AwsCredentials;
@@ -266,8 +261,9 @@ public static S3Client createS3(final String uri, @Nullable final Consumer objectsToDelete = objectsListing.contents().stream().map(x -> {
+ return ObjectIdentifier.builder().key(x.key()).build();
+ }).collect(Collectors.toList());
+
+ if (!objectsToDelete.isEmpty()) {
+ final DeleteObjectsRequest deleteRequest = DeleteObjectsRequest.builder()
+ .bucket(bucketName)
+ .delete(Delete.builder().objects(objectsToDelete).build())
+ .build();
+
+ s3.deleteObjects(deleteRequest);
+ }
+
+ // TODO what about continuation token?
+
+ } while (objectsListing.isTruncated());
+ }
+ }
+
+ class EtagMatch extends Unsafe {
+
+ public EtagMatch(S3Client s3, String bucketName) {
+ super(s3, bucketName);
+ }
+
+ @Override
+ public VolatileReadData read(String key) {
+ return VolatileReadData.from(new S3LazyRead(s3, bucketName, key, true));
+ }
+ }
+
+ class S3LazyRead implements LazyRead {
+
+ private final String s3Key;
+ private final boolean verifyEtag;
+ private final S3Client s3;
+ private final String bucketName;
+ private String eTag = null;
+
+
+ S3LazyRead(final S3Client s3, final String bucketName, final String s3Key, final boolean verifyEtag) {
+ this.s3 = s3;
+ this.bucketName = bucketName;
+ this.s3Key = s3Key;
+ this.verifyEtag = verifyEtag;
+ }
+
+ private GetObjectRequest createObjectRequest(final String s3Key, long offset, long length) {
+
+
+ final GetObjectRequest.Builder requestBuilder = GetObjectRequest.builder()
+ .key(s3Key)
+ .bucket(bucketName);
+
+ // Only add range header if we're doing a partial read
+ if (offset > 0 || length > 0) {
+ // HTTP Range header format: "bytes=start-end"
+ // If length is 0 or negative, read from offset to end of file
+ final String range = length > 0
+ ? String.format("bytes=%d-%d", offset, offset + length - 1)
+ : String.format("bytes=%d-", offset);
+ requestBuilder.range(range);
+ }
+
+ if (verifyEtag && eTag != null)
+ requestBuilder.ifMatch(eTag);
+
+ return requestBuilder.build();
+ }
+
+ @Override public ReadData materialize(long offset, long length) throws N5Exception.N5IOException {
+
+ final ResponseBytes response = rethrowS3Exceptions(() -> {
+ final GetObjectRequest request = createObjectRequest(s3Key, offset, length);
+ ResponseBytes responseBytes = s3.getObject(request, ResponseTransformer.toBytes());
+ if (verifyEtag && eTag == null)
+ eTag = responseBytes.response().eTag();
+ return responseBytes;
+ });
+ return ReadData.from(response.asByteArray());
+ }
+
+ @Override public long size() throws N5Exception.N5IOException {
+
+ final HeadObjectResponse response = headObjectRequest(s3, bucketName, s3Key, eTag);
+
+ if (verifyEtag && eTag == null)
+ eTag = response.eTag();
+
+ return response.contentLength();
+ }
+
+ @Override
+ public void close() {
+ eTag = null;
+ }
+ }
+}
diff --git a/src/test/java/org/janelia/saalfeldlab/n5/s3/backend/BackendIoPolicyTests.java b/src/test/java/org/janelia/saalfeldlab/n5/s3/backend/BackendIoPolicyTests.java
new file mode 100644
index 0000000..2650428
--- /dev/null
+++ b/src/test/java/org/janelia/saalfeldlab/n5/s3/backend/BackendIoPolicyTests.java
@@ -0,0 +1,112 @@
+package org.janelia.saalfeldlab.n5.s3.backend;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertThrows;
+
+import java.io.IOException;
+import java.net.URI;
+import java.security.SecureRandom;
+
+import org.janelia.saalfeldlab.n5.N5Exception;
+import org.janelia.saalfeldlab.n5.readdata.ReadData;
+import org.janelia.saalfeldlab.n5.readdata.VolatileReadData;
+import org.janelia.saalfeldlab.n5.s3.AmazonS3KeyValueAccess;
+import org.janelia.saalfeldlab.n5.s3.S3IoPolicy;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import software.amazon.awssdk.services.s3.S3Client;
+
+
+public class BackendIoPolicyTests {
+
+ private static final SecureRandom random = new SecureRandom();
+ static final String UNSAFE_KEY = "unsafe/obj";
+ static final String ETAG_KEY = "etag/obj";
+
+ protected static S3Client s3;
+ protected static AmazonS3KeyValueAccess kva;
+ protected static String bucketName;
+
+ @BeforeClass
+ public static void setup() {
+
+ s3 = BackendS3Factory.getOrCreateS3();
+ bucketName = "n5-test-" + Long.toUnsignedString(random.nextLong());
+ kva = new AmazonS3KeyValueAccess(s3, URI.create("s3://" + bucketName), true);
+ }
+
+ @AfterClass
+ public static void teardown() {
+
+ if (kva != null) {
+ kva.delete("/"); // delete the bucket and everything in it
+ }
+ }
+
+ @Test
+ public void testUnsafe() throws IOException {
+
+ final S3IoPolicy.Unsafe policy = new S3IoPolicy.Unsafe(s3, bucketName);
+ final byte[] data = {0, 1, 2, 3, 4};
+ final byte[] data2 = {5, 6, 7};
+
+ // write and full read roundtrip
+ policy.write(UNSAFE_KEY, ReadData.from(data));
+
+ try (VolatileReadData result = policy.read(UNSAFE_KEY)) {
+ assertArrayEquals(data, result.allBytes());
+ }
+
+ // concurrent modification:
+ // overwrite the blob, then attempt to materialize, shoud succeed
+ try (VolatileReadData vrd = policy.read(UNSAFE_KEY)) {
+
+ // should not store the generation
+ vrd.requireLength();
+
+ // overwrite to new generation
+ policy.write(UNSAFE_KEY, ReadData.from(data2));
+
+ // ensure fetching the data gets the new data and does not error
+ assertArrayEquals(data2, vrd.allBytes());
+ }
+ }
+
+ @Test
+ public void testEtagMatch() throws IOException {
+
+ final S3IoPolicy.EtagMatch policy = new S3IoPolicy.EtagMatch(s3, bucketName);
+ final byte[] data1 = {0, 1, 2, 3, 4};
+ final byte[] data2 = {5, 6, 7};
+
+ // write and full read roundtrip
+ policy.write(ETAG_KEY, ReadData.from(data1));
+ try (VolatileReadData result = policy.read(ETAG_KEY)) {
+ assertArrayEquals(data1, result.allBytes());
+ }
+
+ // after close(), a new read sees updated content
+ policy.write(ETAG_KEY, ReadData.from(data2));
+ try (VolatileReadData result = policy.read(ETAG_KEY)) {
+ assertArrayEquals(data2, result.allBytes());
+ }
+
+ // concurrent modification: pin the generation via requireLength(),
+ // overwrite the blob, then attempt to materialize
+ policy.write(ETAG_KEY, ReadData.from(data1));
+ try (VolatileReadData vrd = policy.read(ETAG_KEY)) {
+
+ // sets the generation of vrd
+ vrd.requireLength();
+
+ // overwrite to new generation
+ policy.write(ETAG_KEY, ReadData.from(data2));
+
+ // ensure fetching the data
+ assertThrows(N5Exception.N5ConcurrentModificationException.class, vrd::allBytes);
+ }
+ }
+
+}