suneet-s commented on code in PR #15287:
URL: https://github.com/apache/druid/pull/15287#discussion_r1439624904
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureStorage.java:
##########
@@ -127,159 +130,137 @@ public OutputStream getBlockBlobOutputStream(
final String blobPath,
@Nullable final Integer streamWriteSizeBytes,
Integer maxAttempts
- ) throws URISyntaxException, StorageException
+ ) throws BlobStorageException
{
- CloudBlobContainer container =
getOrCreateCloudBlobContainer(containerName);
- CloudBlockBlob blockBlobReference =
container.getBlockBlobReference(blobPath);
+ BlobContainerClient blobContainerClient =
getOrCreateBlobContainerClient(containerName, maxAttempts);
+ BlockBlobClient blockBlobClient =
blobContainerClient.getBlobClient(Utility.urlEncode(blobPath)).getBlockBlobClient();
- if (blockBlobReference.exists()) {
+ if (blockBlobClient.exists()) {
throw new RE("Reference already exists");
}
-
+ BlockBlobOutputStreamOptions options = new BlockBlobOutputStreamOptions();
if (streamWriteSizeBytes != null) {
- blockBlobReference.setStreamWriteSizeInBytes(streamWriteSizeBytes);
+ options.setParallelTransferOptions(new
ParallelTransferOptions().setBlockSizeLong(streamWriteSizeBytes.longValue()));
}
-
- return blockBlobReference.openOutputStream(null,
getRequestOptionsWithRetry(maxAttempts), null);
-
+ return blockBlobClient.getBlobOutputStream(options);
}
- public CloudBlob getBlockBlobReferenceWithAttributes(final String
containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ // There's no need to download attributes with the new azure clients, they
will get fetched as needed.
+ public BlockBlobClient getBlockBlobReferenceWithAttributes(final String
containerName, final String blobPath)
+ throws BlobStorageException
{
- final CloudBlockBlob blobReference =
getOrCreateCloudBlobContainer(containerName).getBlockBlobReference(blobPath);
- blobReference.downloadAttributes();
- return blobReference;
+ return
getOrCreateBlobContainerClient(containerName).getBlobClient(Utility.urlEncode(blobPath)).getBlockBlobClient();
}
public long getBlockBlobLength(final String containerName, final String
blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return getBlockBlobReferenceWithAttributes(containerName,
blobPath).getProperties().getLength();
+ return getBlockBlobReferenceWithAttributes(containerName,
blobPath).getProperties().getBlobSize();
}
public InputStream getBlockBlobInputStream(final String containerName, final
String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(0L, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, final String
containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, null, containerName, blobPath);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final
String containerName, final String blobPath)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
return getBlockBlobInputStream(offset, length, containerName, blobPath,
null);
}
public InputStream getBlockBlobInputStream(long offset, Long length, final
String containerName, final String blobPath, Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- CloudBlobContainer container =
getOrCreateCloudBlobContainer(containerName);
- return container.getBlockBlobReference(blobPath)
- .openInputStream(offset, length, null,
getRequestOptionsWithRetry(maxAttempts), null);
+ BlobContainerClient blobContainerClient =
getOrCreateBlobContainerClient(containerName, maxAttempts);
+ return
blobContainerClient.getBlobClient(Utility.urlEncode(blobPath)).openInputStream(new
BlobInputStreamOptions().setRange(new BlobRange(offset, length)));
}
public void batchDeleteFiles(String containerName, Iterable<String> paths,
Integer maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobBatchStorageException
{
- CloudBlobContainer cloudBlobContainer =
getOrCreateCloudBlobContainer(containerName);
- BlobDeleteBatchOperation blobDeleteBatchOperation = new
BlobDeleteBatchOperation();
- for (String path : paths) {
- CloudBlob blobReference = cloudBlobContainer.getBlockBlobReference(path);
- blobDeleteBatchOperation.addSubOperation(blobReference);
- }
- cloudBlobClient.get().executeBatch(blobDeleteBatchOperation,
getRequestOptionsWithRetry(maxAttempts), null);
- }
- public List<String> listDir(final String containerName, final String
virtualDirPath)
- throws URISyntaxException, StorageException
- {
- return listDir(containerName, virtualDirPath, null);
+ BlobBatchClient blobBatchClient = new
BlobBatchClientBuilder(getOrCreateBlobContainerClient(containerName,
maxAttempts)).buildClient();
+ blobBatchClient.deleteBlobs(Lists.newArrayList(paths),
DeleteSnapshotsOptionType.ONLY);
}
public List<String> listDir(final String containerName, final String
virtualDirPath, final Integer maxAttempts)
- throws StorageException, URISyntaxException
+ throws BlobStorageException
{
List<String> files = new ArrayList<>();
- CloudBlobContainer container =
getOrCreateCloudBlobContainer(containerName);
+ BlobContainerClient blobContainerClient =
getOrCreateBlobContainerClient(containerName, maxAttempts);
- for (ListBlobItem blobItem :
- container.listBlobs(virtualDirPath, USE_FLAT_BLOB_LISTING, null,
getRequestOptionsWithRetry(maxAttempts), null)) {
- CloudBlob cloudBlob = (CloudBlob) blobItem;
- files.add(cloudBlob.getName());
- }
+ PagedIterable<BlobItem> blobItems = blobContainerClient.listBlobs(
+ new ListBlobsOptions().setPrefix(virtualDirPath),
+ Duration.ofMillis(DELTA_BACKOFF_MS)
+ );
+
+ blobItems.iterableByPage().forEach(page -> page.getElements().forEach(blob
-> files.add(blob.getName())));
return files;
}
- public boolean getBlockBlobExists(String container, String blobPath) throws
URISyntaxException, StorageException
+ public boolean getBlockBlobExists(String container, String blobPath) throws
BlobStorageException
{
return getBlockBlobExists(container, blobPath, null);
}
public boolean getBlockBlobExists(String container, String blobPath, Integer
maxAttempts)
- throws URISyntaxException, StorageException
+ throws BlobStorageException
{
- return
getOrCreateCloudBlobContainer(container).getBlockBlobReference(blobPath)
- .exists(null,
getRequestOptionsWithRetry(maxAttempts), null);
+ return getOrCreateBlobContainerClient(container,
maxAttempts).getBlobClient(Utility.urlEncode(blobPath)).exists();
}
- /**
- * If maxAttempts is provided, this method returns request options with
retry built in.
- * Retry backoff is exponential backoff, with maxAttempts set to the one
provided
- */
- @Nullable
- private BlobRequestOptions getRequestOptionsWithRetry(Integer maxAttempts)
+ @VisibleForTesting
+ BlobServiceClient getRetriableBlobServiceClient(@Nonnull Integer maxAttempts)
Review Comment:
```suggestion
BlobServiceClient getRetriableBlobServiceClient(int maxAttempts)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
