class HadoopFileSystemRegistrar implements FileSystemRegistrar {

  @Override
  public Iterable<FileSystem> fromOptions(@Nonnull PipelineOptions options) {
    List<Configuration> configurations =
        options.as(HadoopFileSystemOptions.class).getHdfsConfiguration();
    if (configurations == null) {
      configurations = Collections.emptyList();
    }
    checkArgument(configurations.size() <= 1,
        String.format(
            "The %s currently only supports at most a single Hadoop configuration.",
            HadoopFileSystemRegistrar.class.getSimpleName()));

    ImmutableList.Builder<FileSystem> builder = ImmutableList.builder();
    for (Configuration configuration : configurations) {
      try {
        builder.add(new HadoopFileSystem(configuration));
      } catch (IOException e) {
        throw new IllegalArgumentException(String.format(
            "Failed to construct Hadoop filesystem with configuration %s", configuration), e);
      }
    }
    return builder.build();
  }

}