Introduce StorageConnector for GCS

extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleStorage.java

@@ -20,10 +20,13 @@
 package org.apache.druid.storage.google;
 
 import com.google.api.client.http.AbstractInputStreamContent;
+import com.google.api.client.http.HttpHeaders;
+import com.google.api.client.http.HttpResponse;
 import com.google.api.services.storage.Storage;
 import com.google.api.services.storage.Storage.Objects.Get;
 import com.google.api.services.storage.model.StorageObject;
 import com.google.common.base.Supplier;
+import org.apache.druid.java.util.common.StringUtils;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -36,7 +39,7 @@ public class GoogleStorage
    * if we have a Storage instead of a supplier of it, it can cause unnecessary config validation
    * against Google storage even when it's not used at all. To perform the config validation
    * only when it is actually used, we use a supplier.
-   *
+   * <p>
    * See OmniDataSegmentKiller for how DataSegmentKillers are initialized.
    */
   private final Supplier<Storage> storage;
@@ -59,6 +62,26 @@ public InputStream get(final String bucket, final String path) throws IOExceptio
     return get(bucket, path, 0);
   }
 
+  public InputStream getUsingRangeHeaders(final String bucket, final String path, long start, long end)
+      throws IOException
+  {
+    final Get get = storage.get().objects().get(bucket, path);
+    HttpHeaders httpHeaders = new HttpHeaders();
+
+    String rangeString = StringUtils.format("bytes %d-%d/*", start, end);
+
+    httpHeaders.setRange(rangeString);
+    get.setRequestHeaders(httpHeaders);
+
+    HttpResponse httpResponse = get.executeUsingHead();
+    String responseRangeString = httpResponse.getHeaders().getRange();
+    if (!responseRangeString.equals(rangeString)) {
+      throw new IOException("Unable to fetch the correct range");
+    }
+
+    return get.executeMediaAsInputStream();
+  }
+
   public InputStream get(final String bucket, final String path, long start) throws IOException
   {
     final Get get = storage.get().objects().get(bucket, path);
@@ -86,7 +109,7 @@ public boolean exists(final String bucket, final String path)
       return false;
     }
   }
-   
+
   public long size(final String bucket, final String path) throws IOException
   {
     return storage.get().objects().get(bucket, path).execute().getSize().longValue();

extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleUtils.java

@@ -45,7 +45,7 @@ public static boolean isRetryable(Throwable t)
     return t instanceof IOException;
   }
 
-  static <T> T retryGoogleCloudStorageOperation(RetryUtils.Task<T> f) throws Exception
+  public static <T> T retryGoogleCloudStorageOperation(RetryUtils.Task<T> f) throws Exception
   {
     return RetryUtils.retry(f, GOOGLE_RETRY, RetryUtils.DEFAULT_MAX_TRIES);
   }

extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/output/GoogleInputRange.java

@@ -0,0 +1,81 @@
+/*
+ * 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.google.output;
+
+import java.util.Objects;
+
+public class GoogleInputRange
+{
+  private final long start;
+  private final long size;
+  private final String bucket;
+  private final String path;
+
+  public GoogleInputRange(long start, long size, String bucket, String path)
+  {
+    this.start = start;
+    this.size = size;
+    this.bucket = bucket;
+    this.path = path;
+  }
+
+  public long getStart()
+  {
+    return start;
+  }
+
+  public long getSize()
+  {
+    return size;
+  }
+
+  public String getBucket()
+  {
+    return bucket;
+  }
+
+  public String getPath()
+  {
+    return path;
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    GoogleInputRange that = (GoogleInputRange) o;
+    return start == that.start
+           && size == that.size
+           && Objects.equals(bucket, that.bucket)
+           && Objects.equals(path, that.path);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(start, size, bucket, path);
+  }
+}
+

extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/output/GoogleOutputConfig.java

@@ -0,0 +1,129 @@
+/*
+ * 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.google.output;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+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.util.Objects;
+
+public class GoogleOutputConfig
+{
+
+  @JsonProperty
+  private final String bucket;
+
+  @JsonProperty
+  private final String prefix;
+
+  @JsonProperty
+  private final File tempDir;
+
+  @JsonProperty
+  private boolean chunkedDownloads = false;
+
+  @JsonProperty
+  private HumanReadableBytes chunkSize = new HumanReadableBytes("100MiB");
+
+  @JsonProperty
+  private int maxRetry = RetryUtils.DEFAULT_MAX_TRIES;
+
+  public GoogleOutputConfig(
+      final String bucket,
+      final String prefix,
+      final File tempDir,
+      @Nullable final Boolean chunkedDownloads,
+      @Nullable final HumanReadableBytes chunkSize,
+      @Nullable final Integer maxRetry
+  )
+  {
+    this.bucket = bucket;
+    this.prefix = prefix;
+    this.tempDir = tempDir;
+    if (chunkedDownloads != null) {
+      this.chunkedDownloads = chunkedDownloads;
+    }
+    if (chunkSize != null) {
+      this.chunkSize = chunkSize;
+    }
+    if (maxRetry != null) {
+      this.maxRetry = maxRetry;
+    }
+  }
+
+  public String getBucket()
+  {
+    return bucket;
+  }
+
+  public String getPrefix()
+  {
+    return prefix;
+  }
+
+  public File getTempDir()
+  {
+    return tempDir;
+  }
+
+  public Boolean isChunkedDownloads()
+  {
+    return chunkedDownloads;
+  }
+
+  public HumanReadableBytes getChunkSize()
+  {
+    return chunkSize;
+  }
+
+  public Integer getMaxRetry()
+  {
+    return maxRetry;
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    GoogleOutputConfig that = (GoogleOutputConfig) o;
+    return Objects.equals(bucket, that.bucket)
+           && Objects.equals(prefix, that.prefix)
+           && Objects.equals(tempDir, that.tempDir)
+           && Objects.equals(chunkedDownloads, that.chunkedDownloads)
+           && Objects.equals(chunkSize, that.chunkSize)
+           && Objects.equals(maxRetry, that.maxRetry);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(bucket, prefix, tempDir, chunkedDownloads, chunkSize, maxRetry);
+  }
+
+
+}