HADOOP-19343. Add support for hflush()

Closes #7761

Co-authored-by: Chris Nauroth <cnauroth@apache.org>
Signed-off-by: Chris Nauroth <cnauroth@apache.org>

Author: arunkumarchacko

hadoop-tools/hadoop-gcp/src/main/java/org/apache/hadoop/fs/gs/CreateOptions.java renamed to hadoop-tools/hadoop-gcp/src/main/java/org/apache/hadoop/fs/gs/CreateFileOptions.java

@@ -28,24 +28,38 @@
 /**
  * Options that can be specified when creating a file in the {@link GoogleCloudStorageFileSystem}.
  */
-final class CreateOptions {
+final class CreateFileOptions {
   private final ImmutableMap<String, byte[]> attributes;
   private final String contentType;
   private final long overwriteGenerationId;
   private final WriteMode mode;
+  private final boolean ensureNoDirectoryConflict;
 
-  private CreateOptions(CreateOperationOptionsBuilder builder) {
+  private CreateFileOptions(CreateOperationOptionsBuilder builder) {
     this.attributes = ImmutableMap.copyOf(builder.attributes);
     this.contentType = builder.contentType;
     this.overwriteGenerationId = builder.overwriteGenerationId;
     this.mode = builder.writeMode;
+    this.ensureNoDirectoryConflict = builder.ensureNoDirectoryConflict;
   }
 
   boolean isOverwriteExisting() {
     return this.mode == WriteMode.OVERWRITE;
   }
 
+  boolean isEnsureNoDirectoryConflict() {
+    return ensureNoDirectoryConflict;
+  }
+
+  CreateOperationOptionsBuilder toBuilder() {
+    return builder().setWriteMode(this.mode)
+        .setEnsureNoDirectoryConflict(ensureNoDirectoryConflict);
+  }
+
   enum WriteMode {
+    /** Write new bytes to the end of the existing file rather than the beginning. */
+    APPEND,
+
     /**
      * Creates a new file for write and fails if file already exists.
      */
@@ -98,14 +112,20 @@ static class CreateOperationOptionsBuilder {
     private String contentType = "application/octet-stream";
     private long overwriteGenerationId = StorageResourceId.UNKNOWN_GENERATION_ID;
     private WriteMode writeMode = WriteMode.CREATE_NEW;
+    private boolean ensureNoDirectoryConflict = true;
 
     CreateOperationOptionsBuilder setWriteMode(WriteMode mode) {
       this.writeMode = mode;
       return this;
     }
 
-    CreateOptions build() {
-      CreateOptions options = new CreateOptions(this);
+    CreateOperationOptionsBuilder setEnsureNoDirectoryConflict(boolean ensure) {
+      this.ensureNoDirectoryConflict = ensure;
+      return this;
+    }
+
+    CreateFileOptions build() {
+      CreateFileOptions options = new CreateFileOptions(this);
 
       checkArgument(!options.getAttributes().containsKey("Content-Type"),
           "The Content-Type attribute must be set via the contentType option");

hadoop-tools/hadoop-gcp/src/main/java/org/apache/hadoop/fs/gs/ErrorTypeExtractor.java

@@ -20,6 +20,8 @@
 
 import javax.annotation.Nullable;
 
+import com.google.api.client.http.HttpStatusCodes;
+import com.google.cloud.storage.StorageException;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 
@@ -63,8 +65,6 @@ enum ErrorType {
     UNAVAILABLE, UNKNOWN
   }
 
-  //  public static final ErrorTypeExtractor INSTANCE = new ErrorTypeExtractor();
-
   private static final String BUCKET_ALREADY_EXISTS_MESSAGE =
       "FAILED_PRECONDITION: Your previous request to create the named bucket succeeded and you "
           + "already own it.";
@@ -89,7 +89,28 @@ static ErrorType getErrorType(Exception error) {
     case UNAVAILABLE:
       return ErrorType.UNAVAILABLE;
     default:
-      return ErrorType.UNKNOWN;
+      return getErrorTypeFromStorageException(error);
     }
   }
+
+  private static ErrorType getErrorTypeFromStorageException(Exception error) {
+    if (error instanceof StorageException) {
+      StorageException se = (StorageException) error;
+      int httpCode = se.getCode();
+
+      if (httpCode == HttpStatusCodes.STATUS_CODE_PRECONDITION_FAILED) {
+        return ErrorType.FAILED_PRECONDITION;
+      }
+
+      if (httpCode == HttpStatusCodes.STATUS_CODE_NOT_FOUND) {
+        return ErrorType.NOT_FOUND;
+      }
+
+      if (httpCode == HttpStatusCodes.STATUS_CODE_SERVICE_UNAVAILABLE) {
+        return ErrorType.UNAVAILABLE;
+      }
+    }
+
+    return ErrorType.UNKNOWN;
+  }
 }

hadoop-tools/hadoop-gcp/src/main/java/org/apache/hadoop/fs/gs/GoogleCloudStorage.java

@@ -48,6 +48,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  * A wrapper around <a href="https://github.com/googleapis/java-storage">Google cloud storage
@@ -89,7 +90,7 @@ private static Storage createStorage(String projectId) {
     return StorageOptions.newBuilder().build().getService();
   }
 
-  WritableByteChannel create(final StorageResourceId resourceId, final CreateOptions options)
+  WritableByteChannel create(final StorageResourceId resourceId, final CreateFileOptions options)
       throws IOException {
     LOG.trace("create({})", resourceId);
 
@@ -402,6 +403,47 @@ void createEmptyObject(StorageResourceId resourceId, CreateObjectOptions options
     }
   }
 
+
+  public GoogleCloudStorageItemInfo composeObjects(
+      List<StorageResourceId> sources, StorageResourceId destination, CreateObjectOptions options)
+      throws IOException {
+    LOG.trace("composeObjects({}, {}, {})", sources, destination, options);
+    for (StorageResourceId inputId : sources) {
+      if (!destination.getBucketName().equals(inputId.getBucketName())) {
+        throw new IOException(
+            String.format(
+                "Bucket doesn't match for source '%s' and destination '%s'!",
+                inputId, destination));
+      }
+    }
+    Storage.ComposeRequest request =
+        Storage.ComposeRequest.newBuilder()
+            .addSource(
+                sources.stream().map(StorageResourceId::getObjectName).collect(Collectors.toList()))
+            .setTarget(
+                BlobInfo.newBuilder(destination.getBucketName(), destination.getObjectName())
+                    .setContentType(options.getContentType())
+                    .setContentEncoding(options.getContentEncoding())
+                    .setMetadata(encodeMetadata(options.getMetadata()))
+                    .build())
+            .setTargetOptions(
+                Storage.BlobTargetOption.generationMatch(
+                    destination.hasGenerationId()
+                        ? destination.getGenerationId()
+                        : getWriteGeneration(destination, true)))
+            .build();
+
+    Blob composedBlob;
+    try {
+      composedBlob = storage.compose(request);
+    } catch (StorageException e) {
+      throw new IOException(e);
+    }
+    GoogleCloudStorageItemInfo compositeInfo = createItemInfoForBlob(destination, composedBlob);
+    LOG.trace("composeObjects() done, returning: {}", compositeInfo);
+    return compositeInfo;
+  }
+
   /**
    * Helper to check whether an empty object already exists with the expected metadata specified in
    * {@code options}, to be used to determine whether it's safe to ignore an exception that was
@@ -450,6 +492,7 @@ private boolean canIgnoreExceptionForEmptyObject(
         return true;
       }
     }
+
     return false;
   }
 
@@ -472,18 +515,14 @@ private void createEmptyObjectInternal(
       blobTargetOptions.add(Storage.BlobTargetOption.doesNotExist());
     }
 
-    try {
-      // TODO: Set encryption key and related properties
-      storage.create(
-          BlobInfo.newBuilder(BlobId.of(resourceId.getBucketName(), resourceId.getObjectName()))
-              .setMetadata(rewrittenMetadata)
-              .setContentEncoding(createObjectOptions.getContentEncoding())
-              .setContentType(createObjectOptions.getContentType())
-              .build(),
-          blobTargetOptions.toArray(new Storage.BlobTargetOption[0]));
-    } catch (StorageException e) {
-      throw new IOException(String.format("Creating empty object %s failed.", resourceId), e);
-    }
+    // TODO: Set encryption key and related properties
+    storage.create(
+        BlobInfo.newBuilder(BlobId.of(resourceId.getBucketName(), resourceId.getObjectName()))
+            .setMetadata(rewrittenMetadata)
+            .setContentEncoding(createObjectOptions.getContentEncoding())
+            .setContentType(createObjectOptions.getContentType())
+            .build(),
+        blobTargetOptions.toArray(new Storage.BlobTargetOption[0]));
   }
 
   private static Map<String, String> encodeMetadata(Map<String, byte[]> metadata) {
@@ -871,6 +910,23 @@ private static GoogleCloudStorageItemInfo getGoogleCloudStorageItemInfo(
     return storageItemInfo;
   }
 
+  List<GoogleCloudStorageItemInfo> getItemInfos(List<StorageResourceId> resourceIds)
+      throws IOException {
+    LOG.trace("getItemInfos({})", resourceIds);
+
+    if (resourceIds.isEmpty()) {
+      return new ArrayList<>();
+    }
+
+    List<GoogleCloudStorageItemInfo> result = new ArrayList<>(resourceIds.size());
+    for (StorageResourceId resourceId : resourceIds) {
+      // TODO: Do this concurrently
+      result.add(getItemInfo(resourceId));
+    }
+
+    return result;
+  }
+
   // Helper class to capture the results of list operation.
   private class ListOperationResult {
     private final Map<String, Blob> prefixes = new HashMap<>();

hadoop-tools/hadoop-gcp/src/main/java/org/apache/hadoop/fs/gs/GoogleCloudStorageClientWriteChannel.java

@@ -44,8 +44,12 @@ class GoogleCloudStorageClientWriteChannel implements WritableByteChannel {
   private final StorageResourceId resourceId;
   private WritableByteChannel writableByteChannel;
 
-  GoogleCloudStorageClientWriteChannel(final Storage storage,
-      final StorageResourceId resourceId, final CreateOptions createOptions) throws IOException {
+  private GoogleCloudStorageItemInfo completedItemInfo = null;
+
+  GoogleCloudStorageClientWriteChannel(
+      final Storage storage,
+      final StorageResourceId resourceId,
+      final CreateFileOptions createOptions) throws IOException {
     this.resourceId = resourceId;
     BlobWriteSession blobWriteSession = getBlobWriteSession(storage, resourceId, createOptions);
     try {
@@ -56,7 +60,7 @@ class GoogleCloudStorageClientWriteChannel implements WritableByteChannel {
   }
 
   private static BlobInfo getBlobInfo(final StorageResourceId resourceId,
-      final CreateOptions createOptions) {
+      final CreateFileOptions createOptions) {
     BlobInfo blobInfo = BlobInfo.newBuilder(
             BlobId.of(resourceId.getBucketName(), resourceId.getObjectName(),
                 resourceId.getGenerationId())).setContentType(createOptions.getContentType())
@@ -66,12 +70,12 @@ private static BlobInfo getBlobInfo(final StorageResourceId resourceId,
   }
 
   private static BlobWriteSession getBlobWriteSession(final Storage storage,
-      final StorageResourceId resourceId, final CreateOptions createOptions) {
+      final StorageResourceId resourceId, final CreateFileOptions createOptions) {
     return storage.blobWriteSession(getBlobInfo(resourceId, createOptions),
         generateWriteOptions(createOptions));
   }
 
-  private static BlobWriteOption[] generateWriteOptions(final CreateOptions createOptions) {
+  private static BlobWriteOption[] generateWriteOptions(final CreateFileOptions createOptions) {
     List<BlobWriteOption> blobWriteOptions = new ArrayList<>();
 
     blobWriteOptions.add(BlobWriteOption.disableGzipContent());