You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hop.apache.org by ha...@apache.org on 2026/08/03 18:18:47 UTC
(hop) branch main updated: Add append support, fixes #7760 (#7766)
This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 8c4fae50f0 Add append support, fixes #7760 (#7766)
8c4fae50f0 is described below
commit 8c4fae50f0b3e4d7667c18116bb5702b28ad1a57
Author: Hans Van Akelyen <ha...@gmail.com>
AuthorDate: Mon Aug 3 20:18:41 2026 +0200
Add append support, fixes #7760 (#7766)
---
.../org/apache/hop/vfs/s3/s3/vfs/S3FileObject.java | 4 +-
.../hop/vfs/s3/s3common/S3AppendOutputStream.java | 292 +++++++++++++++++++
.../hop/vfs/s3/s3common/S3CommonFileObject.java | 41 ++-
.../hop/vfs/s3/s3common/S3CommonFileProvider.java | 7 +-
.../vfs/s3/s3common/S3CommonPipedOutputStream.java | 2 +-
.../vfs/s3/s3common/S3AppendOutputStreamTest.java | 320 +++++++++++++++++++++
.../S3CommonFileProviderCapabilitiesTest.java} | 35 ++-
.../apache/hop/vfs/s3/vfs/S3FileObjectTest.java | 44 ++-
.../org/apache/hop/vfs/azure/AzureFileObject.java | 83 +++++-
.../apache/hop/vfs/azure/AzureFileProvider.java | 6 +-
.../hop/vfs/azure/BlockBlobAppendOutputStream.java | 218 ++++++++++++++
.../azure/AzureFileProviderCapabilitiesTest.java} | 36 ++-
.../vfs/azure/BlockBlobAppendOutputStreamTest.java | 308 ++++++++++++++++++++
13 files changed, 1345 insertions(+), 51 deletions(-)
diff --git a/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3/vfs/S3FileObject.java b/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3/vfs/S3FileObject.java
index 38f7354f93..678e543a82 100644
--- a/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3/vfs/S3FileObject.java
+++ b/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3/vfs/S3FileObject.java
@@ -21,7 +21,6 @@ import java.io.OutputStream;
import org.apache.commons.vfs2.FileSystemException;
import org.apache.commons.vfs2.provider.AbstractFileName;
import org.apache.hop.vfs.s3.s3common.S3CommonFileObject;
-import org.apache.hop.vfs.s3.s3common.S3CommonPipedOutputStream;
public class S3FileObject extends S3CommonFileObject {
@@ -36,7 +35,6 @@ public class S3FileObject extends S3CommonFileObject {
@Override
public OutputStream doGetOutputStream(boolean bAppend) throws Exception {
- return new S3CommonPipedOutputStream(
- this.fileSystem, bucketName, key, ((S3FileSystem) this.fileSystem).getPartSize());
+ return createOutputStream(bAppend, ((S3FileSystem) this.fileSystem).getPartSize());
}
}
diff --git a/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3AppendOutputStream.java b/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3AppendOutputStream.java
new file mode 100644
index 0000000000..43538f0172
--- /dev/null
+++ b/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3AppendOutputStream.java
@@ -0,0 +1,292 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.vfs.s3.s3common;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.hop.core.logging.LogChannel;
+import software.amazon.awssdk.core.ResponseBytes;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.AbortMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.CompletedMultipartUpload;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
+import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+import software.amazon.awssdk.services.s3.model.UploadPartCopyRequest;
+import software.amazon.awssdk.services.s3.model.UploadPartCopyResponse;
+import software.amazon.awssdk.services.s3.model.UploadPartRequest;
+import software.amazon.awssdk.services.s3.model.UploadPartResponse;
+
+/**
+ * Append output stream for S3, built on <a
+ * href="https://docs.aws.amazon.com/AmazonS3/latest/userguide/mpuoverview.html">multipart
+ * uploads</a>.
+ *
+ * <p>S3 objects are immutable and the API has no append operation, so an append is expressed as a
+ * multipart upload that reassembles the object as {@code target = existing + appended bytes}. The
+ * existing content becomes the first part through {@code UploadPartCopy}, which is a server-side
+ * copy: the old content is never downloaded or re-uploaded, so appending a few rows to a large
+ * object only transfers those rows. The appended bytes follow as one or more regular parts, and a
+ * single {@code CompleteMultipartUpload} on {@link #close()} swaps the object over atomically.
+ *
+ * <p>S3 requires every part but the last to be at least {@value #MIN_PART_SIZE} bytes, so an
+ * existing object smaller than that cannot be copied as part 1. Those objects are read back and
+ * placed in front of the appended bytes instead — bounded by the same 5 MiB, so the fallback stays
+ * cheap.
+ *
+ * <p>Both paths are guarded by an {@code If-Match} on the ETag captured when the stream was opened,
+ * so a concurrent modification of the object fails the append fast instead of silently dropping
+ * data.
+ *
+ * <p>The multipart upload is only started on the first {@code write(...)} and nothing is visible
+ * until the completing call, so an append that writes nothing, or a pipeline that dies mid-write,
+ * leaves the object exactly as it was.
+ */
+public class S3AppendOutputStream extends OutputStream {
+
+ /**
+ * Minimum size of every part but the last, see <a
+ * href="https://docs.aws.amazon.com/AmazonS3/latest/userguide/qfacts.html">S3 multipart upload
+ * limits</a>. An existing object below this size cannot be copied in as part 1.
+ */
+ public static final int MIN_PART_SIZE = 5 * 1024 * 1024;
+
+ private final S3CommonFileSystem fileSystem;
+ private final String bucketName;
+ private final String key;
+ private final int partSize;
+
+ /** The object as it was when the stream opened: its size, ETag and metadata to carry over. */
+ private final HeadObjectResponse existing;
+
+ /** Appended bytes waiting to be uploaded; flushed out whenever it reaches {@link #partSize}. */
+ private final ByteArrayOutputStream buffer;
+
+ private final List<CompletedPart> completedParts = new ArrayList<>();
+ private String uploadId;
+ private int partNumber = 1;
+ private boolean started = false;
+ private boolean closed = false;
+
+ /** Set once the upload has been aborted, so a follow-up close does not report a second error. */
+ private boolean failed = false;
+
+ public S3AppendOutputStream(
+ S3CommonFileSystem fileSystem,
+ String bucketName,
+ String key,
+ int partSize,
+ HeadObjectResponse existing) {
+ this.fileSystem = fileSystem;
+ this.bucketName = bucketName;
+ this.key = key;
+ this.partSize = Math.max(partSize, MIN_PART_SIZE);
+ this.existing = existing;
+ this.buffer = new ByteArrayOutputStream(this.partSize);
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ start();
+ buffer.write(b);
+ uploadFullParts();
+ }
+
+ @Override
+ public void write(byte[] buf, int off, int len) throws IOException {
+ start();
+ buffer.write(buf, off, len);
+ uploadFullParts();
+ }
+
+ /**
+ * Open the multipart upload and seed it with the existing content, either as a server-side copied
+ * first part or, when the object is too small for a part, as the head of the buffer. Deferred to
+ * the first write so that an append writing nothing leaves the object untouched.
+ */
+ private void start() throws IOException {
+ if (started) {
+ return;
+ }
+ started = true;
+ S3Client s3 = fileSystem.getS3Client();
+ try {
+ // Rebuilding the object must not silently drop the metadata it was created with.
+ uploadId =
+ s3.createMultipartUpload(
+ CreateMultipartUploadRequest.builder()
+ .bucket(bucketName)
+ .key(key)
+ .contentType(existing.contentType())
+ .contentEncoding(existing.contentEncoding())
+ .metadata(existing.metadata())
+ .build())
+ .uploadId();
+
+ if (existing.contentLength() >= MIN_PART_SIZE) {
+ copyExistingObjectAsFirstPart(s3);
+ } else {
+ readExistingObjectIntoBuffer(s3);
+ }
+ } catch (RuntimeException e) {
+ failed = true;
+ abortQuietly();
+ throw new IOException(unableToAppend(), e);
+ }
+ }
+
+ /** Server-side copy of the whole existing object into part 1 — no bytes travel to the client. */
+ private void copyExistingObjectAsFirstPart(S3Client s3) {
+ UploadPartCopyResponse copy =
+ s3.uploadPartCopy(
+ UploadPartCopyRequest.builder()
+ .sourceBucket(bucketName)
+ .sourceKey(key)
+ .destinationBucket(bucketName)
+ .destinationKey(key)
+ .uploadId(uploadId)
+ .partNumber(partNumber)
+ .copySourceIfMatch(existing.eTag())
+ .build());
+ completedParts.add(
+ CompletedPart.builder().eTag(copy.copyPartResult().eTag()).partNumber(partNumber).build());
+ partNumber++;
+ }
+
+ /**
+ * The existing object is below the minimum part size, so it cannot be copied as a part. Read it
+ * back and let the appended bytes follow it in the buffer; it is under 5 MiB by definition.
+ */
+ private void readExistingObjectIntoBuffer(S3Client s3) {
+ ResponseBytes<GetObjectResponse> content =
+ s3.getObjectAsBytes(
+ GetObjectRequest.builder()
+ .bucket(bucketName)
+ .key(key)
+ .ifMatch(existing.eTag())
+ .build());
+ buffer.writeBytes(content.asByteArray());
+ }
+
+ private void uploadFullParts() throws IOException {
+ while (buffer.size() >= partSize) {
+ byte[] bytes = buffer.toByteArray();
+ buffer.reset();
+ uploadPart(bytes, partSize);
+ // Whatever overshot the part size is the start of the next one.
+ buffer.write(bytes, partSize, bytes.length - partSize);
+ }
+ }
+
+ /** Upload the first {@code len} bytes of {@code bytes} as the next part of the upload. */
+ private void uploadPart(byte[] bytes, int len) throws IOException {
+ byte[] part = len == bytes.length ? bytes : Arrays.copyOf(bytes, len);
+ try {
+ UploadPartResponse response =
+ fileSystem
+ .getS3Client()
+ .uploadPart(
+ UploadPartRequest.builder()
+ .bucket(bucketName)
+ .key(key)
+ .uploadId(uploadId)
+ .partNumber(partNumber)
+ .build(),
+ RequestBody.fromBytes(part));
+ completedParts.add(
+ CompletedPart.builder().eTag(response.eTag()).partNumber(partNumber).build());
+ partNumber++;
+ } catch (RuntimeException e) {
+ failed = true;
+ abortQuietly();
+ throw new IOException(unableToAppend(), e);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ if (!started || failed) {
+ // Nothing was appended, or the upload already failed and was aborted: either way the object
+ // stays exactly as it was and there is nothing left to complete.
+ return;
+ }
+ try {
+ if (buffer.size() > 0) {
+ byte[] last = buffer.toByteArray();
+ buffer.reset();
+ uploadPart(last, last.length);
+ }
+ fileSystem
+ .getS3Client()
+ .completeMultipartUpload(
+ CompleteMultipartUploadRequest.builder()
+ .bucket(bucketName)
+ .key(key)
+ .uploadId(uploadId)
+ .multipartUpload(CompletedMultipartUpload.builder().parts(completedParts).build())
+ .build());
+ fileSystem.invalidateListCacheForParentOf(bucketName, key);
+ } catch (RuntimeException e) {
+ failed = true;
+ abortQuietly();
+ throw new IOException(unableToAppend(), e);
+ }
+ }
+
+ /**
+ * Abandon the multipart upload so its parts do not linger as billed, invisible storage. A failed
+ * abort only leaves those parts behind, so it is logged rather than masking the original error.
+ */
+ private void abortQuietly() {
+ if (uploadId == null) {
+ return;
+ }
+ try {
+ fileSystem
+ .getS3Client()
+ .abortMultipartUpload(
+ AbortMultipartUploadRequest.builder()
+ .bucket(bucketName)
+ .key(key)
+ .uploadId(uploadId)
+ .build());
+ } catch (RuntimeException e) {
+ LogChannel.GENERAL.logError(
+ "Unable to abort the multipart upload used to append to s3://" + bucketName + "/" + key,
+ e);
+ } finally {
+ uploadId = null;
+ }
+ }
+
+ private String unableToAppend() {
+ return "Unable to append to s3://" + bucketName + "/" + key;
+ }
+}
diff --git a/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonFileObject.java b/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonFileObject.java
index 8ffeb7df33..3af4107a54 100644
--- a/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonFileObject.java
+++ b/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonFileObject.java
@@ -209,6 +209,11 @@ public abstract class S3CommonFileObject extends AbstractFileObject {
return e.getMessage();
}
+ private static boolean isNotFound(S3Exception e) {
+ String err = errorCode(e);
+ return "404".equals(err) || "NotFound".equals(err) || "NoSuchKey".equals(err);
+ }
+
@Override
public void doAttach() throws Exception {
LogChannel.GENERAL.logDebug("Attach called on {0}", getQualifiedName());
@@ -253,10 +258,8 @@ public abstract class S3CommonFileObject extends AbstractFileObject {
protected void handleAttachException(String key, String bucket, S3Exception e)
throws IOException {
String keyWithDelimiter = key + DELIMITER;
- String err = errorCode(e);
- boolean isNotFound = "404".equals(err) || "NotFound".equals(err) || "NoSuchKey".equals(err);
- if (isNotFound) {
+ if (isNotFound(e)) {
handleNotFoundAttach(bucket, keyWithDelimiter);
} else {
resolveTypeViaList(key, bucket, keyWithDelimiter);
@@ -439,7 +442,37 @@ public abstract class S3CommonFileObject extends AbstractFileObject {
@Override
protected OutputStream doGetOutputStream(boolean bAppend) throws Exception {
- return new S3CommonPipedOutputStream(this.fileSystem, bucketName, key);
+ return createOutputStream(bAppend, S3CommonPipedOutputStream.DEFAULT_PART_SIZE);
+ }
+
+ /**
+ * Open a write or append stream for this object. S3 has no append operation, so appending to an
+ * existing object is emulated with a multipart upload, see {@link S3AppendOutputStream}.
+ */
+ protected OutputStream createOutputStream(boolean bAppend, int partSize) throws IOException {
+ // A bucket itself has no content to append to, so only a real key can be appended to.
+ if (bAppend && !isRootBucket()) {
+ HeadObjectResponse head = headExistingObject();
+ if (head != null && head.contentLength() != null && head.contentLength() > 0) {
+ return new S3AppendOutputStream(fileSystem, bucketName, key, partSize, head);
+ }
+ // No object to append to yet, so plainly creating it is the correct append semantic.
+ }
+ return new S3CommonPipedOutputStream(this.fileSystem, bucketName, key, partSize);
+ }
+
+ /** Head this object, returning null when it does not exist yet. */
+ private HeadObjectResponse headExistingObject() {
+ try {
+ return fileSystem
+ .getS3Client()
+ .headObject(HeadObjectRequest.builder().bucket(bucketName).key(key).build());
+ } catch (S3Exception e) {
+ if (isNotFound(e)) {
+ return null;
+ }
+ throw e;
+ }
}
@Override
diff --git a/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonFileProvider.java b/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonFileProvider.java
index 4ceb6aaacc..527cbc4de5 100644
--- a/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonFileProvider.java
+++ b/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonFileProvider.java
@@ -42,7 +42,11 @@ public abstract class S3CommonFileProvider extends AbstractOriginatingFileProvid
return defaultOptions;
}
- /** The provider's capabilities. */
+ /**
+ * The provider's capabilities. APPEND_CONTENT is required for append: without it commons-vfs2
+ * rejects getOutputStream(true) with "does not support append mode". S3 has no append operation,
+ * so append itself is emulated with a multipart upload, see {@link S3AppendOutputStream}.
+ */
protected static final Collection<Capability> capabilities =
Collections.unmodifiableCollection(
Arrays.asList(
@@ -54,6 +58,7 @@ public abstract class S3CommonFileProvider extends AbstractOriginatingFileProvid
Capability.READ_CONTENT,
Capability.URI,
Capability.WRITE_CONTENT,
+ Capability.APPEND_CONTENT,
Capability.GET_LAST_MODIFIED,
Capability.RANDOM_ACCESS_READ));
diff --git a/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStream.java b/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStream.java
index 4566a0e9ae..16004ca4c4 100644
--- a/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStream.java
+++ b/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStream.java
@@ -51,7 +51,7 @@ public class S3CommonPipedOutputStream extends PipedOutputStream {
private static final Class<?> PKG = S3CommonPipedOutputStream.class;
- private static final int DEFAULT_PART_SIZE = 5 * 1024 * 1024;
+ public static final int DEFAULT_PART_SIZE = 5 * 1024 * 1024;
private final ThreadPoolExecutor executor =
(ThreadPoolExecutor)
diff --git a/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/s3common/S3AppendOutputStreamTest.java b/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/s3common/S3AppendOutputStreamTest.java
new file mode 100644
index 0000000000..2c3de3ca97
--- /dev/null
+++ b/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/s3common/S3AppendOutputStreamTest.java
@@ -0,0 +1,320 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.vfs.s3.s3common;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
+import org.mockito.Mockito;
+import software.amazon.awssdk.core.ResponseBytes;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.AbortMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.CopyPartResult;
+import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.CreateMultipartUploadResponse;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+import software.amazon.awssdk.services.s3.model.UploadPartCopyRequest;
+import software.amazon.awssdk.services.s3.model.UploadPartCopyResponse;
+import software.amazon.awssdk.services.s3.model.UploadPartRequest;
+import software.amazon.awssdk.services.s3.model.UploadPartResponse;
+
+/**
+ * Covers the multipart-upload append behaviour of {@link S3AppendOutputStream}: the existing object
+ * is brought in as the head of a new multipart upload — server-side copied when it is large enough
+ * to be a part, read back when it is not — and the appended bytes follow it before a single
+ * completing call swaps the object over.
+ */
+class S3AppendOutputStreamTest {
+
+ private static final String BUCKET = "my-bucket";
+ private static final String KEY = "folder/data.txt";
+ private static final String ETAG = "\"abc123\"";
+ private static final String UPLOAD_ID = "upload-1";
+ private static final int PART_SIZE = S3AppendOutputStream.MIN_PART_SIZE;
+
+ private S3Client s3;
+ private S3CommonFileSystem fileSystem;
+
+ /** Payload of every uploadPart call, in order, so we can assert on what was written. */
+ private List<byte[]> uploadedParts;
+
+ @BeforeEach
+ void setUp() {
+ s3 = mock(S3Client.class);
+ fileSystem = mock(S3CommonFileSystem.class);
+ when(fileSystem.getS3Client()).thenReturn(s3);
+
+ when(s3.createMultipartUpload(any(CreateMultipartUploadRequest.class)))
+ .thenReturn(CreateMultipartUploadResponse.builder().uploadId(UPLOAD_ID).build());
+ when(s3.uploadPartCopy(any(UploadPartCopyRequest.class)))
+ .thenReturn(
+ UploadPartCopyResponse.builder()
+ .copyPartResult(CopyPartResult.builder().eTag("\"copied\"").build())
+ .build());
+
+ uploadedParts = new ArrayList<>();
+ when(s3.uploadPart(any(UploadPartRequest.class), any(RequestBody.class)))
+ .thenAnswer(
+ invocation -> {
+ RequestBody body = invocation.getArgument(1);
+ uploadedParts.add(readFully(body));
+ return UploadPartResponse.builder().eTag("\"part\"").build();
+ });
+ }
+
+ private static byte[] readFully(RequestBody body) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ body.contentStreamProvider().newStream().transferTo(out);
+ return out.toByteArray();
+ }
+
+ private S3AppendOutputStream openAppend(long existingSize) {
+ return openAppend(HeadObjectResponse.builder().contentLength(existingSize).eTag(ETAG).build());
+ }
+
+ private S3AppendOutputStream openAppend(HeadObjectResponse existing) {
+ return new S3AppendOutputStream(fileSystem, BUCKET, KEY, PART_SIZE, existing);
+ }
+
+ @Test
+ void largeExistingObjectIsCopiedServerSideAsTheFirstPart() throws Exception {
+ S3AppendOutputStream out = openAppend(10L * 1024 * 1024);
+ byte[] payload = "appended".getBytes(StandardCharsets.UTF_8);
+ out.write(payload);
+ out.close();
+
+ // The existing content becomes part 1 through a copy: it is never downloaded...
+ ArgumentCaptor<UploadPartCopyRequest> copy =
+ ArgumentCaptor.forClass(UploadPartCopyRequest.class);
+ verify(s3).uploadPartCopy(copy.capture());
+ assertEquals(BUCKET, copy.getValue().sourceBucket());
+ assertEquals(KEY, copy.getValue().sourceKey());
+ assertEquals(BUCKET, copy.getValue().destinationBucket());
+ assertEquals(KEY, copy.getValue().destinationKey());
+ assertEquals(1, copy.getValue().partNumber());
+ verify(s3, never()).getObjectAsBytes(any(GetObjectRequest.class));
+
+ // ...and the appended bytes follow it as part 2, so the result is existing + appended.
+ ArgumentCaptor<UploadPartRequest> part = ArgumentCaptor.forClass(UploadPartRequest.class);
+ verify(s3).uploadPart(part.capture(), any(RequestBody.class));
+ assertEquals(2, part.getValue().partNumber());
+ assertEquals(UPLOAD_ID, part.getValue().uploadId());
+ assertArrayEquals(payload, uploadedParts.get(0));
+
+ verify(s3).completeMultipartUpload(any(CompleteMultipartUploadRequest.class));
+ }
+
+ @Test
+ void smallExistingObjectIsReadBackAheadOfTheAppendedBytes() throws Exception {
+ byte[] existing = "existing;".getBytes(StandardCharsets.UTF_8);
+ when(s3.getObjectAsBytes(any(GetObjectRequest.class)))
+ .thenReturn(ResponseBytes.fromByteArray(GetObjectResponse.builder().build(), existing));
+
+ S3AppendOutputStream out = openAppend(existing.length);
+ byte[] payload = "appended".getBytes(StandardCharsets.UTF_8);
+ out.write(payload);
+ out.close();
+
+ // Below the 5 MiB minimum a part copy is not allowed, so the object is read back instead.
+ verify(s3, never()).uploadPartCopy(any(UploadPartCopyRequest.class));
+ verify(s3).getObjectAsBytes(any(GetObjectRequest.class));
+
+ // Existing content first, appended content after it, in a single part.
+ assertEquals(1, uploadedParts.size());
+ assertEquals("existing;appended", new String(uploadedParts.get(0), StandardCharsets.UTF_8));
+ verify(s3).completeMultipartUpload(any(CompleteMultipartUploadRequest.class));
+ }
+
+ @Test
+ void bothPathsAreGuardedByTheOpenTimeETag() throws Exception {
+ S3AppendOutputStream copyPath = openAppend(10L * 1024 * 1024);
+ copyPath.write("x".getBytes(StandardCharsets.UTF_8));
+ copyPath.close();
+
+ ArgumentCaptor<UploadPartCopyRequest> copy =
+ ArgumentCaptor.forClass(UploadPartCopyRequest.class);
+ verify(s3).uploadPartCopy(copy.capture());
+ assertEquals(
+ ETAG,
+ copy.getValue().copySourceIfMatch(),
+ "the copy must fail fast if the object changed since the stream opened");
+
+ when(s3.getObjectAsBytes(any(GetObjectRequest.class)))
+ .thenReturn(ResponseBytes.fromByteArray(GetObjectResponse.builder().build(), new byte[3]));
+ S3AppendOutputStream readPath = openAppend(3L);
+ readPath.write("x".getBytes(StandardCharsets.UTF_8));
+ readPath.close();
+
+ ArgumentCaptor<GetObjectRequest> get = ArgumentCaptor.forClass(GetObjectRequest.class);
+ verify(s3).getObjectAsBytes(get.capture());
+ assertEquals(
+ ETAG,
+ get.getValue().ifMatch(),
+ "the read-back must fail fast if the object changed since the stream opened");
+ }
+
+ @Test
+ void writesBeyondThePartSizeAreSplitIntoParts() throws Exception {
+ // One and a half parts of appended data on top of a copied first part.
+ byte[] payload = new byte[PART_SIZE + PART_SIZE / 2];
+ for (int i = 0; i < payload.length; i++) {
+ payload[i] = (byte) (i % 251);
+ }
+
+ S3AppendOutputStream out = openAppend(10L * 1024 * 1024);
+ out.write(payload);
+ out.close();
+
+ assertEquals(2, uploadedParts.size(), "a part and a half must be uploaded as two parts");
+ assertEquals(PART_SIZE, uploadedParts.get(0).length, "full parts must be exactly partSize");
+ ByteArrayOutputStream reassembled = new ByteArrayOutputStream();
+ reassembled.write(uploadedParts.get(0));
+ reassembled.write(uploadedParts.get(1));
+ assertArrayEquals(payload, reassembled.toByteArray(), "no bytes may be lost or reordered");
+
+ // Parts are numbered after the copied part 1 and in ascending order.
+ ArgumentCaptor<UploadPartRequest> parts = ArgumentCaptor.forClass(UploadPartRequest.class);
+ verify(s3, times(2)).uploadPart(parts.capture(), any(RequestBody.class));
+ assertEquals(
+ List.of(2, 3), parts.getAllValues().stream().map(UploadPartRequest::partNumber).toList());
+ }
+
+ @Test
+ void theMetadataOfTheExistingObjectIsCarriedOver() throws Exception {
+ S3AppendOutputStream out =
+ openAppend(
+ HeadObjectResponse.builder()
+ .contentLength(10L * 1024 * 1024)
+ .eTag(ETAG)
+ .contentType("text/csv")
+ .contentEncoding("gzip")
+ .metadata(java.util.Map.of("origin", "hop"))
+ .build());
+ out.write("x".getBytes(StandardCharsets.UTF_8));
+ out.close();
+
+ // The append rebuilds the object, so anything it was created with must survive the rewrite.
+ ArgumentCaptor<CreateMultipartUploadRequest> create =
+ ArgumentCaptor.forClass(CreateMultipartUploadRequest.class);
+ verify(s3).createMultipartUpload(create.capture());
+ assertEquals("text/csv", create.getValue().contentType());
+ assertEquals("gzip", create.getValue().contentEncoding());
+ assertEquals(java.util.Map.of("origin", "hop"), create.getValue().metadata());
+ }
+
+ @Test
+ void nothingIsCommittedBeforeClose() throws Exception {
+ S3AppendOutputStream out = openAppend(10L * 1024 * 1024);
+ out.write("streaming".getBytes(StandardCharsets.UTF_8));
+
+ verify(s3, never()).completeMultipartUpload(any(CompleteMultipartUploadRequest.class));
+
+ out.close();
+ verify(s3, times(1)).completeMultipartUpload(any(CompleteMultipartUploadRequest.class));
+ }
+
+ @Test
+ void closingWithoutWritingLeavesTheObjectAlone() throws Exception {
+ S3AppendOutputStream out = openAppend(10L * 1024 * 1024);
+ out.close();
+
+ // Opening an append and writing nothing must not start an upload, let alone rewrite the object.
+ verify(s3, never()).createMultipartUpload(any(CreateMultipartUploadRequest.class));
+ verify(s3, never()).completeMultipartUpload(any(CompleteMultipartUploadRequest.class));
+ }
+
+ @Test
+ void closeIsIdempotent() throws Exception {
+ S3AppendOutputStream out = openAppend(10L * 1024 * 1024);
+ out.write("x".getBytes(StandardCharsets.UTF_8));
+ out.close();
+ out.close();
+
+ // A second close must not upload or complete a second time.
+ verify(s3, times(1)).uploadPart(any(UploadPartRequest.class), any(RequestBody.class));
+ verify(s3, times(1)).completeMultipartUpload(any(CompleteMultipartUploadRequest.class));
+ }
+
+ @Test
+ void aFailedCopyAbortsTheUploadAndIsReportedAsAnIoException() {
+ when(s3.uploadPartCopy(any(UploadPartCopyRequest.class)))
+ .thenThrow(S3Exception.builder().message("precondition failed").statusCode(412).build());
+
+ S3AppendOutputStream out = openAppend(10L * 1024 * 1024);
+ IOException thrown =
+ assertThrows(IOException.class, () -> out.write("x".getBytes(StandardCharsets.UTF_8)));
+ assertTrue(thrown.getMessage().contains(KEY), "the error should name the target object");
+
+ // A half-started multipart upload keeps billing for its parts, so it must be abandoned.
+ verify(s3).abortMultipartUpload(any(AbortMultipartUploadRequest.class));
+ }
+
+ @Test
+ void closingAfterAFailedWriteDoesNotReportASecondError() {
+ when(s3.uploadPartCopy(any(UploadPartCopyRequest.class)))
+ .thenThrow(S3Exception.builder().message("precondition failed").statusCode(412).build());
+
+ S3AppendOutputStream out = openAppend(10L * 1024 * 1024);
+ assertThrows(IOException.class, () -> out.write("x".getBytes(StandardCharsets.UTF_8)));
+
+ // Streams are closed in a finally block, and that close must not bury the real error under a
+ // second failure, nor try to complete an upload that was already abandoned.
+ assertDoesNotThrow(out::close);
+ verify(s3, never()).completeMultipartUpload(any(CompleteMultipartUploadRequest.class));
+ verify(s3, times(1)).abortMultipartUpload(any(AbortMultipartUploadRequest.class));
+ }
+
+ @Test
+ void aFailedCompletionAbortsTheUploadAndIsReportedAsAnIoException() throws Exception {
+ when(s3.completeMultipartUpload(any(CompleteMultipartUploadRequest.class)))
+ .thenThrow(S3Exception.builder().message("boom").statusCode(500).build());
+
+ S3AppendOutputStream out = openAppend(10L * 1024 * 1024);
+ out.write("x".getBytes(StandardCharsets.UTF_8));
+
+ IOException thrown = assertThrows(IOException.class, out::close);
+ assertTrue(thrown.getMessage().contains(KEY), "the error should name the target object");
+
+ InOrder inOrder = Mockito.inOrder(s3);
+ inOrder.verify(s3).completeMultipartUpload(any(CompleteMultipartUploadRequest.class));
+ inOrder.verify(s3).abortMultipartUpload(any(AbortMultipartUploadRequest.class));
+ }
+}
diff --git a/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3/vfs/S3FileObject.java b/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/s3common/S3CommonFileProviderCapabilitiesTest.java
similarity index 50%
copy from plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3/vfs/S3FileObject.java
copy to plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/s3common/S3CommonFileProviderCapabilitiesTest.java
index 38f7354f93..6148becb56 100644
--- a/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3/vfs/S3FileObject.java
+++ b/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/s3common/S3CommonFileProviderCapabilitiesTest.java
@@ -15,28 +15,25 @@
* limitations under the License.
*/
-package org.apache.hop.vfs.s3.s3.vfs;
+package org.apache.hop.vfs.s3.s3common;
-import java.io.OutputStream;
-import org.apache.commons.vfs2.FileSystemException;
-import org.apache.commons.vfs2.provider.AbstractFileName;
-import org.apache.hop.vfs.s3.s3common.S3CommonFileObject;
-import org.apache.hop.vfs.s3.s3common.S3CommonPipedOutputStream;
+import static org.junit.jupiter.api.Assertions.assertTrue;
-public class S3FileObject extends S3CommonFileObject {
+import org.apache.commons.vfs2.Capability;
+import org.junit.jupiter.api.Test;
- public S3FileObject(final AbstractFileName name, final S3FileSystem fileSystem) {
- super(name, fileSystem);
- }
-
- @Override
- public void doDelete() throws FileSystemException {
- super.doDelete();
- }
+class S3CommonFileProviderCapabilitiesTest {
- @Override
- public OutputStream doGetOutputStream(boolean bAppend) throws Exception {
- return new S3CommonPipedOutputStream(
- this.fileSystem, bucketName, key, ((S3FileSystem) this.fileSystem).getPartSize());
+ /**
+ * commons-vfs2 rejects {@code getOutputStream(true)} with "does not support append mode" unless
+ * the file system advertises {@link Capability#APPEND_CONTENT}. Append is emulated with a
+ * multipart upload (see {@link S3AppendOutputStream}), so the capability must be present or every
+ * append is blocked before reaching the provider.
+ */
+ @Test
+ void appendContentIsAdvertised() {
+ assertTrue(
+ S3CommonFileProvider.capabilities.contains(Capability.APPEND_CONTENT),
+ "S3 VFS must advertise APPEND_CONTENT so commons-vfs2 allows append mode");
}
}
diff --git a/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/vfs/S3FileObjectTest.java b/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/vfs/S3FileObjectTest.java
index e44de11a79..94f13f6f07 100644
--- a/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/vfs/S3FileObjectTest.java
+++ b/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/vfs/S3FileObjectTest.java
@@ -29,7 +29,9 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.io.ByteArrayOutputStream;
import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
@@ -56,6 +58,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import software.amazon.awssdk.awscore.exception.AwsErrorDetails;
+import software.amazon.awssdk.core.ResponseBytes;
import software.amazon.awssdk.core.ResponseInputStream;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.Bucket;
@@ -201,8 +204,7 @@ class S3FileObjectTest {
any(software.amazon.awssdk.core.sync.RequestBody.class)))
.thenReturn(uploadPartResponse);
- assertNotNull(s3FileObjectBucketSpy.doGetOutputStream(false));
- OutputStream out = s3FileObjectBucketSpy.doGetOutputStream(true);
+ OutputStream out = s3FileObjectBucketSpy.doGetOutputStream(false);
assertNotNull(out);
out.write(new byte[1024 * 1024 * 6]); // 6MB
out.close();
@@ -216,6 +218,44 @@ class S3FileObjectTest {
any(software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest.class));
}
+ /**
+ * Appending to an existing object must not truncate it: the object is rebuilt as existing +
+ * appended through a multipart upload rather than overwritten, see {@link
+ * org.apache.hop.vfs.s3.s3common.S3AppendOutputStream}.
+ */
+ @Test
+ void testDoGetOutputStreamAppendsToExistingObject() throws Exception {
+ when(s3ClientMock.createMultipartUpload(any(CreateMultipartUploadRequest.class)))
+ .thenReturn(CreateMultipartUploadResponse.builder().uploadId("foo").build());
+ when(s3ClientMock.uploadPart(
+ any(software.amazon.awssdk.services.s3.model.UploadPartRequest.class),
+ any(software.amazon.awssdk.core.sync.RequestBody.class)))
+ .thenReturn(UploadPartResponse.builder().eTag("etag").build());
+ // The existing object is well under the 5MB minimum part size, so it is read back and the
+ // appended bytes are written behind it.
+ when(s3ClientMock.getObjectAsBytes(any(GetObjectRequest.class)))
+ .thenReturn(
+ ResponseBytes.fromByteArray(
+ GetObjectResponse.builder().build(), "existing;".getBytes(StandardCharsets.UTF_8)));
+
+ OutputStream out = s3FileObjectFileSpy.doGetOutputStream(true);
+ assertNotNull(out);
+ out.write("appended".getBytes(StandardCharsets.UTF_8));
+ out.close();
+
+ ArgumentCaptor<software.amazon.awssdk.core.sync.RequestBody> body =
+ ArgumentCaptor.forClass(software.amazon.awssdk.core.sync.RequestBody.class);
+ verify(s3ClientMock)
+ .uploadPart(
+ any(software.amazon.awssdk.services.s3.model.UploadPartRequest.class), body.capture());
+ ByteArrayOutputStream uploaded = new ByteArrayOutputStream();
+ body.getValue().contentStreamProvider().newStream().transferTo(uploaded);
+ assertEquals("existing;appended", uploaded.toString(StandardCharsets.UTF_8));
+ verify(s3ClientMock)
+ .completeMultipartUpload(
+ any(software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest.class));
+ }
+
@Test
void testDoGetInputStream() throws Exception {
assertNotNull(s3FileObjectBucketSpy.getInputStream());
diff --git a/plugins/tech/azure/src/main/java/org/apache/hop/vfs/azure/AzureFileObject.java b/plugins/tech/azure/src/main/java/org/apache/hop/vfs/azure/AzureFileObject.java
index 7682e3da2c..42a4934878 100644
--- a/plugins/tech/azure/src/main/java/org/apache/hop/vfs/azure/AzureFileObject.java
+++ b/plugins/tech/azure/src/main/java/org/apache/hop/vfs/azure/AzureFileObject.java
@@ -20,6 +20,11 @@ package org.apache.hop.vfs.azure;
import com.azure.core.http.rest.PagedIterable;
import com.azure.core.util.Context;
+import com.azure.storage.blob.BlobClientBuilder;
+import com.azure.storage.blob.models.BlobHttpHeaders;
+import com.azure.storage.blob.models.BlobProperties;
+import com.azure.storage.blob.models.BlobStorageException;
+import com.azure.storage.blob.specialized.BlockBlobClient;
import com.azure.storage.file.datalake.DataLakeDirectoryClient;
import com.azure.storage.file.datalake.DataLakeFileClient;
import com.azure.storage.file.datalake.DataLakeFileSystemClient;
@@ -475,7 +480,11 @@ public class AzureFileObject extends AbstractFileObject<AzureFileSystem> {
DataLakeFileClient dataLakeFileClient = fileSystemClient.getFileClient(currentFilePath);
if (dataLakeFileClient != null) {
if (bAppend) {
- throw new UnsupportedOperationException();
+ OutputStream appendStream = openAppendStream(dataLakeFileClient);
+ if (appendStream != null) {
+ return appendStream;
+ }
+ // Nothing to append to yet, so creating the file below is the correct append semantic.
}
type = FileType.FILE;
getAbstractFileSystem().invalidateListCacheForParentOf(containerName, currentFilePath);
@@ -485,6 +494,78 @@ public class AzureFileObject extends AbstractFileObject<AzureFileSystem> {
}
}
+ /**
+ * Open a {@link BlockBlobAppendOutputStream} on an existing, non-empty file.
+ *
+ * <p>Appending has to use the Blob API rather than the Data Lake one: files are written here with
+ * {@code DataLakeFileClient.getOutputStream()}, which is a block blob written through the Blob
+ * API, and the Data Lake {@code append} operation rejects such files with {@code 409
+ * InvalidAppendOperation}.
+ *
+ * @return the append stream, or null when there is no content to append to and the file should
+ * simply be created
+ */
+ private OutputStream openAppendStream(DataLakeFileClient fileClient) throws IOException {
+ BlockBlobClient blob = blockBlobClientFor(fileClient);
+ BlobProperties properties;
+ try {
+ properties = blob.getProperties();
+ } catch (BlobStorageException e) {
+ if (e.getStatusCode() == 404) {
+ return null;
+ }
+ throw e;
+ }
+ if (properties.getBlobSize() <= 0) {
+ return null;
+ }
+
+ List<String> committedBlockIds = BlockBlobAppendOutputStream.committedBlockIds(blob);
+ BlockBlobAppendOutputStream appendStream =
+ new BlockBlobAppendOutputStream(
+ blob,
+ committedBlockIds,
+ contentHeadersOf(properties),
+ properties.getMetadata(),
+ properties.getETag());
+ if (committedBlockIds.isEmpty()) {
+ // A blob uploaded in a single request has no block list to build on, so its content has to be
+ // carried forward or committing our blocks alone would drop it.
+ appendStream.restageExistingContent();
+ }
+
+ type = FileType.FILE;
+ getAbstractFileSystem().invalidateListCacheForParentOf(containerName, currentFilePath);
+ return appendStream;
+ }
+
+ /**
+ * A Blob API client for the same file, reusing the Data Lake client's authenticated pipeline so
+ * no credential handling is duplicated. Only the endpoint differs between the two APIs, which is
+ * the same swap the Azure SDK makes internally.
+ */
+ private static BlockBlobClient blockBlobClientFor(DataLakeFileClient fileClient) {
+ String blobUrl = fileClient.getFileUrl().replaceFirst("\\.dfs\\.", ".blob.");
+ return new BlobClientBuilder()
+ .endpoint(blobUrl)
+ .pipeline(fileClient.getHttpPipeline())
+ .buildClient()
+ .getBlockBlobClient();
+ }
+
+ /**
+ * The content headers the blob already carries. Committing a block list only keeps the headers it
+ * is handed, so these have to be replayed or appending would reset the blob's content type.
+ */
+ private static BlobHttpHeaders contentHeadersOf(BlobProperties properties) {
+ return new BlobHttpHeaders()
+ .setCacheControl(properties.getCacheControl())
+ .setContentDisposition(properties.getContentDisposition())
+ .setContentEncoding(properties.getContentEncoding())
+ .setContentLanguage(properties.getContentLanguage())
+ .setContentType(properties.getContentType());
+ }
+
@Override
protected InputStream doGetInputStream() throws Exception {
if (!currentFilePath.equals("") && type == FileType.FILE) {
diff --git a/plugins/tech/azure/src/main/java/org/apache/hop/vfs/azure/AzureFileProvider.java b/plugins/tech/azure/src/main/java/org/apache/hop/vfs/azure/AzureFileProvider.java
index 4f3616828d..1c3ae3d369 100644
--- a/plugins/tech/azure/src/main/java/org/apache/hop/vfs/azure/AzureFileProvider.java
+++ b/plugins/tech/azure/src/main/java/org/apache/hop/vfs/azure/AzureFileProvider.java
@@ -49,6 +49,9 @@ import org.apache.hop.vfs.azure.metadatatype.AzureMetadataType;
public class AzureFileProvider extends AbstractOriginatingFileProvider {
+ // APPEND_CONTENT is required for append: without it commons-vfs2 rejects getOutputStream(true)
+ // with "does not support append mode". Data Lake Gen2 appends natively, see
+ // DataLakeAppendOutputStream.
public static final Collection<Capability> capabilities =
Collections.unmodifiableCollection(
Arrays.asList(
@@ -61,7 +64,8 @@ public class AzureFileProvider extends AbstractOriginatingFileProvider {
Capability.LIST_CHILDREN,
Capability.READ_CONTENT,
Capability.URI,
- Capability.WRITE_CONTENT));
+ Capability.WRITE_CONTENT,
+ Capability.APPEND_CONTENT));
public static final UserAuthenticationData.Type[] AUTHENTICATOR_TYPES =
new UserAuthenticationData.Type[] {
diff --git a/plugins/tech/azure/src/main/java/org/apache/hop/vfs/azure/BlockBlobAppendOutputStream.java b/plugins/tech/azure/src/main/java/org/apache/hop/vfs/azure/BlockBlobAppendOutputStream.java
new file mode 100644
index 0000000000..0982f26cbb
--- /dev/null
+++ b/plugins/tech/azure/src/main/java/org/apache/hop/vfs/azure/BlockBlobAppendOutputStream.java
@@ -0,0 +1,218 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.hop.vfs.azure;
+
+import com.azure.storage.blob.models.BlobHttpHeaders;
+import com.azure.storage.blob.models.BlobRequestConditions;
+import com.azure.storage.blob.models.Block;
+import com.azure.storage.blob.models.BlockListType;
+import com.azure.storage.blob.options.BlockBlobCommitBlockListOptions;
+import com.azure.storage.blob.specialized.BlockBlobClient;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.hop.core.logging.LogChannel;
+
+/**
+ * Append output stream for Azure block blobs.
+ *
+ * <p>Hop writes Azure files through {@code DataLakeFileClient.getOutputStream()}, which is a block
+ * blob written with the <em>Blob</em> API. The Data Lake {@code append} operation refuses those
+ * files outright — {@code 409 InvalidAppendOperation, "The resource was created or modified by the
+ * Azure Blob Service API and cannot be appended to by the Azure Data Lake Storage Service API"} —
+ * so appending has to speak the same API that wrote the file.
+ *
+ * <p>A block blob is a list of blocks, so appending is: keep the blob's existing committed block
+ * IDs, stage the new bytes as extra blocks, and commit the old IDs followed by the new ones. The
+ * existing blocks are only referenced by ID and are never re-uploaded, so appending a few rows to a
+ * large blob costs only the size of those rows.
+ *
+ * <p>Block IDs are generated exactly as the SDK's own block blob output stream generates them —
+ * base64 of a random UUID string — because Azure requires every block ID within one blob to be the
+ * same length, and the existing IDs came from that writer.
+ *
+ * <p>The commit replays the blob's headers and metadata, since committing a block list otherwise
+ * resets them, and is guarded by an {@code If-Match} on the ETag captured when the stream was
+ * opened so a concurrent modification fails the append instead of silently dropping data.
+ *
+ * <p>Nothing is visible until {@link #close()}: staged but uncommitted blocks are not part of the
+ * blob, so a pipeline that dies mid-write leaves the blob with its original content.
+ */
+public class BlockBlobAppendOutputStream extends OutputStream {
+
+ /** Matches the block size the SDK's own block blob output stream uses for buffered writes. */
+ public static final int DEFAULT_BLOCK_SIZE = 4 * 1024 * 1024;
+
+ private final BlockBlobClient blob;
+ private final BlobHttpHeaders headers;
+ private final Map<String, String> metadata;
+ private final String eTag;
+
+ /** The blob's existing committed block IDs, followed by the ones staged by this stream. */
+ private final List<String> blockIds;
+
+ private final byte[] block;
+ private int buffered;
+
+ private boolean staged = false;
+ private boolean failed = false;
+ private boolean closed = false;
+
+ public BlockBlobAppendOutputStream(
+ BlockBlobClient blob,
+ List<String> committedBlockIds,
+ BlobHttpHeaders headers,
+ Map<String, String> metadata,
+ String eTag) {
+ this(blob, committedBlockIds, headers, metadata, eTag, DEFAULT_BLOCK_SIZE);
+ }
+
+ public BlockBlobAppendOutputStream(
+ BlockBlobClient blob,
+ List<String> committedBlockIds,
+ BlobHttpHeaders headers,
+ Map<String, String> metadata,
+ String eTag,
+ int blockSize) {
+ this.blob = blob;
+ this.blockIds = new ArrayList<>(committedBlockIds);
+ this.headers = headers;
+ this.metadata = metadata;
+ this.eTag = eTag;
+ this.block = new byte[blockSize];
+ }
+
+ /** The committed block IDs of an existing blob, in order. */
+ public static List<String> committedBlockIds(BlockBlobClient blob) {
+ return blob.listBlocks(BlockListType.COMMITTED).getCommittedBlocks().stream()
+ .map(Block::getName)
+ .toList();
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ if (buffered == block.length) {
+ stageBufferedBlock();
+ }
+ block[buffered++] = (byte) b;
+ }
+
+ @Override
+ public void write(byte[] buf, int off, int len) throws IOException {
+ int remaining = len;
+ int offset = off;
+ while (remaining > 0) {
+ if (buffered == block.length) {
+ stageBufferedBlock();
+ }
+ int copied = Math.min(remaining, block.length - buffered);
+ System.arraycopy(buf, offset, block, buffered, copied);
+ buffered += copied;
+ offset += copied;
+ remaining -= copied;
+ }
+ }
+
+ /**
+ * Stage the buffered bytes as one new block. A staged block is not part of the blob until the
+ * block list is committed.
+ */
+ private void stageBufferedBlock() throws IOException {
+ if (buffered == 0) {
+ return;
+ }
+ String blockId = newBlockId();
+ try {
+ blob.stageBlock(blockId, new ByteArrayInputStream(block, 0, buffered), buffered);
+ } catch (RuntimeException e) {
+ failed = true;
+ throw new IOException(
+ "Unable to stage " + buffered + " appended bytes for " + blob.getBlobName(), e);
+ }
+ blockIds.add(blockId);
+ buffered = 0;
+ staged = true;
+ }
+
+ /**
+ * Block IDs must be the same length for every block in a blob. This mirrors the SDK's own block
+ * blob writer — base64 of a random UUID string — so IDs staged here line up with the ones already
+ * on the blob.
+ */
+ private static String newBlockId() {
+ return Base64.getEncoder()
+ .encodeToString(UUID.randomUUID().toString().getBytes(StandardCharsets.UTF_8));
+ }
+
+ /**
+ * Copy the blob's current content into freshly staged blocks. Only needed when the blob has no
+ * committed block list of its own — a blob uploaded in a single {@code Put Blob} call has none,
+ * and its content would be lost by a commit that only lists our new blocks.
+ */
+ void restageExistingContent() throws IOException {
+ LogChannel.GENERAL.logBasic(
+ "Azure blob "
+ + blob.getBlobName()
+ + " has no committed block list (it was uploaded in one request), so its content is"
+ + " copied forward to append to it.");
+ try {
+ // Downloading through this stream's own write path chunks the content into blocks exactly
+ // like appended bytes, and leaves the tail buffered for the appended data to follow.
+ blob.downloadStream(this);
+ } catch (RuntimeException e) {
+ failed = true;
+ throw new IOException("Unable to read the current content of " + blob.getBlobName(), e);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ if (failed) {
+ // A staging call already failed and reported itself. Staged blocks are not part of the blob,
+ // so it keeps its original content and there is nothing to commit.
+ return;
+ }
+ stageBufferedBlock();
+ if (!staged) {
+ // Nothing was written, so the blob must stay exactly as it was.
+ return;
+ }
+ try {
+ blob.commitBlockListWithResponse(
+ new BlockBlobCommitBlockListOptions(blockIds)
+ .setHeaders(headers)
+ .setMetadata(metadata)
+ .setRequestConditions(new BlobRequestConditions().setIfMatch(eTag)),
+ null,
+ null);
+ } catch (RuntimeException e) {
+ throw new IOException("Unable to commit the appended blocks to " + blob.getBlobName(), e);
+ }
+ }
+}
diff --git a/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3/vfs/S3FileObject.java b/plugins/tech/azure/src/test/java/org/apache/hop/vfs/azure/AzureFileProviderCapabilitiesTest.java
similarity index 50%
copy from plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3/vfs/S3FileObject.java
copy to plugins/tech/azure/src/test/java/org/apache/hop/vfs/azure/AzureFileProviderCapabilitiesTest.java
index 38f7354f93..594d5b3a25 100644
--- a/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3/vfs/S3FileObject.java
+++ b/plugins/tech/azure/src/test/java/org/apache/hop/vfs/azure/AzureFileProviderCapabilitiesTest.java
@@ -13,30 +13,28 @@
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
+ *
*/
-package org.apache.hop.vfs.s3.s3.vfs;
-
-import java.io.OutputStream;
-import org.apache.commons.vfs2.FileSystemException;
-import org.apache.commons.vfs2.provider.AbstractFileName;
-import org.apache.hop.vfs.s3.s3common.S3CommonFileObject;
-import org.apache.hop.vfs.s3.s3common.S3CommonPipedOutputStream;
+package org.apache.hop.vfs.azure;
-public class S3FileObject extends S3CommonFileObject {
+import static org.junit.jupiter.api.Assertions.assertTrue;
- public S3FileObject(final AbstractFileName name, final S3FileSystem fileSystem) {
- super(name, fileSystem);
- }
+import org.apache.commons.vfs2.Capability;
+import org.junit.jupiter.api.Test;
- @Override
- public void doDelete() throws FileSystemException {
- super.doDelete();
- }
+class AzureFileProviderCapabilitiesTest {
- @Override
- public OutputStream doGetOutputStream(boolean bAppend) throws Exception {
- return new S3CommonPipedOutputStream(
- this.fileSystem, bucketName, key, ((S3FileSystem) this.fileSystem).getPartSize());
+ /**
+ * commons-vfs2 rejects {@code getOutputStream(true)} with "does not support append mode" unless
+ * the file system advertises {@link Capability#APPEND_CONTENT}. Data Lake Gen2 appends natively
+ * (see {@link DataLakeAppendOutputStream}), so the capability must be present or every append is
+ * blocked before reaching the provider.
+ */
+ @Test
+ void appendContentIsAdvertised() {
+ assertTrue(
+ AzureFileProvider.capabilities.contains(Capability.APPEND_CONTENT),
+ "Azure VFS must advertise APPEND_CONTENT so commons-vfs2 allows append mode");
}
}
diff --git a/plugins/tech/azure/src/test/java/org/apache/hop/vfs/azure/BlockBlobAppendOutputStreamTest.java b/plugins/tech/azure/src/test/java/org/apache/hop/vfs/azure/BlockBlobAppendOutputStreamTest.java
new file mode 100644
index 0000000000..bfc6ab6db5
--- /dev/null
+++ b/plugins/tech/azure/src/test/java/org/apache/hop/vfs/azure/BlockBlobAppendOutputStreamTest.java
@@ -0,0 +1,308 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.hop.vfs.azure;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.azure.storage.blob.models.BlobHttpHeaders;
+import com.azure.storage.blob.models.BlobStorageException;
+import com.azure.storage.blob.options.BlockBlobCommitBlockListOptions;
+import com.azure.storage.blob.specialized.BlockBlobClient;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
+import org.mockito.Mockito;
+
+/**
+ * Covers the block-list append behaviour of {@link BlockBlobAppendOutputStream}: the blob's
+ * existing committed blocks are kept as-is and the appended bytes are staged as extra blocks, so a
+ * single commit yields existing + appended without re-uploading the existing content.
+ *
+ * <p>Appending has to go through the Blob API because Hop writes these files as block blobs; the
+ * Data Lake {@code append} operation rejects them with {@code 409 InvalidAppendOperation}.
+ */
+class BlockBlobAppendOutputStreamTest {
+
+ private static final String BLOB = "folder/data.txt";
+ private static final String ETAG = "\"0x8DB000000000000\"";
+
+ /** Block IDs as the SDK's own block blob writer produces them: base64 of a UUID string. */
+ private static final List<String> EXISTING_BLOCKS =
+ List.of(
+ Base64.getEncoder()
+ .encodeToString(
+ "11111111-1111-1111-1111-111111111111".getBytes(StandardCharsets.UTF_8)),
+ Base64.getEncoder()
+ .encodeToString(
+ "22222222-2222-2222-2222-222222222222".getBytes(StandardCharsets.UTF_8)));
+
+ private static final BlobHttpHeaders HEADERS = new BlobHttpHeaders().setContentType("text/csv");
+ private static final Map<String, String> METADATA = Map.of("origin", "hop");
+
+ private BlockBlobClient blob;
+
+ /** Payload of every stageBlock call, in order. */
+ private List<byte[]> stagedContent;
+
+ private List<String> stagedIds;
+
+ @BeforeEach
+ void setUp() {
+ blob = mock(BlockBlobClient.class);
+ when(blob.getBlobName()).thenReturn(BLOB);
+ stagedContent = new ArrayList<>();
+ stagedIds = new ArrayList<>();
+ Mockito.doAnswer(
+ invocation -> {
+ stagedIds.add(invocation.getArgument(0));
+ stagedContent.add(
+ readFully(invocation.getArgument(1), invocation.getArgument(2, Long.class)));
+ return null;
+ })
+ .when(blob)
+ .stageBlock(anyString(), any(InputStream.class), anyLong());
+ }
+
+ private static byte[] readFully(InputStream in, long length) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ for (long i = 0; i < length; i++) {
+ out.write(in.read());
+ }
+ return out.toByteArray();
+ }
+
+ private BlockBlobAppendOutputStream openAppend(List<String> committed, int blockSize) {
+ return new BlockBlobAppendOutputStream(blob, committed, HEADERS, METADATA, ETAG, blockSize);
+ }
+
+ private BlockBlobCommitBlockListOptions capturedCommit() {
+ ArgumentCaptor<BlockBlobCommitBlockListOptions> commit =
+ ArgumentCaptor.forClass(BlockBlobCommitBlockListOptions.class);
+ verify(blob).commitBlockListWithResponse(commit.capture(), isNull(), isNull());
+ return commit.getValue();
+ }
+
+ @Test
+ void existingBlocksAreKeptAndTheAppendedBytesFollowThem() throws Exception {
+ BlockBlobAppendOutputStream out = openAppend(EXISTING_BLOCKS, 1024);
+ byte[] payload = "appended".getBytes(StandardCharsets.UTF_8);
+ out.write(payload);
+ out.close();
+
+ // Only the appended bytes are uploaded; the existing content is referenced by block ID.
+ assertEquals(1, stagedContent.size());
+ assertArrayEquals(payload, stagedContent.get(0));
+
+ // The committed list is the existing blocks, in order, then the newly staged one.
+ List<String> expected = new ArrayList<>(EXISTING_BLOCKS);
+ expected.addAll(stagedIds);
+ assertEquals(expected, capturedCommit().getBase64BlockIds());
+ }
+
+ @Test
+ void stagedBlockIdsMatchTheLengthOfTheExistingOnes() throws Exception {
+ BlockBlobAppendOutputStream out = openAppend(EXISTING_BLOCKS, 1024);
+ out.write("x".getBytes(StandardCharsets.UTF_8));
+ out.close();
+
+ // Azure rejects a commit whose block IDs are not all the same length, and the existing IDs came
+ // from the SDK's writer, so ours have to be generated the same way.
+ assertEquals(1, stagedIds.size());
+ assertEquals(
+ EXISTING_BLOCKS.get(0).length(),
+ stagedIds.get(0).length(),
+ "a staged block ID must be the same length as the blob's existing block IDs");
+ }
+
+ @Test
+ void headersAndMetadataAndETagGuardAreCarriedOnTheCommit() throws Exception {
+ BlockBlobAppendOutputStream out = openAppend(EXISTING_BLOCKS, 1024);
+ out.write("x".getBytes(StandardCharsets.UTF_8));
+ out.close();
+
+ BlockBlobCommitBlockListOptions commit = capturedCommit();
+ // Committing a block list resets whatever it is not given, so the blob's own headers and
+ // metadata must be replayed or an append would strip them.
+ assertEquals("text/csv", commit.getHeaders().getContentType());
+ assertEquals(METADATA, commit.getMetadata());
+ // And a concurrent modification must fail the append rather than silently drop data.
+ assertEquals(ETAG, commit.getRequestConditions().getIfMatch());
+ }
+
+ @Test
+ void writesBeyondTheBlockSizeAreStagedAsSeveralBlocks() throws Exception {
+ int blockSize = 8;
+ BlockBlobAppendOutputStream out = openAppend(EXISTING_BLOCKS, blockSize);
+ byte[] payload = "0123456789abcdefghij".getBytes(StandardCharsets.UTF_8);
+ out.write(payload);
+ out.close();
+
+ assertEquals(3, stagedContent.size(), "20 bytes at a block size of 8 must be 3 blocks");
+ ByteArrayOutputStream reassembled = new ByteArrayOutputStream();
+ for (byte[] staged : stagedContent) {
+ reassembled.write(staged);
+ }
+ assertArrayEquals(payload, reassembled.toByteArray(), "no bytes may be lost or reordered");
+
+ // One commit, listing the existing blocks followed by all three new ones in order.
+ List<String> expected = new ArrayList<>(EXISTING_BLOCKS);
+ expected.addAll(stagedIds);
+ assertEquals(expected, capturedCommit().getBase64BlockIds());
+ }
+
+ @Test
+ void aBlobWithoutABlockListHasItsContentCarriedForward() throws Exception {
+ byte[] existing = "existing;".getBytes(StandardCharsets.UTF_8);
+ Mockito.doAnswer(
+ invocation -> {
+ invocation.getArgument(0, OutputStream.class).write(existing);
+ return null;
+ })
+ .when(blob)
+ .downloadStream(any(OutputStream.class));
+
+ // A blob uploaded in a single Put Blob call has no committed blocks to build on.
+ BlockBlobAppendOutputStream out = openAppend(List.of(), 1024);
+ out.restageExistingContent();
+ out.write("appended".getBytes(StandardCharsets.UTF_8));
+ out.close();
+
+ // Its content is re-staged ahead of the appended bytes instead of being lost by the commit.
+ ByteArrayOutputStream reassembled = new ByteArrayOutputStream();
+ for (byte[] staged : stagedContent) {
+ reassembled.write(staged);
+ }
+ assertEquals("existing;appended", reassembled.toString(StandardCharsets.UTF_8));
+ assertEquals(stagedIds, capturedCommit().getBase64BlockIds());
+ }
+
+ @Test
+ void nothingIsCommittedBeforeClose() throws Exception {
+ BlockBlobAppendOutputStream out = openAppend(EXISTING_BLOCKS, 1024);
+ out.write("streaming".getBytes(StandardCharsets.UTF_8));
+
+ // Staged blocks are not part of the blob until the list is committed.
+ verify(blob, never())
+ .commitBlockListWithResponse(
+ any(BlockBlobCommitBlockListOptions.class), isNull(), isNull());
+
+ out.close();
+ verify(blob, times(1))
+ .commitBlockListWithResponse(
+ any(BlockBlobCommitBlockListOptions.class), isNull(), isNull());
+ }
+
+ @Test
+ void closingWithoutWritingLeavesTheBlobAlone() throws Exception {
+ BlockBlobAppendOutputStream out = openAppend(EXISTING_BLOCKS, 1024);
+ out.close();
+
+ verify(blob, never()).stageBlock(anyString(), any(InputStream.class), anyLong());
+ verify(blob, never())
+ .commitBlockListWithResponse(
+ any(BlockBlobCommitBlockListOptions.class), isNull(), isNull());
+ }
+
+ @Test
+ void closeIsIdempotent() throws Exception {
+ BlockBlobAppendOutputStream out = openAppend(EXISTING_BLOCKS, 1024);
+ out.write("x".getBytes(StandardCharsets.UTF_8));
+ out.close();
+ out.close();
+
+ verify(blob, times(1)).stageBlock(anyString(), any(InputStream.class), anyLong());
+ verify(blob, times(1))
+ .commitBlockListWithResponse(
+ any(BlockBlobCommitBlockListOptions.class), isNull(), isNull());
+ }
+
+ @Test
+ void aFailedCommitIsReportedAsAnIoException() throws Exception {
+ when(blob.commitBlockListWithResponse(
+ any(BlockBlobCommitBlockListOptions.class), isNull(), isNull()))
+ .thenThrow(new BlobStorageException("precondition failed", null, null));
+
+ BlockBlobAppendOutputStream out = openAppend(EXISTING_BLOCKS, 1024);
+ out.write("x".getBytes(StandardCharsets.UTF_8));
+
+ IOException thrown = assertThrows(IOException.class, out::close);
+ assertTrue(thrown.getMessage().contains(BLOB), "the error should name the blob");
+ }
+
+ @Test
+ void closingAfterAFailedStageDoesNotReportASecondError() {
+ doThrow(new BlobStorageException("boom", null, null))
+ .when(blob)
+ .stageBlock(anyString(), any(InputStream.class), anyLong());
+
+ // A block size of 4 makes this write big enough to stage mid-stream.
+ BlockBlobAppendOutputStream out = openAppend(EXISTING_BLOCKS, 4);
+ assertThrows(
+ IOException.class,
+ () -> out.write("more than four bytes".getBytes(StandardCharsets.UTF_8)));
+
+ // Streams are closed in a finally block, and that close must not bury the real error under a
+ // second failure, nor commit a partial block list over the blob.
+ assertDoesNotThrow(out::close);
+ verify(blob, never())
+ .commitBlockListWithResponse(
+ any(BlockBlobCommitBlockListOptions.class), isNull(), isNull());
+ }
+
+ @Test
+ void theBlobIsOnlyReadWhenItHasNoBlockList() throws Exception {
+ BlockBlobAppendOutputStream out = openAppend(EXISTING_BLOCKS, 1024);
+ out.write("x".getBytes(StandardCharsets.UTF_8));
+ out.close();
+
+ // The whole point of the block list is that existing content never travels to the client.
+ verify(blob, never()).downloadStream(any(OutputStream.class));
+
+ InOrder inOrder = Mockito.inOrder(blob);
+ inOrder.verify(blob).stageBlock(anyString(), any(InputStream.class), anyLong());
+ inOrder
+ .verify(blob)
+ .commitBlockListWithResponse(
+ any(BlockBlobCommitBlockListOptions.class), isNull(), isNull());
+ }
+}