Disable support for reading "Content-Encoding: gzip" files by default GoogleCloudDataproc#228 (GoogleCloudDataproc#232)

Author: medb

cloudbuild/cloudbuild.yaml

@@ -58,8 +58,8 @@ steps:
       - 'VCS_TAG=$TAG_NAME'
       - 'CI_BUILD_ID=$BUILD_ID'
 
-# Tests take on average 17 minutes to run
-timeout: 1800s
+# Tests take on average 25 minutes to run
+timeout: 2400s
 
 options:
   machineType: 'N1_HIGHCPU_32'

gcs/CHANGES.md

@@ -64,6 +64,15 @@
 
 1.  Implement Hadoop File System `append` method using GCS compose API.
 
+1.  Disable support for reading GZIP encoded files (HTTP header
+    `Content-Encoding: gzip`) because processing of
+    [GZIP encoded](https://cloud.google.com/storage/docs/transcoding#decompressive_transcoding)
+    files is inefficient and error-prone in Hadoop and Spark.
+
+    This feature is configurable with the property:
+
+        fs.gs.inputstream.support.gzip.encoding.enable (default: false)
+
 ### 1.9.14 - 2019-02-13
 
 1.  Implement Hadoop File System `concat` method using GCS compose API.

gcs/CONFIGURATION.md

@@ -253,6 +253,15 @@ exist at the same path on all nodes
     has independently already ensured that a file exists before calling open(),
     then set this property to false for more efficient reads.
 
+*   `fs.gs.inputstream.support.gzip.encoding.enable` (default: `false`)
+
+    If set to `false` then reading files with GZIP content encoding (HTTP header
+    `Content-Encoding: gzip`) will result in failure (`IOException` is thrown).
+
+    This feature is disabled by default because processing of
+    [GZIP encoded](https://cloud.google.com/storage/docs/transcoding#decompressive_transcoding)
+    files is inefficient and error-prone in Hadoop and Spark.
+
 *   `fs.gs.generation.read.consistency` (default: `LATEST`)
 
     Determines read consistency across different generations of a Cloud Storage

gcs/conf/gcs-core-default.xml

@@ -572,6 +572,15 @@
     </description>
   </property>
 
+  <property>
+    <name>fs.gs.inputstream.support.gzip.encoding.enable</name>
+    <value>false</value>
+    <description>
+      If set to false then reading files with GZIP content encoding (HTTP header
+      "Content-Encoding: gzip") will result in failure (`IOException` is thrown).
+    </description>
+  </property>
+
   <property>
     <name>fs.gs.inputstream.inplace.seek.limit</name>
     <value>8388608</value>

gcs/src/main/java/com/google/cloud/hadoop/fs/gcs/GoogleHadoopFileSystemConfiguration.java

@@ -478,6 +478,15 @@ public class GoogleHadoopFileSystemConfiguration {
           new GoogleHadoopFileSystemConfigurationProperty<>(
               "fs.gs.inputstream.fast.fail.on.not.found.enable", true);
 
+  /**
+   * If true, reading a file with GZIP content encoding (HTTP header "Content-Encoding: gzip") will
+   * result in failure (IOException is thrown).
+   */
+  public static final GoogleHadoopFileSystemConfigurationProperty<Boolean>
+      GCS_INPUT_STREAM_SUPPORT_GZIP_ENCODING_ENABLE =
+          new GoogleHadoopFileSystemConfigurationProperty<>(
+              "fs.gs.inputstream.support.gzip.encoding.enable", false);
+
   /**
    * If forward seeks are within this many bytes of the current position, seeks are performed by
    * reading and discarding bytes in-place rather than opening a new underlying stream.
@@ -600,6 +609,8 @@ private static GoogleCloudStorageReadOptions getReadChannelOptions(Configuration
     return GoogleCloudStorageReadOptions.builder()
         .setFastFailOnNotFound(
             GCS_INPUT_STREAM_FAST_FAIL_ON_NOT_FOUND_ENABLE.get(config, config::getBoolean))
+        .setSupportGzipEncoding(
+            GCS_INPUT_STREAM_SUPPORT_GZIP_ENCODING_ENABLE.get(config, config::getBoolean))
         .setInplaceSeekLimit(GCS_INPUT_STREAM_INPLACE_SEEK_LIMIT.get(config, config::getLong))
         .setBufferSize(GCS_INPUT_STREAM_BUFFER_SIZE.get(config, config::getInt))
         .setFadvise(GCS_INPUT_STREAM_FADVISE.get(config, config::getEnum))

gcs/src/test/java/com/google/cloud/hadoop/fs/gcs/GoogleHadoopFileSystemConfigurationPropertyTest.java

@@ -88,6 +88,7 @@ public class GoogleHadoopFileSystemConfigurationPropertyTest {
           put("fs.gs.auth.client.file", null);
           put("fs.gs.inputstream.buffer.size", 0);
           put("fs.gs.inputstream.fast.fail.on.not.found.enable", true);
+          put("fs.gs.inputstream.support.gzip.encoding.enable", false);
           put("fs.gs.generation.read.consistency", GenerationReadConsistency.LATEST);
           put("fs.gs.outputstream.buffer.size", 8388608);
           put("fs.gs.outputstream.pipe.buffer.size", 1048576);

gcsio/src/main/java/com/google/cloud/hadoop/gcsio/CreateObjectOptions.java

@@ -17,21 +17,23 @@
 
 package com.google.cloud.hadoop.gcsio;
 
-import com.google.common.base.Preconditions;
+import static com.google.common.base.Preconditions.checkArgument;
+
 import com.google.common.collect.ImmutableMap;
 import java.util.Map;
 
 /**
  * Options for creating objects in GCS.
  */
 public class CreateObjectOptions {
-  public static final ImmutableMap<String, byte[]> EMPTY_METADATA =
-      ImmutableMap.<String, byte[]>of();
+  public static final ImmutableMap<String, byte[]> EMPTY_METADATA = ImmutableMap.of();
   public static final String DEFAULT_CONTENT_TYPE = "application/octet-stream";
+  public static final String DEFAULT_CONTENT_ENCODING = null;
   public static final CreateObjectOptions DEFAULT = new CreateObjectOptions(true);
 
   private final boolean overwriteExisting;
   private final String contentType;
+  private final String contentEncoding;
   private final Map<String, byte[]> metadata;
   private final boolean requireMetadataMatchForEmptyObjects;
 
@@ -43,7 +45,7 @@ public class CreateObjectOptions {
    * @param overwriteExisting True to overwrite any existing objects with the same name.
    */
   public CreateObjectOptions(boolean overwriteExisting) {
-    this(overwriteExisting, DEFAULT_CONTENT_TYPE, EMPTY_METADATA, false);
+    this(overwriteExisting, DEFAULT_CONTENT_TYPE, DEFAULT_CONTENT_ENCODING, EMPTY_METADATA, false);
   }
 
   /**
@@ -56,7 +58,7 @@ public CreateObjectOptions(boolean overwriteExisting) {
    * @param metadata A dictionary of metadata to apply to created objects.
    */
   public CreateObjectOptions(boolean overwriteExisting, Map<String, byte[]> metadata) {
-    this(overwriteExisting, DEFAULT_CONTENT_TYPE, metadata, true);
+    this(overwriteExisting, DEFAULT_CONTENT_TYPE, metadata);
   }
 
   /**
@@ -70,30 +72,38 @@ public CreateObjectOptions(boolean overwriteExisting, Map<String, byte[]> metada
    */
   public CreateObjectOptions(
       boolean overwriteExisting, String contentType, Map<String, byte[]> metadata) {
-    this(overwriteExisting, contentType, metadata, true);
+    this(overwriteExisting, contentType, DEFAULT_CONTENT_ENCODING, metadata, true);
   }
 
   /**
    * Construct a new CreateObjectOptions with the spec metadata and content-type.
    *
    * @param overwriteExisting True to overwrite any existing objects with the same name
    * @param contentType content-type for the created file
+   * @param contentEncoding content-encoding for the created file
    * @param metadata A dictionary of metadata to apply to created objects
-   * @param requireMetadataMatchForEmptyObjects if true, when creating an empty object and
-   *     certain types of errors occur, any existing object is checked for an exact metadata
-   *     match to the metadata in this CreateObjectOptions before accepting the creation as
-   *     successful. If false, then on error for creating empty objects, as long as an
-   *     appropriate empty object already exists, even if it holds different metadata than
-   *     provided in this CreateObjectOptions instance, it may be considered created
-   *     successfully.
+   * @param requireMetadataMatchForEmptyObjects if true, when creating an empty object and certain
+   *     types of errors occur, any existing object is checked for an exact metadata match to the
+   *     metadata in this CreateObjectOptions before accepting the creation as successful. If false,
+   *     then on error for creating empty objects, as long as an appropriate empty object already
+   *     exists, even if it holds different metadata than provided in this CreateObjectOptions
+   *     instance, it may be considered created successfully.
    */
   public CreateObjectOptions(
-      boolean overwriteExisting, String contentType, Map<String, byte[]> metadata,
+      boolean overwriteExisting,
+      String contentType,
+      String contentEncoding,
+      Map<String, byte[]> metadata,
       boolean requireMetadataMatchForEmptyObjects) {
-    Preconditions.checkArgument(!metadata.containsKey("Content-Type"),
-        "The Content-Type metadata must be provided explicitly via the 'contentType' parameter");
+    checkArgument(
+        !metadata.containsKey("Content-Type"),
+        "The Content-Type must be provided explicitly via the 'contentType' parameter");
+    checkArgument(
+        !metadata.containsKey("Content-Encoding"),
+        "The Content-Encoding must be provided explicitly via the 'contentEncoding' parameter");
     this.overwriteExisting = overwriteExisting;
     this.contentType = contentType;
+    this.contentEncoding = contentEncoding;
     this.metadata = metadata;
     this.requireMetadataMatchForEmptyObjects = requireMetadataMatchForEmptyObjects;
   }
@@ -112,6 +122,11 @@ public String getContentType() {
     return contentType;
   }
 
+  /** Content type to set when creating a file. */
+  public String getContentEncoding() {
+    return contentEncoding;
+  }
+
   /**
    * Custom metadata to apply to this object.
    */

gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageImpl.java

@@ -376,10 +376,12 @@ public WritableByteChannel create(final StorageResourceId resourceId, CreateObje
             clientRequestHelper,
             resourceId.getBucketName(),
             resourceId.getObjectName(),
+            options.getContentType(),
+            options.getContentEncoding(),
+            /* kmsKeyName= */ null,
             storageOptions.getWriteChannelOptions(),
             writeConditions,
-            rewrittenMetadata,
-            options.getContentType()) {
+            rewrittenMetadata) {
 
           @Override
           public Storage.Objects.Insert createRequest(InputStreamContent inputStream)
@@ -1120,10 +1122,12 @@ public List<GoogleCloudStorageItemInfo> listBucketInfo()
    */
   private Storage.Objects.Insert prepareEmptyInsert(
       StorageResourceId resourceId, CreateObjectOptions createObjectOptions) throws IOException {
-    StorageObject object = new StorageObject();
-    object.setName(resourceId.getObjectName());
     Map<String, String> rewrittenMetadata = encodeMetadata(createObjectOptions.getMetadata());
-    object.setMetadata(rewrittenMetadata);
+    StorageObject object =
+        new StorageObject()
+            .setName(resourceId.getObjectName())
+            .setMetadata(rewrittenMetadata)
+            .setContentEncoding(createObjectOptions.getContentEncoding());
 
     // Ideally we'd use EmptyContent, but Storage requires an AbstractInputStreamContent and not
     // just an HttpContent, so we'll just use the next easiest thing.
@@ -2004,6 +2008,7 @@ public GoogleCloudStorageItemInfo composeObjects(
                         .setDestination(
                             new StorageObject()
                                 .setContentType(options.getContentType())
+                                .setContentEncoding(options.getContentEncoding())
                                 .setMetadata(encodeMetadata(options.getMetadata())))),
             destination.getBucketName());
 

gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannel.java

@@ -809,11 +809,11 @@ protected void initMetadata(
         !metadataInitialized,
         "can not initialize metadata, it already initialized for '%s'", resourceIdString);
     gzipEncoded = nullToEmpty(encoding).contains(GZIP_ENCODING);
-    if (gzipEncoded) {
-      size = Long.MAX_VALUE;
-    } else {
-      size = sizeFromMetadata;
+    if (gzipEncoded && !readOptions.getSupportGzipEncoding()) {
+      throw new IOException(
+          "Can't read GZIP encoded files - content encoding support is disabled.");
     }
+    size = gzipEncoded ? Long.MAX_VALUE : sizeFromMetadata;
     randomAccess = !gzipEncoded && readOptions.getFadvise() == Fadvise.RANDOM;
     checkEncodingAndAccess();
 

gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadOptions.java

@@ -28,7 +28,7 @@
 public abstract class GoogleCloudStorageReadOptions {
 
   /** Operational modes of fadvise feature. */
-  public static enum Fadvise {
+  public enum Fadvise {
     AUTO,
     RANDOM,
     SEQUENTIAL
@@ -47,6 +47,7 @@ public enum GenerationReadConsistency {
   public static final int DEFAULT_BACKOFF_MAX_INTERVAL_MILLIS = 10 * 1000;
   public static final int DEFAULT_BACKOFF_MAX_ELAPSED_TIME_MILLIS = 2 * 60 * 1000;
   public static final boolean DEFAULT_FAST_FAIL_ON_NOT_FOUND = true;
+  public static final boolean DEFAULT_SUPPORT_GZIP_ENCODING = true;
   public static final int DEFAULT_BUFFER_SIZE = 0;
   public static final long DEFAULT_INPLACE_SEEK_LIMIT = 0L;
   public static final Fadvise DEFAULT_FADVISE = Fadvise.SEQUENTIAL;
@@ -66,6 +67,7 @@ public static Builder builder() {
         .setBackoffMaxIntervalMillis(DEFAULT_BACKOFF_MAX_INTERVAL_MILLIS)
         .setBackoffMaxElapsedTimeMillis(DEFAULT_BACKOFF_MAX_ELAPSED_TIME_MILLIS)
         .setFastFailOnNotFound(DEFAULT_FAST_FAIL_ON_NOT_FOUND)
+        .setSupportGzipEncoding(DEFAULT_SUPPORT_GZIP_ENCODING)
         .setBufferSize(DEFAULT_BUFFER_SIZE)
         .setInplaceSeekLimit(DEFAULT_INPLACE_SEEK_LIMIT)
         .setFadvise(DEFAULT_FADVISE)
@@ -91,6 +93,9 @@ public static Builder builder() {
   /** See {@link Builder#setFastFailOnNotFound}. */
   public abstract boolean getFastFailOnNotFound();
 
+  /** See {@link Builder#setSupportGzipEncoding}. */
+  public abstract boolean getSupportGzipEncoding();
+
   /** See {@link Builder#setBufferSize}. */
   public abstract int getBufferSize();
 
@@ -153,6 +158,13 @@ public abstract static class Builder {
      */
     public abstract Builder setFastFailOnNotFound(boolean fastFailOnNotFound);
 
+    /**
+     * If false then reading a file with GZIP content encoding (HTTP header "Content-Encoding:
+     * gzip") will result in failure (IOException is thrown). If true then GZIP-encoded files will
+     * be read successfully.
+     */
+    public abstract Builder setSupportGzipEncoding(boolean supportGzipEncoding);
+
     /**
      * If set to a positive value, low-level streams will be wrapped inside a BufferedInputStream of
      * this size. Otherwise no buffer will be created to wrap the low-level streams. Note that the

gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageWriteChannel.java

@@ -38,6 +38,7 @@ public class GoogleCloudStorageWriteChannel
   private final Storage gcs;
   private final String bucketName;
   private final String objectName;
+  private final String contentEncoding;
   private final String kmsKeyName;
   private final ObjectWriteConditions writeConditions;
   private final Map<String, String> metadata;
@@ -127,13 +128,53 @@ public GoogleCloudStorageWriteChannel(
    * @param writeConditions conditions on which write should be allowed to continue
    * @param objectMetadata metadata to apply to the newly created object
    */
+  @Deprecated
+  public GoogleCloudStorageWriteChannel(
+      ExecutorService uploadThreadPool,
+      Storage gcs,
+      ClientRequestHelper<StorageObject> requestHelper,
+      String bucketName,
+      String objectName,
+      String contentType,
+      String kmsKeyName,
+      AsyncWriteChannelOptions options,
+      ObjectWriteConditions writeConditions,
+      Map<String, String> objectMetadata) {
+    this(
+        uploadThreadPool,
+        gcs,
+        requestHelper,
+        bucketName,
+        objectName,
+        contentType,
+        /* contentEncoding= */ null,
+        kmsKeyName,
+        options,
+        writeConditions,
+        objectMetadata);
+  }
+  /**
+   * Constructs an instance of GoogleCloudStorageWriteChannel.
+   *
+   * @param uploadThreadPool thread pool to use for running the upload operation
+   * @param gcs storage object instance
+   * @param requestHelper a ClientRequestHelper to set extra headers
+   * @param bucketName name of the bucket to create object in
+   * @param objectName name of the object to create
+   * @param contentType content type
+   * @param contentEncoding content encoding
+   * @param kmsKeyName Name of Cloud KMS key to use to encrypt the newly created object
+   * @param writeConditions conditions on which write should be allowed to continue
+   * @param objectMetadata metadata to apply to the newly created object
+   */
   public GoogleCloudStorageWriteChannel(
       ExecutorService uploadThreadPool,
       Storage gcs,
       ClientRequestHelper<StorageObject> requestHelper,
       String bucketName,
       String objectName,
       String contentType,
+      String contentEncoding,
       String kmsKeyName,
       AsyncWriteChannelOptions options,
       ObjectWriteConditions writeConditions,
@@ -146,6 +187,7 @@ public GoogleCloudStorageWriteChannel(
     if (contentType != null) {
       setContentType(contentType);
     }
+    this.contentEncoding = contentEncoding;
     this.kmsKeyName = kmsKeyName;
     this.writeConditions = writeConditions;
     this.metadata = objectMetadata;
@@ -156,6 +198,7 @@ public Insert createRequest(InputStreamContent inputStream) throws IOException {
     // Create object with the given name and metadata.
     StorageObject object =
         new StorageObject()
+            .setContentEncoding(contentEncoding)
             .setMetadata(metadata)
             .setName(objectName);
 

gcsio/src/main/java/com/google/cloud/hadoop/gcsio/testing/InMemoryGoogleCloudStorage.java

@@ -143,6 +143,7 @@ public synchronized WritableByteChannel create(
             resourceId.getObjectName(),
             clock.currentTimeMillis(),
             options.getContentType(),
+            options.getContentEncoding(),
             options.getMetadata());
     bucketLookup.get(resourceId.getBucketName()).add(entry);
     return entry.getWriteChannel();