You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "LakshSingla (via GitHub)" <gi...@apache.org> on 2023/07/25 16:53:55 UTC
[PR] Introduce StorageConnector for Azure (druid)
LakshSingla opened a new pull request, #14660:
URL: https://github.com/apache/druid/pull/14660
### Description
This PR adds the storage connector to interact with Azure's blob storage using the current Azure API used in Druid. This will allow Durable storage and MSQ's interactive APIs to work with Azure
This also refactors the currently available S3 connector so that the chunking downloads that is currently done by the S3 connector can be extended to other connectors. (note: This refactoring is ported from the PR https://github.com/apache/druid/pull/14611 since that is currently parked for work).
#### Testing plan
<!--
In each section, please describe design decisions made, including:
- Choice of algorithms
- Behavioral aspects. What configuration values are acceptable? How are corner cases and error conditions handled, such as when there are insufficient resources?
- Class organization and design (how the logic is split between classes, inheritance, composition, design patterns)
- Method organization and design (how the logic is split between methods, parameters and return types)
- Naming (class, method, API, configuration, HTTP endpoint, names of emitted metrics)
-->
<!-- It's good to describe an alternative design (or mention an alternative name) for every design (or naming) decision point and compare the alternatives with the designs that you've implemented (or the names you've chosen) to highlight the advantages of the chosen designs and names. -->
<!-- If there was a discussion of the design of the feature implemented in this PR elsewhere (e. g. a "Proposal" issue, any other issue, or a thread in the development mailing list), link to that discussion from this PR description and explain what have changed in your final design compared to your original proposal or the consensus version in the end of the discussion. If something hasn't changed since the original discussion, you can omit a detailed discussion of those aspects of the design here, perhaps apart from brief mentioning for the sake of readability of this PR description. -->
<!-- Some of the aspects mentioned above may be omitted for simple and small changes. -->
#### Release note
To be added
If your change doesn't have end user impact, you can skip this section.
For tips about how to write a good release note, see [Release notes](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#release-notes).
-->
<hr>
##### Key changed/added classes in this PR
<hr>
<!-- Check the items by putting "x" in the brackets for the done things. Not all of these items apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items from the checklist below are strictly necessary, but it would be very helpful if you at least self-review the PR. -->
This PR has:
- [ ] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] added documentation for new or modified features or behaviors.
- [ ] a release note entry in the PR description.
- [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [ ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [ ] been tested in a test Druid cluster.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Introduce StorageConnector for Azure (druid)
Posted by "adarshsanjeev (via GitHub)" <gi...@apache.org>.
adarshsanjeev commented on code in PR #14660:
URL: https://github.com/apache/druid/pull/14660#discussion_r1285330752
##########
processing/src/main/java/org/apache/druid/storage/remote/ChunkingStorageConnector.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * 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.druid.storage.remote;
+
+import com.google.common.base.Predicates;
+import org.apache.commons.io.input.NullInputStream;
+import org.apache.druid.data.input.impl.RetryingInputStream;
+import org.apache.druid.java.util.common.FileUtils;
+import org.apache.druid.java.util.common.IOE;
+import org.apache.druid.java.util.common.RE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.storage.StorageConnector;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.util.Enumeration;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * An abstract implementation of the storage connectors that download the file from the remote storage in chunks
+ * and presents the downloaded chunks as a single {@link InputStream} for the consumers of the connector.
+ * This implementation benefits over keeping the InputStream to the remote source open since we don't require the
+ * connection to be open for the entire duration.
+ * Checkout {@link ChunkingStorageConnectorParameters} to see the inputs required to support chunking
+ */
+public abstract class ChunkingStorageConnector<T> implements StorageConnector
+{
+ private static final long DOWNLOAD_MAX_CHUNK_SIZE_BYTES = 100_000_000;
+
+ private final long chunkSizeBytes;
+
+ public ChunkingStorageConnector()
+ {
+ this(DOWNLOAD_MAX_CHUNK_SIZE_BYTES);
+ }
+
+ public ChunkingStorageConnector(
+ final long chunkSizeBytes
+ )
+ {
+ this.chunkSizeBytes = chunkSizeBytes;
+ }
+
+ @Override
+ public InputStream read(String path) throws IOException
+ {
+ return buildInputStream(buildInputParams(path));
+ }
+
+ @Override
+ public InputStream readRange(String path, long from, long size)
+ {
+ return buildInputStream(buildInputParams(path, from, size));
+ }
+
+ public abstract ChunkingStorageConnectorParameters<T> buildInputParams(String path) throws IOException;
+
+ public abstract ChunkingStorageConnectorParameters<T> buildInputParams(String path, long from, long size);
+
+ private InputStream buildInputStream(ChunkingStorageConnectorParameters<T> params)
+ {
+ // Position from where the read needs to be resumed
+ final AtomicLong currentReadStartPosition = new AtomicLong(params.getStart());
+
+ // Final position, exclusive
+ long readEnd = params.getEnd();
+
+ AtomicBoolean isSequenceStreamClosed = new AtomicBoolean(false);
+
+ return new SequenceInputStream(
+
+ new Enumeration<InputStream>()
+ {
+ boolean initStream = false;
+
+ @Override
+ public boolean hasMoreElements()
+ {
+ // Checking if the stream was already closed. If it was, then don't iterate over the remaining chunks
+ // SequenceInputStream's close method closes all the chunk streams in its close. Since we're opening them
+ // lazily, we don't need to close them.
+ if (isSequenceStreamClosed.get()) {
+ return false;
+ }
+ // Don't stop until the whole object is downloaded
+ return currentReadStartPosition.get() < readEnd;
+ }
+
+ @Override
+ public InputStream nextElement()
+ {
+ if (!initStream) {
+ initStream = true;
+ return new NullInputStream();
+ }
+
+ File outFile = new File(
+ params.getTempDirSupplier().get().getAbsolutePath(),
+ UUID.randomUUID().toString()
+ );
+
+ long currentReadEndPosition = Math.min(
+ currentReadStartPosition.get() + chunkSizeBytes,
+ readEnd
+ );
+
+ try {
+ if (!outFile.createNewFile()) {
+ throw new IOE(
+ StringUtils.format(
+ "Could not create temporary file [%s] for copying [%s]",
+ outFile.getAbsolutePath(),
+ params.getCloudStoragePath()
+ )
+ );
+ }
+
+ FileUtils.copyLarge(
+ () -> new RetryingInputStream<>(
+ params.getObjectSupplier().getObject(currentReadStartPosition.get(), currentReadEndPosition),
+ params.getObjectOpenFunction(),
+ params.getRetryCondition(),
+ params.getMaxRetry()
+ ),
+ outFile,
+ new byte[8 * 1024],
Review Comment:
I know this code was only moved, but could you add a comment on why these numbers are chosen?
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Introduce StorageConnector for Azure (druid)
Posted by "LakshSingla (via GitHub)" <gi...@apache.org>.
LakshSingla commented on code in PR #14660:
URL: https://github.com/apache/druid/pull/14660#discussion_r1283037249
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureStorage.java:
##########
@@ -120,11 +145,44 @@
public InputStream getBlobInputStream(long offset, final String containerName, final String blobPath)
throws URISyntaxException, StorageException
+ {
+ return getBlobInputStream(offset, null, containerName, blobPath);
+ }
+
+ public InputStream getBlobInputStream(long offset, Long length, final String containerName, final String blobPath)
+ throws URISyntaxException, StorageException
{
CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- return container.getBlockBlobReference(blobPath).openInputStream(offset, null, null, null, null);
+ return container.getBlockBlobReference(blobPath).openInputStream(offset, length, null, null, null);
}
+ public void batchDeleteFiles(String containerName, Iterable<String> paths)
+ throws URISyntaxException, StorageException
+ {
+ CloudBlobContainer cloudBlobContainer = getOrCreateCloudBlobContainer(containerName);
Review Comment:
Resolved
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Introduce StorageConnector for Azure (druid)
Posted by "github-code-scanning[bot] (via GitHub)" <gi...@apache.org>.
github-code-scanning[bot] commented on code in PR #14660:
URL: https://github.com/apache/druid/pull/14660#discussion_r1274408405
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureStorage.java:
##########
@@ -120,11 +145,44 @@
public InputStream getBlobInputStream(long offset, final String containerName, final String blobPath)
throws URISyntaxException, StorageException
+ {
+ return getBlobInputStream(offset, null, containerName, blobPath);
+ }
+
+ public InputStream getBlobInputStream(long offset, Long length, final String containerName, final String blobPath)
+ throws URISyntaxException, StorageException
{
CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- return container.getBlockBlobReference(blobPath).openInputStream(offset, null, null, null, null);
+ return container.getBlockBlobReference(blobPath).openInputStream(offset, length, null, null, null);
}
+ public void batchDeleteFiles(String containerName, Iterable<String> paths)
+ throws URISyntaxException, StorageException
+ {
+ CloudBlobContainer cloudBlobContainer = getOrCreateCloudBlobContainer(containerName);
Review Comment:
## Unread local variable
Variable 'CloudBlobContainer cloudBlobContainer' is never read.
[Show more details](https://github.com/apache/druid/security/code-scanning/5294)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Introduce StorageConnector for Azure (druid)
Posted by "github-code-scanning[bot] (via GitHub)" <gi...@apache.org>.
github-code-scanning[bot] commented on code in PR #14660:
URL: https://github.com/apache/druid/pull/14660#discussion_r1287367403
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/output/AzureOutputConfig.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * 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.druid.storage.azure.output;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.java.util.common.FileUtils;
+import org.apache.druid.java.util.common.HumanReadableBytes;
+import org.apache.druid.java.util.common.RetryUtils;
+
+import javax.annotation.Nullable;
+import java.io.File;
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * Configuration of the Azure storage connector
+ */
+public class AzureOutputConfig
+{
+ @JsonProperty
+ private final String container;
+
+ @JsonProperty
+ private final String prefix;
+
+ @JsonProperty
+ private final File tempDir;
+
+ @JsonProperty
+ private final HumanReadableBytes chunkSize;
+
+ private static final HumanReadableBytes DEFAULT_CHUNK_SIZE = new HumanReadableBytes("4MiB");
+
+ // Minimum limit is self-imposed, so that chunks are appropriately sized, and we don't spend a lot of time downloading
+ // the part of the blobs
+ private static final long AZURE_MIN_CHUNK_SIZE_BYTES = new HumanReadableBytes("256KiB").getBytes();
+
+ // Maximum limit is imposed by Azure, on the size of one block blob
+ private static final long AZURE_MAX_CHUNK_SIZE_BYTES = new HumanReadableBytes("4000MiB").getBytes();
+
+
+ @JsonProperty
+ private final int maxRetry;
+
+ public AzureOutputConfig(
+ @JsonProperty(value = "container", required = true) String container,
+ @JsonProperty(value = "prefix", required = true) String prefix,
+ @JsonProperty(value = "tempDir", required = true) File tempDir,
+ @JsonProperty(value = "chunkSize") @Nullable HumanReadableBytes chunkSize,
+ @JsonProperty(value = "maxRetry") @Nullable Integer maxRetry
+ )
+ {
+ this.container = container;
+ this.prefix = prefix;
+ this.tempDir = tempDir;
+ this.chunkSize = chunkSize != null ? chunkSize : DEFAULT_CHUNK_SIZE;
+ this.maxRetry = maxRetry != null ? maxRetry : RetryUtils.DEFAULT_MAX_TRIES;
+ validateFields();
+ }
+
+
+ public String getContainer()
+ {
+ return container;
+ }
+
+ public String getPrefix()
+ {
+ return prefix;
+ }
+
+ public File getTempDir()
+ {
+ return tempDir;
+ }
+
+ public HumanReadableBytes getChunkSize()
+ {
+ return chunkSize;
+ }
+
+ public int getMaxRetry()
+ {
+ return maxRetry;
+ }
+
+ private void validateFields()
+ {
+ if (chunkSize.getBytes() < AZURE_MIN_CHUNK_SIZE_BYTES || chunkSize.getBytes() > AZURE_MAX_CHUNK_SIZE_BYTES) {
+ throw InvalidInput.exception(
+ "'chunkSize' [%d] bytes to the AzureConfig should be between [%d] bytes and [%d] bytes",
+ chunkSize.getBytes(),
+ AZURE_MIN_CHUNK_SIZE_BYTES,
+ AZURE_MAX_CHUNK_SIZE_BYTES
+ );
+ }
+
+ try {
+ FileUtils.mkdirp(tempDir);
+ }
+ catch (IOException e) {
+ throw DruidException.forPersona(DruidException.Persona.ADMIN)
+ .ofCategory(DruidException.Category.RUNTIME_FAILURE)
+ .build(e, "Unable to create temporary directory [%s]", tempDir.getAbsolutePath());
+ }
+
+ if (!tempDir.canRead() || !tempDir.canWrite()) {
+ throw DruidException.forPersona(DruidException.Persona.ADMIN)
+ .ofCategory(DruidException.Category.RUNTIME_FAILURE)
+ .build("Cannot read or write on the 'tempDir' [%s]. "
+ + "Please provide a different path to store the intermediate contents of AzureStorageConnector");
Review Comment:
## Missing format argument
This format call refers to 1 argument(s) but only supplies 0 argument(s).
[Show more details](https://github.com/apache/druid/security/code-scanning/5421)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Introduce StorageConnector for Azure (druid)
Posted by "cryptoe (via GitHub)" <gi...@apache.org>.
cryptoe commented on code in PR #14660:
URL: https://github.com/apache/druid/pull/14660#discussion_r1277126539
##########
processing/src/main/java/org/apache/druid/storage/remote/ChunkingStorageConnector.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.druid.storage.remote;
+
+import com.google.common.base.Predicates;
+import org.apache.commons.io.input.NullInputStream;
+import org.apache.druid.data.input.impl.RetryingInputStream;
+import org.apache.druid.java.util.common.FileUtils;
+import org.apache.druid.java.util.common.IOE;
+import org.apache.druid.java.util.common.RE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.storage.StorageConnector;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.util.Enumeration;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+public abstract class ChunkingStorageConnector<T> implements StorageConnector
Review Comment:
Can you please java doc this since this is the crux of this PR .
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Introduce StorageConnector for Azure (druid)
Posted by "LakshSingla (via GitHub)" <gi...@apache.org>.
LakshSingla commented on code in PR #14660:
URL: https://github.com/apache/druid/pull/14660#discussion_r1285360523
##########
processing/src/main/java/org/apache/druid/storage/remote/ChunkingStorageConnector.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.druid.storage.remote;
+
+import com.google.common.base.Predicates;
+import org.apache.commons.io.input.NullInputStream;
+import org.apache.druid.data.input.impl.RetryingInputStream;
+import org.apache.druid.java.util.common.FileUtils;
+import org.apache.druid.java.util.common.IOE;
+import org.apache.druid.java.util.common.RE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.storage.StorageConnector;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.util.Enumeration;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * An abstract implementation of the storage connectors that download the file from the remote storage in chunks
+ * and presents the downloaded chunks as a single {@link InputStream} for the consumers of the connector.
+ * This implementation benefits over keeping the InputStream to the remote source open since we don't require the
+ * connection to be open for the entire duration.
+ * Checkout {@link ChunkingStorageConnectorParameters} to see the inputs required to support chunking
+ */
+public abstract class ChunkingStorageConnector<T> implements StorageConnector
+{
+ private static final long DOWNLOAD_MAX_CHUNK_SIZE_BYTES = 100_000_000;
+
+ public ChunkingStorageConnector()
Review Comment:
Reverted the change so that the individual connectors can control the chunk sizes. Used primarily for testing for now, though this can be extended to the real implementations as well.
##########
processing/src/main/java/org/apache/druid/storage/remote/ChunkingStorageConnector.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * 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.druid.storage.remote;
+
+import com.google.common.base.Predicates;
+import org.apache.commons.io.input.NullInputStream;
+import org.apache.druid.data.input.impl.RetryingInputStream;
+import org.apache.druid.java.util.common.FileUtils;
+import org.apache.druid.java.util.common.IOE;
+import org.apache.druid.java.util.common.RE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.storage.StorageConnector;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.util.Enumeration;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * An abstract implementation of the storage connectors that download the file from the remote storage in chunks
+ * and presents the downloaded chunks as a single {@link InputStream} for the consumers of the connector.
+ * This implementation benefits over keeping the InputStream to the remote source open since we don't require the
+ * connection to be open for the entire duration.
+ * Checkout {@link ChunkingStorageConnectorParameters} to see the inputs required to support chunking
+ */
+public abstract class ChunkingStorageConnector<T> implements StorageConnector
+{
+ private static final long DOWNLOAD_MAX_CHUNK_SIZE_BYTES = 100_000_000;
+
+ private final long chunkSizeBytes;
+
+ public ChunkingStorageConnector()
+ {
+ this(DOWNLOAD_MAX_CHUNK_SIZE_BYTES);
+ }
+
+ public ChunkingStorageConnector(
+ final long chunkSizeBytes
+ )
+ {
+ this.chunkSizeBytes = chunkSizeBytes;
+ }
+
+ @Override
+ public InputStream read(String path) throws IOException
+ {
+ return buildInputStream(buildInputParams(path));
+ }
+
+ @Override
+ public InputStream readRange(String path, long from, long size)
+ {
+ return buildInputStream(buildInputParams(path, from, size));
+ }
+
+ public abstract ChunkingStorageConnectorParameters<T> buildInputParams(String path) throws IOException;
+
+ public abstract ChunkingStorageConnectorParameters<T> buildInputParams(String path, long from, long size);
+
+ private InputStream buildInputStream(ChunkingStorageConnectorParameters<T> params)
+ {
+ // Position from where the read needs to be resumed
+ final AtomicLong currentReadStartPosition = new AtomicLong(params.getStart());
+
+ // Final position, exclusive
+ long readEnd = params.getEnd();
+
+ AtomicBoolean isSequenceStreamClosed = new AtomicBoolean(false);
+
+ return new SequenceInputStream(
+
+ new Enumeration<InputStream>()
+ {
+ boolean initStream = false;
+
+ @Override
+ public boolean hasMoreElements()
+ {
+ // Checking if the stream was already closed. If it was, then don't iterate over the remaining chunks
+ // SequenceInputStream's close method closes all the chunk streams in its close. Since we're opening them
+ // lazily, we don't need to close them.
+ if (isSequenceStreamClosed.get()) {
+ return false;
+ }
+ // Don't stop until the whole object is downloaded
+ return currentReadStartPosition.get() < readEnd;
+ }
+
+ @Override
+ public InputStream nextElement()
+ {
+ if (!initStream) {
+ initStream = true;
+ return new NullInputStream();
+ }
+
+ File outFile = new File(
+ params.getTempDirSupplier().get().getAbsolutePath(),
+ UUID.randomUUID().toString()
+ );
+
+ long currentReadEndPosition = Math.min(
+ currentReadStartPosition.get() + chunkSizeBytes,
+ readEnd
+ );
+
+ try {
+ if (!outFile.createNewFile()) {
+ throw new IOE(
+ StringUtils.format(
+ "Could not create temporary file [%s] for copying [%s]",
+ outFile.getAbsolutePath(),
+ params.getCloudStoragePath()
+ )
+ );
+ }
+
+ FileUtils.copyLarge(
+ () -> new RetryingInputStream<>(
+ params.getObjectSupplier().getObject(currentReadStartPosition.get(), currentReadEndPosition),
+ params.getObjectOpenFunction(),
+ params.getRetryCondition(),
+ params.getMaxRetry()
+ ),
+ outFile,
+ new byte[8 * 1024],
Review Comment:
Sure
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Introduce StorageConnector for Azure (druid)
Posted by "github-code-scanning[bot] (via GitHub)" <gi...@apache.org>.
github-code-scanning[bot] commented on code in PR #14660:
URL: https://github.com/apache/druid/pull/14660#discussion_r1280141782
##########
extensions-core/azure-extensions/src/test/java/org/apache/druid/storage/azure/output/AzureOutputConfigTest.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * 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.druid.storage.azure.output;
+
+
+import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.HumanReadableBytes;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.IOException;
+
+public class AzureOutputConfigTest
+{
+
+ @Rule
+ public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ private static final String CONTAINER = "container";
+ private static final String PREFIX = "prefix";
+ private static final int MAX_RETRY_COUNT = 0;
+
+ @Test
+ public void testTooLargeChunkSize()
+ {
+ HumanReadableBytes chunkSize = new HumanReadableBytes("4001MiB");
+ Assert.assertThrows(
+ DruidException.class,
+ () -> new AzureOutputConfig(CONTAINER, PREFIX, temporaryFolder.newFolder(), chunkSize, MAX_RETRY_COUNT)
+ );
+ }
+
+ @Test
+ public void testTempDirectoryNotWritable() throws IOException
+ {
+ File tempDir = temporaryFolder.newFolder();
+ tempDir.setWritable(false);
Review Comment:
## Ignored error status of call
Method testTempDirectoryNotWritable ignores exceptional return value of File.setWritable.
[Show more details](https://github.com/apache/druid/security/code-scanning/5320)
##########
extensions-core/azure-extensions/src/test/java/org/apache/druid/storage/azure/output/AzureOutputSerdeTest.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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.druid.storage.azure.output;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.exc.MismatchedInputException;
+import com.fasterxml.jackson.databind.exc.ValueInstantiationException;
+import org.apache.druid.java.util.common.HumanReadableBytes;
+import org.apache.druid.java.util.common.StringUtils;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import java.io.File;
+import java.io.IOException;
+
+public class AzureOutputSerdeTest
+{
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
Review Comment:
## Deprecated method or constructor invocation
Invoking [ExpectedException.none](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/5321)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Introduce StorageConnector for Azure (druid)
Posted by "LakshSingla (via GitHub)" <gi...@apache.org>.
LakshSingla commented on PR #14660:
URL: https://github.com/apache/druid/pull/14660#issuecomment-1671228847
Thanks, @adarshsanjeev @cryptoe for the reviews and @dhananjay1308 for testing the changes out on a cluster.
Testing for Azure has been ongoing for a day. Queries for durable storage on Azure are taking comparable times to durable storage on S3, and there don't seem to be any performance concerns for the new storage connector. Going ahead with the merge.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Introduce StorageConnector for Azure (druid)
Posted by "adarshsanjeev (via GitHub)" <gi...@apache.org>.
adarshsanjeev commented on code in PR #14660:
URL: https://github.com/apache/druid/pull/14660#discussion_r1281384658
##########
processing/src/main/java/org/apache/druid/storage/remote/ChunkingStorageConnectorParameters.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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.druid.storage.remote;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Predicate;
+import org.apache.druid.data.input.impl.prefetch.ObjectOpenFunction;
+
+import java.io.File;
+import java.util.Objects;
+import java.util.function.Supplier;
+
+/**
+ * POJO for storing the parameters required to support chunking of the downloads by {@link ChunkingStorageConnector}.
+ * The implementations of the {@link ChunkingStorageConnector} should essentially provide a way to build this object,
+ * which contains the information required to support chunking.
+ * Therefore, to a call of {@link org.apache.druid.storage.StorageConnector#readRange(String, long, long)}, the
+ * implementations of the chunking storage connectors would fetch the required chunks using the information present in
+ * this POJO.
+ */
+public class ChunkingStorageConnectorParameters<T>
+{
+ /**
+ * Starting point from where to begin reading the cloud object. This is inclusive.
+ */
+ private final long start;
+
+ /**
+ * Ending point till where to end reading the cloud object. This is exclusive.
+ */
+ private final long end;
+
+ /**
+ * Absolute storage path of the cloud object.
+ */
+ private final String cloudStoragePath;
+
+ /**
+ * Given a range (start inclusive, end exclusive), fetch the object which represents the provided range of the remote
+ * object
+ */
+ private final ChunkingStorageConnector.GetObjectFromRangeFunction<T> objectSupplier;
+
+ /**
+ * Fetching function, which opens the input stream to the range provided by the given object
+ */
+ private final ObjectOpenFunction<T> objectOpenFunction;
+
+ /**
+ * Condition to initiate a retry if downloading the chunk errors out
+ */
+ private final Predicate<Throwable> retryCondition;
+
+ /**
+ * Max number of retries while reading the storage connector
+ */
+ private final int maxRetry;
+
+ /**
+ * Temporary directory where the chunks are stored
+ */
+ private final Supplier<File> tempDirSupplier;
+
+ public ChunkingStorageConnectorParameters(
+ long start,
+ long end,
+ String cloudStoragePath,
+ ChunkingStorageConnector.GetObjectFromRangeFunction<T> objectSupplier,
+ ObjectOpenFunction<T> objectOpenFunction,
+ Predicate<Throwable> retryCondition,
+ int maxRetry,
+ Supplier<File> tempDirSupplier
+ )
+ {
+ this.start = start;
+ this.end = end;
+ this.cloudStoragePath = cloudStoragePath;
+ this.objectSupplier = objectSupplier;
+ this.objectOpenFunction = objectOpenFunction;
+ this.retryCondition = retryCondition;
+ this.maxRetry = maxRetry;
+ this.tempDirSupplier = tempDirSupplier;
+ }
+
+ public long getStart()
+ {
+ return start;
+ }
+
+ public long getEnd()
+ {
+ return end;
+ }
+
+ public String getCloudStoragePath()
+ {
+ return cloudStoragePath;
+ }
+
+ public ChunkingStorageConnector.GetObjectFromRangeFunction<T> getObjectSupplier()
+ {
+ return objectSupplier;
+ }
+
+ public ObjectOpenFunction<T> getObjectOpenFunction()
+ {
+ return objectOpenFunction;
+ }
+
+ public Predicate<Throwable> getRetryCondition()
+ {
+ return retryCondition;
+ }
+
+ public int getMaxRetry()
+ {
+ return maxRetry;
+ }
+
+ public Supplier<File> getTempDirSupplier()
+ {
+ return tempDirSupplier;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ChunkingStorageConnectorParameters<?> that = (ChunkingStorageConnectorParameters<?>) o;
+ return start == that.start &&
+ end == that.end &&
+ maxRetry == that.maxRetry &&
+ Objects.equals(cloudStoragePath, that.cloudStoragePath) &&
+ Objects.equals(objectSupplier, that.objectSupplier) &&
+ Objects.equals(objectOpenFunction, that.objectOpenFunction) &&
+ Objects.equals(retryCondition, that.retryCondition) &&
+ Objects.equals(tempDirSupplier, that.tempDirSupplier);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(
+ start,
+ end,
+ cloudStoragePath,
+ objectSupplier,
+ objectOpenFunction,
+ retryCondition,
+ maxRetry,
+ tempDirSupplier
+ );
+ }
+
+ /**
+ * Builder for {@link ChunkingStorageConnectorParameters}. Performs null checks and asserts preconditions before
+ * building the instance
+ */
+ public static class Builder<T>
+ {
+ private long start;
+ private long end;
+ private String cloudStoragePath;
+ private ChunkingStorageConnector.GetObjectFromRangeFunction<T> objectSupplier;
+ private ObjectOpenFunction<T> objectOpenFunction;
+ private Predicate<Throwable> retryCondition;
+ private int maxRetry;
+ private Supplier<File> tempDirSupplier;
+
+
+ public Builder<T> start(long start)
+ {
+ this.start = start;
+ return this;
+ }
+
+ public Builder<T> end(long end)
+ {
+ this.end = end;
+ return this;
+ }
+
+ public Builder<T> cloudStoragePath(String cloudStoragePath)
+ {
+ this.cloudStoragePath = cloudStoragePath;
+ return this;
+ }
+
+ public Builder<T> objectSupplier(ChunkingStorageConnector.GetObjectFromRangeFunction<T> objectSupplier)
+ {
+ this.objectSupplier = objectSupplier;
+ return this;
+ }
+
+ public Builder<T> objectOpenFunction(ObjectOpenFunction<T> objectOpenFunction)
+ {
+ this.objectOpenFunction = objectOpenFunction;
+ return this;
+ }
+
+ public Builder<T> retryCondition(Predicate<Throwable> retryCondition)
+ {
+ this.retryCondition = retryCondition;
+ return this;
+ }
+
+ public Builder<T> maxRetry(int maxRetry)
+ {
+ this.maxRetry = maxRetry;
+ return this;
+ }
+
+ public Builder<T> tempDirSupplier(Supplier<File> tempDirSupplier)
+ {
+ this.tempDirSupplier = tempDirSupplier;
+ return this;
+ }
+
+ public ChunkingStorageConnectorParameters<T> build()
+ {
+ Preconditions.checkArgument(start >= 0, "'start' not provided or an incorrect value [%s] passed", start);
+ Preconditions.checkArgument(end >= 0, "'end' not provided or an incorrect value [%s] passed", end);
Review Comment:
Would end < start return a good error message?
##########
processing/src/main/java/org/apache/druid/storage/remote/ChunkingStorageConnector.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.druid.storage.remote;
+
+import com.google.common.base.Predicates;
+import org.apache.commons.io.input.NullInputStream;
+import org.apache.druid.data.input.impl.RetryingInputStream;
+import org.apache.druid.java.util.common.FileUtils;
+import org.apache.druid.java.util.common.IOE;
+import org.apache.druid.java.util.common.RE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.storage.StorageConnector;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.util.Enumeration;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * An abstract implementation of the storage connectors that download the file from the remote storage in chunks
+ * and presents the downloaded chunks as a single {@link InputStream} for the consumers of the connector.
+ * This implementation benefits over keeping the InputStream to the remote source open since we don't require the
+ * connection to be open for the entire duration.
+ * Checkout {@link ChunkingStorageConnectorParameters} to see the inputs required to support chunking
+ */
+public abstract class ChunkingStorageConnector<T> implements StorageConnector
+{
+ private static final long DOWNLOAD_MAX_CHUNK_SIZE_BYTES = 100_000_000;
+
+ public ChunkingStorageConnector()
Review Comment:
Does this need to be public?
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Introduce StorageConnector for Azure (druid)
Posted by "LakshSingla (via GitHub)" <gi...@apache.org>.
LakshSingla commented on code in PR #14660:
URL: https://github.com/apache/druid/pull/14660#discussion_r1289133006
##########
processing/src/main/java/org/apache/druid/storage/remote/ChunkingStorageConnectorParameters.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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.druid.storage.remote;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Predicate;
+import org.apache.druid.data.input.impl.prefetch.ObjectOpenFunction;
+
+import java.io.File;
+import java.util.Objects;
+import java.util.function.Supplier;
+
+/**
+ * POJO for storing the parameters required to support chunking of the downloads by {@link ChunkingStorageConnector}.
+ * The implementations of the {@link ChunkingStorageConnector} should essentially provide a way to build this object,
+ * which contains the information required to support chunking.
+ * Therefore, to a call of {@link org.apache.druid.storage.StorageConnector#readRange(String, long, long)}, the
+ * implementations of the chunking storage connectors would fetch the required chunks using the information present in
+ * this POJO.
+ */
+public class ChunkingStorageConnectorParameters<T>
+{
+ /**
+ * Starting point from where to begin reading the cloud object. This is inclusive.
+ */
+ private final long start;
+
+ /**
+ * Ending point till where to end reading the cloud object. This is exclusive.
+ */
+ private final long end;
+
+ /**
+ * Absolute storage path of the cloud object.
+ */
+ private final String cloudStoragePath;
+
+ /**
+ * Given a range (start inclusive, end exclusive), fetch the object which represents the provided range of the remote
+ * object
+ */
+ private final ChunkingStorageConnector.GetObjectFromRangeFunction<T> objectSupplier;
+
+ /**
+ * Fetching function, which opens the input stream to the range provided by the given object
+ */
+ private final ObjectOpenFunction<T> objectOpenFunction;
+
+ /**
+ * Condition to initiate a retry if downloading the chunk errors out
+ */
+ private final Predicate<Throwable> retryCondition;
+
+ /**
+ * Max number of retries while reading the storage connector
+ */
+ private final int maxRetry;
+
+ /**
+ * Temporary directory where the chunks are stored
+ */
+ private final Supplier<File> tempDirSupplier;
+
+ public ChunkingStorageConnectorParameters(
+ long start,
+ long end,
+ String cloudStoragePath,
+ ChunkingStorageConnector.GetObjectFromRangeFunction<T> objectSupplier,
+ ObjectOpenFunction<T> objectOpenFunction,
+ Predicate<Throwable> retryCondition,
+ int maxRetry,
+ Supplier<File> tempDirSupplier
+ )
+ {
+ this.start = start;
+ this.end = end;
+ this.cloudStoragePath = cloudStoragePath;
+ this.objectSupplier = objectSupplier;
+ this.objectOpenFunction = objectOpenFunction;
+ this.retryCondition = retryCondition;
+ this.maxRetry = maxRetry;
+ this.tempDirSupplier = tempDirSupplier;
+ }
+
+ public long getStart()
+ {
+ return start;
+ }
+
+ public long getEnd()
+ {
+ return end;
+ }
+
+ public String getCloudStoragePath()
+ {
+ return cloudStoragePath;
+ }
+
+ public ChunkingStorageConnector.GetObjectFromRangeFunction<T> getObjectSupplier()
+ {
+ return objectSupplier;
+ }
+
+ public ObjectOpenFunction<T> getObjectOpenFunction()
+ {
+ return objectOpenFunction;
+ }
+
+ public Predicate<Throwable> getRetryCondition()
+ {
+ return retryCondition;
+ }
+
+ public int getMaxRetry()
+ {
+ return maxRetry;
+ }
+
+ public Supplier<File> getTempDirSupplier()
+ {
+ return tempDirSupplier;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ChunkingStorageConnectorParameters<?> that = (ChunkingStorageConnectorParameters<?>) o;
+ return start == that.start &&
+ end == that.end &&
+ maxRetry == that.maxRetry &&
+ Objects.equals(cloudStoragePath, that.cloudStoragePath) &&
+ Objects.equals(objectSupplier, that.objectSupplier) &&
+ Objects.equals(objectOpenFunction, that.objectOpenFunction) &&
+ Objects.equals(retryCondition, that.retryCondition) &&
+ Objects.equals(tempDirSupplier, that.tempDirSupplier);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(
+ start,
+ end,
+ cloudStoragePath,
+ objectSupplier,
+ objectOpenFunction,
+ retryCondition,
+ maxRetry,
+ tempDirSupplier
+ );
+ }
+
+ /**
+ * Builder for {@link ChunkingStorageConnectorParameters}. Performs null checks and asserts preconditions before
+ * building the instance
+ */
+ public static class Builder<T>
+ {
+ private long start;
+ private long end;
+ private String cloudStoragePath;
+ private ChunkingStorageConnector.GetObjectFromRangeFunction<T> objectSupplier;
+ private ObjectOpenFunction<T> objectOpenFunction;
+ private Predicate<Throwable> retryCondition;
+ private int maxRetry;
+ private Supplier<File> tempDirSupplier;
+
+
+ public Builder<T> start(long start)
+ {
+ this.start = start;
+ return this;
+ }
+
+ public Builder<T> end(long end)
+ {
+ this.end = end;
+ return this;
+ }
+
+ public Builder<T> cloudStoragePath(String cloudStoragePath)
+ {
+ this.cloudStoragePath = cloudStoragePath;
+ return this;
+ }
+
+ public Builder<T> objectSupplier(ChunkingStorageConnector.GetObjectFromRangeFunction<T> objectSupplier)
+ {
+ this.objectSupplier = objectSupplier;
+ return this;
+ }
+
+ public Builder<T> objectOpenFunction(ObjectOpenFunction<T> objectOpenFunction)
+ {
+ this.objectOpenFunction = objectOpenFunction;
+ return this;
+ }
+
+ public Builder<T> retryCondition(Predicate<Throwable> retryCondition)
+ {
+ this.retryCondition = retryCondition;
+ return this;
+ }
+
+ public Builder<T> maxRetry(int maxRetry)
+ {
+ this.maxRetry = maxRetry;
+ return this;
+ }
+
+ public Builder<T> tempDirSupplier(Supplier<File> tempDirSupplier)
+ {
+ this.tempDirSupplier = tempDirSupplier;
+ return this;
+ }
+
+ public ChunkingStorageConnectorParameters<T> build()
+ {
+ Preconditions.checkArgument(start >= 0, "'start' not provided or an incorrect value [%s] passed", start);
+ Preconditions.checkArgument(end >= 0, "'end' not provided or an incorrect value [%s] passed", end);
Review Comment:
Updated a check with this as well in the PR!
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Introduce StorageConnector for Azure (druid)
Posted by "LakshSingla (via GitHub)" <gi...@apache.org>.
LakshSingla merged PR #14660:
URL: https://github.com/apache/druid/pull/14660
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Introduce StorageConnector for Azure (druid)
Posted by "LakshSingla (via GitHub)" <gi...@apache.org>.
LakshSingla commented on code in PR #14660:
URL: https://github.com/apache/druid/pull/14660#discussion_r1283037249
##########
extensions-core/azure-extensions/src/main/java/org/apache/druid/storage/azure/AzureStorage.java:
##########
@@ -120,11 +145,44 @@
public InputStream getBlobInputStream(long offset, final String containerName, final String blobPath)
throws URISyntaxException, StorageException
+ {
+ return getBlobInputStream(offset, null, containerName, blobPath);
+ }
+
+ public InputStream getBlobInputStream(long offset, Long length, final String containerName, final String blobPath)
+ throws URISyntaxException, StorageException
{
CloudBlobContainer container = getOrCreateCloudBlobContainer(containerName);
- return container.getBlockBlobReference(blobPath).openInputStream(offset, null, null, null, null);
+ return container.getBlockBlobReference(blobPath).openInputStream(offset, length, null, null, null);
}
+ public void batchDeleteFiles(String containerName, Iterable<String> paths)
+ throws URISyntaxException, StorageException
+ {
+ CloudBlobContainer cloudBlobContainer = getOrCreateCloudBlobContainer(containerName);
Review Comment:
Resovled
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org