/*
 * Copyright (c) 2015, Cloudera, Inc. All Rights Reserved.
 *
 * Cloudera, Inc. 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
 *
 * This software 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 com.cloudera.dataflow.hadoop;

import com.google.cloud.dataflow.sdk.transforms.PTransform;
import com.google.cloud.dataflow.sdk.util.WindowingStrategy;
import com.google.cloud.dataflow.sdk.values.KV;
import com.google.cloud.dataflow.sdk.values.PCollection;
import com.google.cloud.dataflow.sdk.values.PInput;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

public class HadoopIO {

  public static class Read {
    public static Bound from(String filepattern) {
      return new Bound().from(filepattern);
    }

    public static Bound withFormatClass(Class<? extends FileInputFormat<?, ?>> format) {
      return new Bound().withFormatClass(format);
    }

    public static Bound withKeyClass(Class<?> key) {
      return new Bound().withKeyClass(key);
    }

    public static Bound withValueClass(Class<?> value) {
      return new Bound().withValueClass(value);
    }


    public static class Bound<K, V> extends PTransform<PInput, PCollection<KV<K, V>>> {

      private final String filepattern;
      private final Class<? extends FileInputFormat<K, V>> formatClass;
      private final Class<K> keyClass;
      private final Class<V> valueClass;

      public Bound() {
        this(null, null, null, null);
      }

      public Bound(String filepattern, Class<? extends FileInputFormat<K, V>> format, Class<K> key,
          Class<V> value) {
        this.filepattern = filepattern;
        this.formatClass = format;
        this.keyClass = key;
        this.valueClass = value;
      }

      public Bound<K, V> from(String file) {
        return new Bound<>(file, formatClass, keyClass, valueClass);
      }

      public Bound<K, V> withFormatClass(Class<? extends FileInputFormat<K, V>> format) {
        return new Bound<>(filepattern, format, keyClass, valueClass);
      }

      public Bound<K, V> withKeyClass(Class<K> key) {
        return new Bound<>(filepattern, formatClass, key, valueClass);
      }

      public Bound<K, V> withValueClass(Class<V> value) {
        return new Bound<>(filepattern, formatClass, keyClass, value);
      }

      public String getFilepattern() {
        return filepattern;
      }

      public Class<? extends FileInputFormat<K, V>> getFormatClass() {
        return formatClass;
      }

      public Class<V> getValueClass() {
        return valueClass;
      }

      public Class<K> getKeyClass() {
        return keyClass;
      }

      @Override
      public PCollection<KV<K, V>> apply(PInput input) {
        if (filepattern == null) {
          throw new IllegalStateException(
              "need to set the filepattern of an HadoopIO.Read transform");
        }
        if (formatClass == null) {
          throw new IllegalStateException(
              "need to set the format class of an HadoopIO.Read transform");
        }
        if (keyClass == null) {
          throw new IllegalStateException(
              "need to set the key class of an HadoopIO.Read transform");
        }
        if (valueClass == null) {
          throw new IllegalStateException(
              "need to set the value class of an HadoopIO.Read transform");
        }

        return PCollection.createPrimitiveOutputInternal(WindowingStrategy
            .globalDefault());
      }

    }

  }
}

@@ -875,16 +875,61 @@
 PInput;%0A
+import com.google.common.base.Preconditions;%0A
 import o
@@ -990,16 +990,22 @@
 %0Apublic 
+final 
 class Ha
@@ -1014,16 +1014,44 @@
 opIO %7B%0A%0A
+  private HadoopIO() %7B%0A  %7D%0A%0A
   public
@@ -1054,24 +1054,30 @@
 blic static 
+final 
 class Read %7B
@@ -1077,16 +1077,45 @@
  Read %7B%0A
+%0A    private Read() %7B%0A    %7D%0A%0A
     publ
@@ -1570,17 +1570,16 @@
 %0A    %7D%0A%0A
-%0A
     publ
@@ -1854,31 +1854,24 @@
 ass;%0A%0A      
-public 
 Bound() %7B%0A  
@@ -1917,31 +1917,24 @@
    %7D%0A%0A      
-public 
 Bound(String
@@ -3164,80 +3164,48 @@
     
-if (filepattern == null) %7B%0A          throw new IllegalStateException(%0A  
+Preconditions.checkNotNull(filepattern,%0A
     
@@ -3286,90 +3286,48 @@
     
-%7D%0A        if (formatClass == null) %7B%0A          throw new IllegalStateException(%0A  
+Preconditions.checkNotNull(formatClass,%0A
     
@@ -3409,87 +3409,45 @@
     
-%7D%0A        if (keyClass == null) %7B%0A          throw new IllegalStateException(%0A  
+Preconditions.checkNotNull(keyClass,%0A
     
@@ -3526,88 +3526,47 @@
     
-%7D%0A        if (valueClass == null) %7B%0A          throw new IllegalStateException(%0A 
+Preconditions.checkNotNull(valueClass,%0A
     
@@ -3565,33 +3565,32 @@
 ss,%0A            
- 
 %22need to set the
@@ -3638,18 +3638,8 @@
 m%22);
-%0A        %7D
 %0A%0A  
@@ -3710,29 +3710,16 @@
 Strategy
-%0A            
 .globalD

Address review feedback.