class HDFSFileSource extends BoundedSource<T> {

  public static <T, K, V, W extends FileInputFormat<K, V>> HDFSFileSource<T, K, V>
  from(String filepattern,
       Class<W> formatClass,
       Coder<T> coder,
       SerializableFunction<KV<K, V>, T> inputConverter) {
    return HDFSFileSource.<T, K, V>builder()
        .setFilepattern(filepattern)
        .setFormatClass(formatClass)
        .setCoder(coder)
        .setInputConverter(inputConverter)
        .setConfiguration(null)
        .setUsername(null)
        .setValidateSource(true)
        .setSerializableSplit(null)
        .build();
  }

}