Merge pull request #425 from lukecwik/backport-pr-934

Backport Apache Beam PR/934

Author: dhalperi

sdk/src/main/java/com/google/cloud/dataflow/sdk/io/TextIO.java

@@ -36,7 +36,6 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.protobuf.ByteString;
-
 import java.io.IOException;
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
@@ -47,7 +46,6 @@
 import java.nio.charset.StandardCharsets;
 import java.util.NoSuchElementException;
 import java.util.regex.Pattern;
-
 import javax.annotation.Nullable;
 
 /**
@@ -473,25 +471,25 @@ public static Bound<String> withoutValidation() {
 
     /**
      * Returns a transform for writing to text files that adds a header string to the files
-     * it writes.
+     * it writes. Note that a newline character will be added after the header.
      *
      * <p>A {@code null} value will clear any previously configured header.
      *
      * @param header the string to be added as file header
      */
-    public static Bound<String> withHeader(String header) {
+    public static Bound<String> withHeader(@Nullable String header) {
       return new Bound<>(DEFAULT_TEXT_CODER).withHeader(header);
     }
 
     /**
      * Returns a transform for writing to text files that adds a footer string to the files
-     * it writes.
+     * it writes. Note that a newline character will be added after the header.
      *
      * <p>A {@code null} value will clear any previously configured footer.
      *
      * @param footer the string to be added as file footer
      */
-    public static Bound<String> withFooter(String footer) {
+    public static Bound<String> withFooter(@Nullable String footer) {
       return new Bound<>(DEFAULT_TEXT_CODER).withFooter(footer);
     }
 
@@ -513,10 +511,10 @@ public static class Bound<T> extends PTransform<PCollection<T>, PDone> {
       private final String filenameSuffix;
 
       /** An optional header to add to each file. */
-      private final String header;
+      @Nullable private final String header;
 
       /** An optional footer to add to each file. */
-      private final String footer;
+      @Nullable private final String footer;
 
       /** The Coder to use to decode each line. */
       private final Coder<T> coder;
@@ -534,9 +532,9 @@ public static class Bound<T> extends PTransform<PCollection<T>, PDone> {
         this(null, null, "", null, null, coder, 0, DEFAULT_SHARD_TEMPLATE, true);
       }
 
-      private Bound(String name, String filenamePrefix, String filenameSuffix, String header,
-                    String footer, Coder<T> coder, int numShards, String shardTemplate,
-                    boolean validate) {
+      private Bound(String name, String filenamePrefix, String filenameSuffix,
+          @Nullable String header, @Nullable String footer, Coder<T> coder, int numShards,
+          String shardTemplate, boolean validate) {
         super(name);
         this.header = header;
         this.footer = footer;
@@ -653,30 +651,30 @@ public <X> Bound<X> withCoder(Coder<X> coder) {
 
       /**
        * Returns a transform for writing to text files that adds a header string to the files
-       * it writes.
+       * it writes. Note that a newline character will be added after the header.
        *
        * <p>A {@code null} value will clear any previously configured header.
        *
        * <p>Does not modify this object.
        *
        * @param header the string to be added as file header
        */
-      public Bound<T> withHeader(String header) {
+      public Bound<T> withHeader(@Nullable String header) {
         return new Bound<>(name, filenamePrefix, filenameSuffix, header, footer, coder, numShards,
             shardTemplate, false);
       }
 
       /**
        * Returns a transform for writing to text files that adds a footer string to the files
-       * it writes.
+       * it writes. Note that a newline character will be added after the header.
        *
        * <p>A {@code null} value will clear any previously configured footer.
        *
        * <p>Does not modify this object.
        *
        * @param footer the string to be added as file footer
        */
-      public Bound<T> withFooter(String footer) {
+      public Bound<T> withFooter(@Nullable String footer) {
         return new Bound<>(name, filenamePrefix, filenameSuffix, header, footer, coder, numShards,
             shardTemplate, false);
       }
@@ -767,10 +765,12 @@ public Coder<T> getCoder() {
         return coder;
       }
 
+      @Nullable
       public String getHeader() {
         return header;
       }
 
+      @Nullable
       public String getFooter() {
         return footer;
       }
@@ -835,7 +835,7 @@ private static void validateOutputComponent(String partialFilePattern) {
   private TextIO() {}
 
   /**
-   * A {@link FileBasedSource} which can decode records delimited by new line characters.
+   * A {@link FileBasedSource} which can decode records delimited by newline characters.
    *
    * <p>This source splits the data into records using {@code UTF-8} {@code \n}, {@code \r}, or
    * {@code \r\n} as the delimiter. This source is not strict and supports decoding the last record
@@ -883,7 +883,7 @@ public Coder<T> getDefaultOutputCoder() {
 
     /**
      * A {@link com.google.cloud.dataflow.sdk.io.FileBasedSource.FileBasedReader FileBasedReader}
-     * which can decode records delimited by new line characters.
+     * which can decode records delimited by newline characters.
      *
      * See {@link TextSource} for further details.
      */
@@ -1046,19 +1046,20 @@ private boolean tryToEnsureNumberOfBytesInBuffer(int minCapacity) throws IOExcep
   }
 
   /**
-   * A {@link FileBasedSink} for text files. Produces text files with the new line separator
+   * A {@link FileBasedSink} for text files. Produces text files with the newline separator
    * {@code '\n'} represented in {@code UTF-8} format as the record separator.
    * Each record (including the last) is terminated.
    */
   @VisibleForTesting
   static class TextSink<T> extends FileBasedSink<T> {
     private final Coder<T> coder;
-    private final String header;
-    private final String footer;
+    @Nullable private final String header;
+    @Nullable private final String footer;
 
     @VisibleForTesting
     TextSink(
-        String baseOutputFilename, String extension, String header, String footer,
+        String baseOutputFilename, String extension,
+        @Nullable String header, @Nullable String footer,
         String fileNameTemplate, Coder<T> coder) {
       super(baseOutputFilename, extension, fileNameTemplate);
       this.coder = coder;
@@ -1077,10 +1078,11 @@ public FileBasedSink.FileBasedWriteOperation<T> createWriteOperation(PipelineOpt
      */
     private static class TextWriteOperation<T> extends FileBasedWriteOperation<T> {
       private final Coder<T> coder;
-      private final String header;
-      private final String footer;
+      @Nullable private final String header;
+      @Nullable private final String footer;
 
-      private TextWriteOperation(TextSink<T> sink, Coder<T> coder, String header, String footer) {
+      private TextWriteOperation(TextSink<T> sink, Coder<T> coder,
+          @Nullable String header, @Nullable String footer) {
         super(sink);
         this.coder = coder;
         this.header = header;
@@ -1100,30 +1102,25 @@ public FileBasedWriter<T> createWriter(PipelineOptions options) throws Exception
     private static class TextWriter<T> extends FileBasedWriter<T> {
       private static final byte[] NEWLINE = "\n".getBytes(StandardCharsets.UTF_8);
       private final Coder<T> coder;
-      private final String header;
-      private final String footer;
+      @Nullable private final String header;
+      @Nullable private final String footer;
       private OutputStream out;
 
-      public TextWriter(FileBasedWriteOperation<T> writeOperation, Coder<T> coder) {
-        this(writeOperation, coder, null, null);
-      }
-
-      public TextWriter(FileBasedWriteOperation<T> writeOperation, Coder<T> coder, String header) {
-        this(writeOperation, coder, header, null);
-      }
-
-      public TextWriter(FileBasedWriteOperation<T> writeOperation, Coder<T> coder, String header,
-                        String footer) {
+      public TextWriter(FileBasedWriteOperation<T> writeOperation, Coder<T> coder,
+          @Nullable String header, @Nullable String footer) {
         super(writeOperation);
         this.header = header;
         this.footer = footer;
         this.mimeType = MimeTypes.TEXT;
         this.coder = coder;
       }
 
-      private void writeLine(String line) throws IOException {
-        if (line != null) {
-          out.write(line.getBytes(StandardCharsets.UTF_8));
+      /**
+       * Writes {@code value} followed by a newline if {@code value} is not null.
+       */
+      private void writeIfNotNull(@Nullable String value) throws IOException {
+        if (value != null) {
+          out.write(value.getBytes(StandardCharsets.UTF_8));
           out.write(NEWLINE);
         }
       }
@@ -1133,14 +1130,14 @@ protected void prepareWrite(WritableByteChannel channel) throws Exception {
         out = Channels.newOutputStream(channel);
       }
 
-     @Override
+      @Override
       protected void writeHeader() throws Exception {
-        writeLine(header);
+        writeIfNotNull(header);
       }
 
       @Override
       protected void writeFooter() throws Exception {
-        writeLine(footer);
+        writeIfNotNull(footer);
       }
 
       @Override

sdk/src/test/java/com/google/cloud/dataflow/sdk/io/TextIOTest.java

@@ -91,7 +91,6 @@
 import java.nio.file.StandardOpenOption;
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.Set;
 import java.util.zip.GZIPOutputStream;
@@ -310,7 +309,7 @@ public static <T> void assertOutputFiles(
     List<List<String>> actual = new ArrayList<>();
     for (File tmpFile : expectedFiles) {
       try (BufferedReader reader = new BufferedReader(new FileReader(tmpFile))) {
-        List<String> currentFile = Lists.newArrayList();
+        List<String> currentFile = new ArrayList<>();
         for (;;) {
           String line = reader.readLine();
           if (line == null) {
@@ -322,15 +321,15 @@ public static <T> void assertOutputFiles(
       }
     }
 
-    LinkedList<String> expectedElements = Lists.newLinkedList();
+    List<String> expectedElements = new ArrayList<>(elems.length);
     for (int i = 0; i < elems.length; i++) {
       T elem = elems[i];
       byte[] encodedElem = CoderUtils.encodeToByteArray(coder, elem);
       String line = new String(encodedElem);
       expectedElements.add(line);
     }
 
-    ArrayList<String> actualElements =
+    List<String> actualElements =
         Lists.newArrayList(
             Iterables.concat(
                 FluentIterable