From 375d2f94c774bc7b4e1d2f168c32351e2abb8d60 Mon Sep 17 00:00:00 2001 From: Gauri Prasad Date: Mon, 30 Dec 2019 10:39:25 -0800 Subject: [PATCH 1/9] Initial work to port over to using buffered upload --- sdk/storage/azure-storage-blob/pom.xml | 7 ++ .../blob/specialized/BlobOutputStream.java | 84 +++++++++++-------- .../storage/blob/BlobOutputStreamTest.groovy | 29 +++++++ .../storage/common/StorageOutputStream.java | 4 +- 4 files changed, 88 insertions(+), 36 deletions(-) diff --git a/sdk/storage/azure-storage-blob/pom.xml b/sdk/storage/azure-storage-blob/pom.xml index ee71483c1407..b1c73012b10c 100644 --- a/sdk/storage/azure-storage-blob/pom.xml +++ b/sdk/storage/azure-storage-blob/pom.xml @@ -126,6 +126,13 @@ 3.2.7 test + + + org.springframework + spring-core + 5.2.2.RELEASE + test + diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java index c5caea22918a..e0ee2c8f5dde 100644 --- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java +++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java @@ -3,11 +3,17 @@ package com.azure.storage.blob.specialized; import com.azure.core.util.logging.ClientLogger; +import com.azure.storage.blob.BlobAsyncClient; +import com.azure.storage.blob.BlobClientBuilder; +import com.azure.storage.blob.implementation.util.ModelHelper; import com.azure.storage.blob.models.AppendBlobRequestConditions; import com.azure.storage.blob.models.BlobRequestConditions; import com.azure.storage.blob.models.BlobStorageException; +import com.azure.storage.blob.models.CpkInfo; +import com.azure.storage.blob.models.CustomerProvidedKey; import com.azure.storage.blob.models.PageBlobRequestConditions; import com.azure.storage.blob.models.PageRange; +import com.azure.storage.blob.models.ParallelTransferOptions; import com.azure.storage.common.StorageOutputStream; import com.azure.storage.common.implementation.Constants; import reactor.core.publisher.Flux; @@ -52,7 +58,8 @@ static BlobOutputStream pageBlobOutputStream(final PageBlobAsyncClient client, f * Closes this output stream and releases any system resources associated with this stream. If any data remains in * the buffer it is committed to the service. * - * @throws IOException If an I/O error occurs. + * @throws IOException + * If an I/O error occurs. */ @Override public synchronized void close() throws IOException { @@ -134,39 +141,35 @@ void commit() { } private static final class BlockBlobOutputStream extends BlobOutputStream { + private final BlobRequestConditions requestConditions; - private final String blockIdPrefix; - private final List blockList; - private final BlockBlobAsyncClient client; + private final BlobAsyncClient client; private BlockBlobOutputStream(final BlockBlobAsyncClient client, final BlobRequestConditions requestConditions) { - super(BlockBlobClient.MAX_STAGE_BLOCK_BYTES); - this.client = client; + super(BlockBlobClient.MAX_UPLOAD_BLOB_BYTES); + this.client = prepareBuilder(client).buildAsyncClient(); this.requestConditions = (requestConditions == null) ? new BlobRequestConditions() : requestConditions; - this.blockIdPrefix = UUID.randomUUID().toString() + '-'; - this.blockList = new ArrayList<>(); } - /** - * Generates a new block ID to be used for PutBlock. - * - * @return Base64 encoded block ID - */ - private String getCurrentBlockId() { - String blockIdSuffix = String.format("%06d", this.blockList.size()); - return Base64.getEncoder().encodeToString((this.blockIdPrefix + blockIdSuffix) - .getBytes(StandardCharsets.UTF_8)); + @Override + synchronized void commit() { + // BlockBlob is now based on buffered upload, no need to commit. } - private Mono writeBlock(Flux blockData, String blockId, long writeLength) { - return client.stageBlockWithResponse(blockId, blockData, writeLength, null, - this.requestConditions.getLeaseId()) - .then() - .onErrorResume(BlobStorageException.class, e -> { - this.lastError = new IOException(e); - return Mono.empty(); - }); + private BlobClientBuilder prepareBuilder(BlobAsyncClientBase client) { + BlobClientBuilder builder = new BlobClientBuilder() + .pipeline(client.getHttpPipeline()) + .endpoint(client.getBlobUrl()) + .snapshot(client.getSnapshotId()) + .serviceVersion(client.getServiceVersion()); + + CpkInfo cpk = client.getCustomerProvidedKey(); + if (cpk != null) { + builder.customerProvidedKey(new CustomerProvidedKey(cpk.getEncryptionKey())); + } + + return builder; } @Override @@ -175,21 +178,34 @@ protected Mono dispatchWrite(byte[] data, int writeLength, long offset) { return Mono.empty(); } - final String blockID = this.getCurrentBlockId(); - this.blockList.add(blockID); - Flux fbb = Flux.range(0, 1) .concatMap(pos -> Mono.fromCallable(() -> ByteBuffer.wrap(data, (int) offset, writeLength))); - return this.writeBlock(fbb.subscribeOn(Schedulers.elastic()), blockID, writeLength); + ParallelTransferOptions parallelTransferOptions = ModelHelper.populateAndApplyDefaults(null); + + return this.writeBlock(fbb.subscribeOn(Schedulers.elastic()), parallelTransferOptions); + } + + private Mono writeBlock(Flux data, ParallelTransferOptions parallelTransferOptions) { + return client.uploadWithResponse(data, parallelTransferOptions, null, null, null, requestConditions) + .then() + .onErrorResume(BlobStorageException.class, e -> { + this.lastError = new IOException(e); + return Mono.empty(); + }); } - /** - * Commits the blob, for block blob this uploads the block list. - */ @Override - synchronized void commit() { - client.commitBlockListWithResponse(this.blockList, null, null, null, this.requestConditions).block(); + protected void writeInternal(final byte[] data, int offset, int length) { + dispatchWrite(data, length, offset) + .doOnError(t -> { + if (t instanceof IOException) { + lastError = (IOException) t; + } else { + lastError = new IOException(t); + } + }) + .block(); } } diff --git a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy index d42f857fee31..d55cddeeb237 100644 --- a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy +++ b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy @@ -4,8 +4,13 @@ import com.azure.storage.blob.models.BlobErrorCode import com.azure.storage.blob.models.BlobStorageException import com.azure.storage.blob.models.PageRange import com.azure.storage.common.implementation.Constants +import org.springframework.util.StreamUtils import spock.lang.Requires +import java.nio.file.Files +import java.nio.file.Path +import java.nio.file.Paths + class BlobOutputStreamTest extends APISpec { private static int FOUR_MB = 4 * Constants.MB @@ -133,4 +138,28 @@ class BlobOutputStreamTest extends APISpec { return outputStream.toByteArray() } + + @Requires({ liveMode() }) + def "BlockBlob large file"() { + setup: + Path filePath = Paths.get("C:/Users/gapra/temptestfile1g.txt") + InputStream is = Files.newInputStream(filePath) + + when: + OutputStream bos = cc.getBlobClient(generateBlobName()).getBlockBlobClient().getBlobOutputStream(null) + + def startTime = System.nanoTime() +// + StreamUtils.copy(is, bos) + +// cc.getBlobClient(generateBlobName()).uploadFromFile(filePath.toString()) + + def endTime = System.nanoTime(); + + def duration = (endTime - startTime); + + then: + System.out.println("Done. " + duration) + + } } diff --git a/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/StorageOutputStream.java b/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/StorageOutputStream.java index cdeb0a129e81..78a7c6e0bf8b 100644 --- a/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/StorageOutputStream.java +++ b/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/StorageOutputStream.java @@ -42,7 +42,7 @@ protected StorageOutputStream(final int writeThreshold) { * @param offset An int which represents the start offset in the data. * @param length An int which represents the number of bytes to write. */ - private void writeInternal(final byte[] data, int offset, int length) { + protected void writeInternal(final byte[] data, int offset, int length) { int chunks = (int) (Math.ceil((double) length / (double) this.writeThreshold)); Flux.range(0, chunks).map(c -> offset + c * this.writeThreshold) .concatMap(pos -> processChunk(data, pos, offset, length)) @@ -125,7 +125,7 @@ public void write(@NonNull final byte[] data, final int offset, final int length *

* true is acceptable for you. * - * @param byteVal An int which represents the bye value to write. + * @param byteVal An int which represents the byte value to write. */ @Override public void write(final int byteVal) { From d3364a133bed0d7b87a755769e60bff82d66a1c9 Mon Sep 17 00:00:00 2001 From: Gauri Prasad Date: Thu, 2 Jan 2020 09:48:22 -0800 Subject: [PATCH 2/9] Uses FluxSink to use buffered upload --- sdk/storage/azure-storage-blob/pom.xml | 7 -- .../blob/specialized/BlobOutputStream.java | 92 +++++++++---------- .../blob/specialized/BlockBlobClient.java | 25 ++++- .../storage/blob/BlobOutputStreamTest.groovy | 29 ------ 4 files changed, 69 insertions(+), 84 deletions(-) diff --git a/sdk/storage/azure-storage-blob/pom.xml b/sdk/storage/azure-storage-blob/pom.xml index b1c73012b10c..ee71483c1407 100644 --- a/sdk/storage/azure-storage-blob/pom.xml +++ b/sdk/storage/azure-storage-blob/pom.xml @@ -126,13 +126,6 @@ 3.2.7 test - - - org.springframework - spring-core - 5.2.2.RELEASE - test - diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java index e0ee2c8f5dde..2088cc39e70a 100644 --- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java +++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java @@ -2,13 +2,16 @@ // Licensed under the MIT License. package com.azure.storage.blob.specialized; +import com.azure.core.http.rest.Response; import com.azure.core.util.logging.ClientLogger; import com.azure.storage.blob.BlobAsyncClient; import com.azure.storage.blob.BlobClientBuilder; -import com.azure.storage.blob.implementation.util.ModelHelper; +import com.azure.storage.blob.models.AccessTier; import com.azure.storage.blob.models.AppendBlobRequestConditions; +import com.azure.storage.blob.models.BlobHttpHeaders; import com.azure.storage.blob.models.BlobRequestConditions; import com.azure.storage.blob.models.BlobStorageException; +import com.azure.storage.blob.models.BlockBlobItem; import com.azure.storage.blob.models.CpkInfo; import com.azure.storage.blob.models.CustomerProvidedKey; import com.azure.storage.blob.models.PageBlobRequestConditions; @@ -17,16 +20,13 @@ import com.azure.storage.common.StorageOutputStream; import com.azure.storage.common.implementation.Constants; import reactor.core.publisher.Flux; +import reactor.core.publisher.FluxSink; import reactor.core.publisher.Mono; import reactor.core.scheduler.Schedulers; import java.io.IOException; import java.nio.ByteBuffer; -import java.nio.charset.StandardCharsets; -import java.util.ArrayList; -import java.util.Base64; -import java.util.List; -import java.util.UUID; +import java.util.Map; /** * BlobOutputStream allows for the uploading of data to a blob using a stream-like approach. @@ -43,8 +43,9 @@ static BlobOutputStream appendBlobOutputStream(final AppendBlobAsyncClient clien } static BlobOutputStream blockBlobOutputStream(final BlockBlobAsyncClient client, - final BlobRequestConditions requestConditions) { - return new BlockBlobOutputStream(client, requestConditions); + final ParallelTransferOptions parallelTransferOptions, final BlobHttpHeaders headers, + final Map metadata, final AccessTier tier, final BlobRequestConditions requestConditions) { + return new BlockBlobOutputStream(client, parallelTransferOptions, headers, metadata, tier, requestConditions); } static BlobOutputStream pageBlobOutputStream(final PageBlobAsyncClient client, final PageRange pageRange, @@ -58,8 +59,7 @@ static BlobOutputStream pageBlobOutputStream(final PageBlobAsyncClient client, f * Closes this output stream and releases any system resources associated with this stream. If any data remains in * the buffer it is committed to the service. * - * @throws IOException - * If an I/O error occurs. + * @throws IOException If an I/O error occurs. */ @Override public synchronized void close() throws IOException { @@ -142,19 +142,28 @@ void commit() { private static final class BlockBlobOutputStream extends BlobOutputStream { - private final BlobRequestConditions requestConditions; - private final BlobAsyncClient client; + private FluxSink sink; + + boolean complete; private BlockBlobOutputStream(final BlockBlobAsyncClient client, - final BlobRequestConditions requestConditions) { - super(BlockBlobClient.MAX_UPLOAD_BLOB_BYTES); - this.client = prepareBuilder(client).buildAsyncClient(); - this.requestConditions = (requestConditions == null) ? new BlobRequestConditions() : requestConditions; - } + final ParallelTransferOptions parallelTransferOptions, final BlobHttpHeaders headers, + final Map metadata, final AccessTier tier, final BlobRequestConditions requestConditions) { + super(BlockBlobClient.MAX_STAGE_BLOCK_BYTES); - @Override - synchronized void commit() { - // BlockBlob is now based on buffered upload, no need to commit. + BlobAsyncClient blobClient = prepareBuilder(client).buildAsyncClient(); + + Flux fbb = Flux.create((FluxSink sink) -> this.sink = sink); + fbb.subscribe(); // Subscribe by upload takes too long + + blobClient.uploadWithResponse(fbb, parallelTransferOptions, + headers, + metadata, tier, requestConditions) + .doOnSuccess(s -> complete = true) + .doOnError(BlobStorageException.class, e -> { + this.lastError = new IOException(e); + }) + .subscribe(); } private BlobClientBuilder prepareBuilder(BlobAsyncClientBase client) { @@ -173,39 +182,28 @@ private BlobClientBuilder prepareBuilder(BlobAsyncClientBase client) { } @Override - protected Mono dispatchWrite(byte[] data, int writeLength, long offset) { - if (writeLength == 0) { - return Mono.empty(); + void commit() { + sink.complete(); + + while (!complete) { + try { + Thread.sleep(1000L); + } catch (InterruptedException e) { + // Does this need to be caught by logger? + e.printStackTrace(); + } } - - Flux fbb = Flux.range(0, 1) - .concatMap(pos -> Mono.fromCallable(() -> ByteBuffer.wrap(data, (int) offset, writeLength))); - - ParallelTransferOptions parallelTransferOptions = ModelHelper.populateAndApplyDefaults(null); - - return this.writeBlock(fbb.subscribeOn(Schedulers.elastic()), parallelTransferOptions); } - private Mono writeBlock(Flux data, ParallelTransferOptions parallelTransferOptions) { - return client.uploadWithResponse(data, parallelTransferOptions, null, null, null, requestConditions) - .then() - .onErrorResume(BlobStorageException.class, e -> { - this.lastError = new IOException(e); - return Mono.empty(); - }); + @Override + protected void writeInternal(final byte[] data, int offset, int length) { + sink.next(ByteBuffer.wrap(data, offset, length)); } + // Never called @Override - protected void writeInternal(final byte[] data, int offset, int length) { - dispatchWrite(data, length, offset) - .doOnError(t -> { - if (t instanceof IOException) { - lastError = (IOException) t; - } else { - lastError = new IOException(t); - } - }) - .block(); + protected Mono dispatchWrite(byte[] data, int writeLength, long offset) { + return Mono.empty(); } } diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlockBlobClient.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlockBlobClient.java index 7d8a48e6b865..08be8ec2e766 100644 --- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlockBlobClient.java +++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlockBlobClient.java @@ -18,6 +18,7 @@ import com.azure.storage.blob.models.BlockBlobItem; import com.azure.storage.blob.models.BlockList; import com.azure.storage.blob.models.BlockListType; +import com.azure.storage.blob.models.ParallelTransferOptions; import com.azure.storage.common.Utility; import com.azure.storage.common.implementation.Constants; import reactor.core.publisher.Flux; @@ -115,7 +116,29 @@ public BlobOutputStream getBlobOutputStream(boolean overwrite) { * @throws BlobStorageException If a storage service error occurred. */ public BlobOutputStream getBlobOutputStream(BlobRequestConditions requestConditions) { - return BlobOutputStream.blockBlobOutputStream(client, requestConditions); + return getBlobOutputStream(null, null, null, null, requestConditions); + } + + /** + * Creates and opens an output stream to write data to the block blob. If the blob already exists on the service, it + * will be overwritten. + *

+ * To avoid overwriting, pass "*" to {@link BlobRequestConditions#setIfNoneMatch(String)}. + * + * @param parallelTransferOptions {@link ParallelTransferOptions} used to configure buffered uploading. + * @param headers {@link BlobHttpHeaders} + * @param metadata Metadata to associate with the blob. + * @param tier {@link AccessTier} for the destination blob. + * @param requestConditions {@link BlobRequestConditions} + * + * @return A {@link BlobOutputStream} object used to write data to the blob. + * @throws BlobStorageException If a storage service error occurred. + */ + public BlobOutputStream getBlobOutputStream(ParallelTransferOptions parallelTransferOptions, + BlobHttpHeaders headers, Map metadata, AccessTier tier, + BlobRequestConditions requestConditions) { + return BlobOutputStream.blockBlobOutputStream(client, parallelTransferOptions, headers, metadata, tier, + requestConditions); } /** diff --git a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy index d55cddeeb237..d42f857fee31 100644 --- a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy +++ b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy @@ -4,13 +4,8 @@ import com.azure.storage.blob.models.BlobErrorCode import com.azure.storage.blob.models.BlobStorageException import com.azure.storage.blob.models.PageRange import com.azure.storage.common.implementation.Constants -import org.springframework.util.StreamUtils import spock.lang.Requires -import java.nio.file.Files -import java.nio.file.Path -import java.nio.file.Paths - class BlobOutputStreamTest extends APISpec { private static int FOUR_MB = 4 * Constants.MB @@ -138,28 +133,4 @@ class BlobOutputStreamTest extends APISpec { return outputStream.toByteArray() } - - @Requires({ liveMode() }) - def "BlockBlob large file"() { - setup: - Path filePath = Paths.get("C:/Users/gapra/temptestfile1g.txt") - InputStream is = Files.newInputStream(filePath) - - when: - OutputStream bos = cc.getBlobClient(generateBlobName()).getBlockBlobClient().getBlobOutputStream(null) - - def startTime = System.nanoTime() -// - StreamUtils.copy(is, bos) - -// cc.getBlobClient(generateBlobName()).uploadFromFile(filePath.toString()) - - def endTime = System.nanoTime(); - - def duration = (endTime - startTime); - - then: - System.out.println("Done. " + duration) - - } } From 1cfe526bff019ce082440e06acebaa9684c756e9 Mon Sep 17 00:00:00 2001 From: Gauri Prasad Date: Thu, 2 Jan 2020 09:50:52 -0800 Subject: [PATCH 3/9] Added complete to error case --- .../com/azure/storage/blob/specialized/BlobOutputStream.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java index 2088cc39e70a..e4d2142e000a 100644 --- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java +++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java @@ -161,6 +161,7 @@ private BlockBlobOutputStream(final BlockBlobAsyncClient client, metadata, tier, requestConditions) .doOnSuccess(s -> complete = true) .doOnError(BlobStorageException.class, e -> { + complete = true; this.lastError = new IOException(e); }) .subscribe(); @@ -185,6 +186,7 @@ private BlobClientBuilder prepareBuilder(BlobAsyncClientBase client) { void commit() { sink.complete(); + // Need to wait until the uploadTask completes while (!complete) { try { Thread.sleep(1000L); From 118a69def4741801c12f481c32be9a5c503dca59 Mon Sep 17 00:00:00 2001 From: Gauri Prasad Date: Thu, 2 Jan 2020 11:29:43 -0800 Subject: [PATCH 4/9] Updated error handling of output stream --- .../storage/blob/specialized/BlobOutputStream.java | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java index e4d2142e000a..1f842b3d8e9f 100644 --- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java +++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java @@ -156,14 +156,10 @@ private BlockBlobOutputStream(final BlockBlobAsyncClient client, Flux fbb = Flux.create((FluxSink sink) -> this.sink = sink); fbb.subscribe(); // Subscribe by upload takes too long - blobClient.uploadWithResponse(fbb, parallelTransferOptions, - headers, - metadata, tier, requestConditions) - .doOnSuccess(s -> complete = true) - .doOnError(BlobStorageException.class, e -> { - complete = true; - this.lastError = new IOException(e); - }) + blobClient.uploadWithResponse(fbb, parallelTransferOptions, headers, metadata, tier, requestConditions) + // This allows the operation to continue while maintaining the error that occurred. + .onErrorContinue(BlobStorageException.class, (e, s) -> this.lastError = new IOException(e)) + .doOnTerminate(() -> complete = true) .subscribe(); } From ecc6828af32ea40852c9d023d85bfa4c593bfdce Mon Sep 17 00:00:00 2001 From: Gauri Prasad Date: Thu, 2 Jan 2020 12:19:38 -0800 Subject: [PATCH 5/9] Updated unused imports --- .../com/azure/storage/blob/specialized/BlobOutputStream.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java index 1f842b3d8e9f..d6c4884cf30a 100644 --- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java +++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java @@ -2,7 +2,6 @@ // Licensed under the MIT License. package com.azure.storage.blob.specialized; -import com.azure.core.http.rest.Response; import com.azure.core.util.logging.ClientLogger; import com.azure.storage.blob.BlobAsyncClient; import com.azure.storage.blob.BlobClientBuilder; @@ -11,7 +10,6 @@ import com.azure.storage.blob.models.BlobHttpHeaders; import com.azure.storage.blob.models.BlobRequestConditions; import com.azure.storage.blob.models.BlobStorageException; -import com.azure.storage.blob.models.BlockBlobItem; import com.azure.storage.blob.models.CpkInfo; import com.azure.storage.blob.models.CustomerProvidedKey; import com.azure.storage.blob.models.PageBlobRequestConditions; From bf13f6512035906df2ec899a09288af8b75a62dc Mon Sep 17 00:00:00 2001 From: Gauri Prasad Date: Thu, 2 Jan 2020 13:06:03 -0800 Subject: [PATCH 6/9] Reduced poll time --- .../com/azure/storage/blob/specialized/BlobOutputStream.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java index d6c4884cf30a..8dbadb253601 100644 --- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java +++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java @@ -183,7 +183,7 @@ void commit() { // Need to wait until the uploadTask completes while (!complete) { try { - Thread.sleep(1000L); + Thread.sleep(100L); } catch (InterruptedException e) { // Does this need to be caught by logger? e.printStackTrace(); From 4bcebea1515f2cfe999fa1416f02936a5262e23b Mon Sep 17 00:00:00 2001 From: Gauri Prasad Date: Fri, 3 Jan 2020 10:32:49 -0800 Subject: [PATCH 7/9] Added change log entry and comments to the output stream class --- sdk/storage/azure-storage-blob/CHANGELOG.md | 2 ++ .../azure/storage/blob/specialized/BlobOutputStream.java | 6 +++++- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/sdk/storage/azure-storage-blob/CHANGELOG.md b/sdk/storage/azure-storage-blob/CHANGELOG.md index 0f0cfc08654e..8292a119d6f4 100644 --- a/sdk/storage/azure-storage-blob/CHANGELOG.md +++ b/sdk/storage/azure-storage-blob/CHANGELOG.md @@ -2,6 +2,8 @@ ## 12.2.0-beta.2 (Unreleased) - Added a field to ParallelTransferOptions that allows customers to configure the maximum size to upload in a single PUT. Data sizes larger than this value will be chunked and parallelized. +- Changed implementation of BlockBlobOutputStream to take advantage of the buffered upload method, improving performance. +- Added overloads to BlockBlobClient.getBlobOutputStream to allow users to provide parallel transfer options, http headers, metadata, access tier, and request conditions. ## 12.2.0-beta.1 (2019-12-17) - Added SAS generation methods on clients to improve discoverability and convenience of sas. Deprecated setContainerName, setBlobName, setSnapshotId, generateSasQueryParameters methods on BlobServiceSasSignatureValues to direct users to using the methods added on clients. diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java index 8dbadb253601..0b6ae14f500a 100644 --- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java +++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java @@ -152,7 +152,11 @@ private BlockBlobOutputStream(final BlockBlobAsyncClient client, BlobAsyncClient blobClient = prepareBuilder(client).buildAsyncClient(); Flux fbb = Flux.create((FluxSink sink) -> this.sink = sink); - fbb.subscribe(); // Subscribe by upload takes too long + + /* Subscribe by upload takes too long. We need to subscribe so that the sink is actually created. Since + this subscriber doesn't do anything and no data has started flowing, there are no drawbacks to this extra + subscribe. */ + fbb.subscribe(); blobClient.uploadWithResponse(fbb, parallelTransferOptions, headers, metadata, tier, requestConditions) // This allows the operation to continue while maintaining the error that occurred. From 5271744b000472def4801a9b1482c50a57644c30 Mon Sep 17 00:00:00 2001 From: Gauri Prasad Date: Fri, 3 Jan 2020 13:03:25 -0800 Subject: [PATCH 8/9] Updated error handling --- .../storage/blob/specialized/BlobOutputStream.java | 10 +++++++++- .../com/azure/storage/blob/BlobOutputStreamTest.groovy | 1 - 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java index 0b6ae14f500a..2ed7d6ca5d18 100644 --- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java +++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobOutputStream.java @@ -75,6 +75,11 @@ public synchronized void close() throws IOException { } catch (final BlobStorageException e) { throw new IOException(e); } + /* Need this check because for block blob the buffered upload error only manifests itself after commit is + called */ + if (this.lastError != null) { + throw lastError; + } } finally { // if close() is called again, an exception will be thrown this.lastError = new IOException(Constants.STREAM_CLOSED); @@ -160,7 +165,10 @@ private BlockBlobOutputStream(final BlockBlobAsyncClient client, blobClient.uploadWithResponse(fbb, parallelTransferOptions, headers, metadata, tier, requestConditions) // This allows the operation to continue while maintaining the error that occurred. - .onErrorContinue(BlobStorageException.class, (e, s) -> this.lastError = new IOException(e)) + .onErrorResume(BlobStorageException.class, e -> { + this.lastError = new IOException(e); + return Mono.empty(); + }) .doOnTerminate(() -> complete = true) .subscribe(); } diff --git a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy index d42f857fee31..667013540f7d 100644 --- a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy +++ b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy @@ -39,7 +39,6 @@ class BlobOutputStreamTest extends APISpec { and: blockBlobClient.getBlobOutputStream() - then: thrown(IllegalArgumentException) } From c413ef88565a085f0eb99e0e86a7eca39ccd0ab7 Mon Sep 17 00:00:00 2001 From: Gauri Prasad Date: Fri, 3 Jan 2020 14:50:26 -0800 Subject: [PATCH 9/9] Edited changelog --- sdk/storage/azure-storage-blob/CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/storage/azure-storage-blob/CHANGELOG.md b/sdk/storage/azure-storage-blob/CHANGELOG.md index b2e7fc066f9b..85bf9fb81d67 100644 --- a/sdk/storage/azure-storage-blob/CHANGELOG.md +++ b/sdk/storage/azure-storage-blob/CHANGELOG.md @@ -3,7 +3,7 @@ ## 12.2.0-beta.2 (Unreleased) - Added a field to ParallelTransferOptions that allows customers to configure the maximum size to upload in a single PUT. Data sizes larger than this value will be chunked and parallelized. - Added overloads to downloadToFile to add the option to overwrite existing files. Default behavior is to not overwrite. -- Changed implementation of BlockBlobOutputStream to take advantage of the buffered upload method, improving performance. +- Improved performance of BlockBlobOutputStream. - Added overloads to BlockBlobClient.getBlobOutputStream to allow users to provide parallel transfer options, http headers, metadata, access tier, and request conditions.