Fix javadocs + license

passaro · passaro · commit 2e18baa7dd94 · 2022-11-09T16:35:03.000Z
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
@@ -1699,8 +1699,8 @@ private final class WriteOperationHelperCallbacksImpl
     @Override
     public CompletableFuture<Void> selectObjectContent(
         SelectObjectContentRequest request,
-        SelectObjectContentResponseHandler th) {
-     return s3AsyncClient.selectObjectContent(request, th);
+        SelectObjectContentResponseHandler responseHandler) {
+     return s3AsyncClient.selectObjectContent(request, responseHandler);
     }
 
     @Override
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/select/BlockingEnumeration.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/select/BlockingEnumeration.java
@@ -1,3 +1,21 @@
+/*
+ * 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.hadoop.fs.s3a.select;
 
 import java.util.Enumeration;
@@ -7,9 +25,18 @@
 
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
+
 import software.amazon.awssdk.core.async.SdkPublisher;
 import software.amazon.awssdk.core.exception.SdkException;
 
+/**
+ * Implements the {@link Enumeration} interface by subscribing to a
+ * {@link SdkPublisher} instance. The enumeration will buffer a fixed
+ * number of events received from the publisher and block until the first is
+ * consumed. Calls to {@link #hasMoreElements()} and {@link #nextElement()}
+ * can also block while waiting for new events.
+ * @param <T> the type of element.
+ */
 public final class BlockingEnumeration<T> implements Enumeration<T> {
   private static final class Token<T> {
     public T element;
@@ -31,6 +58,11 @@ public Token(Throwable error) {
   private final BlockingQueue<Token<T>> blockingQueue;
   private Token<T> current = null;
 
+  /**
+   * Create an enumeration with a fixed buffer size.
+   * @param publisher the publisher feeding the enumeration.
+   * @param limit the buffer size.
+   */
   public BlockingEnumeration(SdkPublisher<T> publisher, final int limit) {
     this.blockingQueue = new LinkedBlockingQueue<>(limit);
     publisher.subscribe(new Subscriber<T>() {
@@ -68,6 +100,13 @@ private void enqueue(Token<T> token) {
     });
   }
 
+  /**
+   * Create an enumeration with a fixed buffer size and an
+   * injected first element.
+   * @param publisher the publisher feeding the enumeration.
+   * @param limit the buffer size.
+   * @param firstElement first element the enumeration will return.
+   */
   public BlockingEnumeration(SdkPublisher<T> publisher,
       final int limit,
       T firstElement) {
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/select/SelectEventStreamPublisher.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/select/SelectEventStreamPublisher.java
@@ -1,18 +1,30 @@
+/*
+ * 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.hadoop.fs.s3a.select;
 
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
 import java.io.SequenceInputStream;
-import java.util.Enumeration;
-import java.util.Iterator;
-import java.util.NoSuchElementException;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
 
 import org.reactivestreams.Subscriber;
-import org.reactivestreams.Subscription;
 
 import software.amazon.awssdk.core.async.SdkPublisher;
 import software.amazon.awssdk.http.AbortableInputStream;
@@ -23,26 +35,67 @@
 import software.amazon.awssdk.utils.ToString;
 import software.amazon.awssdk.utils.Validate;
 
+/**
+ * Async publisher of {@link SelectObjectContentEventStream}s returned
+ * from a SelectObjectContent call.
+ */
 public final class SelectEventStreamPublisher implements
     SdkPublisher<SelectObjectContentEventStream> {
 
-  private final CompletableFuture<Void> future;
+  private final CompletableFuture<Void> selectOperationFuture;
   private final SelectObjectContentResponse response;
   private final SdkPublisher<SelectObjectContentEventStream> publisher;
 
+  /**
+   * Create the publisher.
+   * @param selectOperationFuture SelectObjectContent future
+   * @param response SelectObjectContent response
+   * @param publisher SelectObjectContentEventStream publisher to wrap
+   */
   public SelectEventStreamPublisher(
-      CompletableFuture<Void> future,
+      CompletableFuture<Void> selectOperationFuture,
       SelectObjectContentResponse response,
       SdkPublisher<SelectObjectContentEventStream> publisher) {
-    this.future = Validate.paramNotNull(future, "future");
-    this.response = Validate.paramNotNull(response, "response");
-    this.publisher = Validate.paramNotNull(publisher, "publisher");
+    this.selectOperationFuture = selectOperationFuture;
+    this.response = response;
+    this.publisher = publisher;
   }
 
-  public void cancel() {
-    future.cancel(true);
+  /**
+   * Retrieve an input stream to the subset of the S3 object that matched the select query.
+   * This is equivalent to loading the content of all RecordsEvents into an InputStream.
+   * This will lazily-load the content from S3, minimizing the amount of memory used.
+   * @param onEndEvent callback on the end event
+   * @return the input stream
+   */
+  public AbortableInputStream toRecordsInputStream(Consumer<EndEvent> onEndEvent) {
+    SdkPublisher<InputStream> recordInputStreams = this.publisher
+        .filter(e -> {
+          if (e instanceof RecordsEvent) {
+            return true;
+          } else if (e instanceof EndEvent) {
+            onEndEvent.accept((EndEvent) e);
+          }
+          return false;
+        })
+        .map(e -> ((RecordsEvent) e).payload().asInputStream());
+
+    // Subscribe to the async publisher using an enumeration that will
+    // buffer a single chunk (RecordsEvent's payload) at a time and
+    // block until it is consumed.
+    // Also inject an empty stream as the first element that
+    // SequenceInputStream will request on construction.
+    BlockingEnumeration enumeration =
+        new BlockingEnumeration(recordInputStreams, 1, EMPTY_STREAM);
+    return AbortableInputStream.create(
+        new SequenceInputStream(enumeration),
+        this::cancel);
   }
 
+  /**
+   * The response from the SelectObjectContent call.
+   * @return the response object
+   */
   public SelectObjectContentResponse response() {
     return response;
   }
@@ -52,6 +105,13 @@ public void subscribe(Subscriber<? super SelectObjectContentEventStream> subscri
     publisher.subscribe(subscriber);
   }
 
+  /**
+   * Cancel the operation.
+   */
+  public void cancel() {
+    selectOperationFuture.cancel(true);
+  }
+
   @Override
   public String toString() {
     return ToString.builder("SelectObjectContentEventStream")
@@ -60,25 +120,6 @@ public String toString() {
         .build();
   }
 
-  public AbortableInputStream toRecordsInputStream(Consumer<EndEvent> onEndEvent) {
-    SdkPublisher<InputStream> recordInputStreams = this.publisher
-        .filter(e -> {
-          if (e instanceof RecordsEvent) {
-            return true;
-          } else if (e instanceof EndEvent) {
-            onEndEvent.accept((EndEvent) e);
-          }
-          return false;
-        })
-        .map(e -> ((RecordsEvent) e).payload().asInputStream());
-
-    return AbortableInputStream.create(
-        new SequenceInputStream(
-            new BlockingEnumeration(recordInputStreams, 1,
-                EMPTY_STREAM)),
-        this::cancel);
-  }
-
   private static final InputStream EMPTY_STREAM =
       new ByteArrayInputStream(new byte[0]);
 }
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/select/SelectInputStream.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/select/SelectInputStream.java
@@ -20,7 +20,6 @@
 
 import java.io.EOFException;
 import java.io.IOException;
-import java.io.InputStream;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -43,9 +42,6 @@
 
 import software.amazon.awssdk.core.exception.AbortedException;
 import software.amazon.awssdk.http.AbortableInputStream;
-import software.amazon.awssdk.services.s3.model.RecordsEvent;
-import software.amazon.awssdk.services.s3.model.SelectObjectContentEventStream.EventType;
-import software.amazon.awssdk.services.s3.model.SelectObjectContentResponseHandler;
 
 import static org.apache.hadoop.util.Preconditions.checkNotNull;
 import static org.apache.commons.lang3.StringUtils.isNotEmpty;
@@ -115,7 +111,7 @@ public class SelectInputStream extends FSInputStream implements
    * The read attempt is initiated immediately.
    * @param readContext read context
    * @param objectAttributes object attributes from a HEAD request
-   * @param selectResponse response from the already executed call
+   * @param selectPublisher event stream publisher from the already executed call
    * @throws IOException failure
    */
   @Retries.OnceTranslated
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/select/SelectObjectContentHelper.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/select/SelectObjectContentHelper.java
@@ -1,3 +1,21 @@
+/*
+ * 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.hadoop.fs.s3a.select;
 
 import java.io.IOException;
@@ -17,8 +35,20 @@
 
 import static org.apache.hadoop.fs.s3a.WriteOperationHelper.WriteOperationHelperCallbacks;
 
+/**
+ * Helper for SelectObjectContent queries against an S3 Bucket.
+ */
 public final class SelectObjectContentHelper {
-
+  
+  /**
+   * Execute an S3 Select operation.
+   * @param writeOperationHelperCallbacks helper callbacks
+   * @param source  source for selection
+   * @param request Select request to issue.
+   * @param action  the action for use in exception creation
+   * @return the select response event stream publisher
+   * @throws IOException on failure
+   */
   public static SelectEventStreamPublisher select(
       WriteOperationHelperCallbacks writeOperationHelperCallbacks,
       Path source,
@@ -27,9 +57,9 @@ public static SelectEventStreamPublisher select(
       throws IOException {
     try {
       Handler handler = new Handler();
-      CompletableFuture<Void> operationFuture =
+      CompletableFuture<Void> selectOperationFuture =
           writeOperationHelperCallbacks.selectObjectContent(request, handler);
-      return handler.eventPublisher(operationFuture).join();
+      return handler.eventPublisher(selectOperationFuture).join();
     } catch (Throwable e) {
       if (e instanceof CompletionException) {
         e = e.getCause();
@@ -47,15 +77,15 @@ public static SelectEventStreamPublisher select(
 
   private static class Handler implements SelectObjectContentResponseHandler {
     private volatile CompletableFuture<Pair<SelectObjectContentResponse,
-        SdkPublisher<SelectObjectContentEventStream>>> future =
+        SdkPublisher<SelectObjectContentEventStream>>> responseAndPublisherFuture =
         new CompletableFuture<>();
 
     private volatile SelectObjectContentResponse response;
 
     public CompletableFuture<SelectEventStreamPublisher> eventPublisher(
-        CompletableFuture<Void> operationFuture) {
-      return future.thenApply(p ->
-          new SelectEventStreamPublisher(operationFuture,
+        CompletableFuture<Void> selectOperationFuture) {
+      return responseAndPublisherFuture.thenApply(p ->
+          new SelectEventStreamPublisher(selectOperationFuture,
               p.getLeft(), p.getRight()));
     }
 
@@ -66,12 +96,12 @@ public void responseReceived(SelectObjectContentResponse response) {
 
     @Override
     public void onEventStream(SdkPublisher<SelectObjectContentEventStream> publisher) {
-      future.complete(Pair.of(response, publisher));
+      responseAndPublisherFuture.complete(Pair.of(response, publisher));
     }
 
     @Override
     public void exceptionOccurred(Throwable error) {
-      future.completeExceptionally(error);
+      responseAndPublisherFuture.completeExceptionally(error);
     }
 
     @Override
