HadoopFileSystemRegistrar implements FileSystemRegistrar { @Override public Iterable<FileSystem<?>> fromOptions(@Nonnull PipelineOptions options) { final List<Configuration> configurations = options.as(HadoopFileSystemOptions.class).getHdfsConfiguration(); if (configurations == null) { return Collections.emptyList(); } checkArgument( configurations.size() == 1, String.format( "The %s currently only supports at most a single Hadoop configuration.", HadoopFileSystemRegistrar.class.getSimpleName())); final ImmutableList.Builder<FileSystem<?>> builder = ImmutableList.builder(); final Set<String> registeredSchemes = new HashSet<>(); final Configuration configuration = Iterables.getOnlyElement(configurations); final String defaultFs = configuration.get(org.apache.hadoop.fs.FileSystem.FS_DEFAULT_NAME_KEY); if (defaultFs != null && !defaultFs.isEmpty()) { final String scheme = Objects.requireNonNull( URI.create(defaultFs).getScheme(), String.format( "Empty scheme for %s value.", org.apache.hadoop.fs.FileSystem.FS_DEFAULT_NAME_KEY)); builder.add(new HadoopFileSystem(scheme, configuration)); registeredSchemes.add(scheme); } final String nameServices = configuration.get(CONFIG_KEY_DFS_NAMESERVICES); if (nameServices != null && !nameServices.isEmpty()) { for (String scheme : HA_SCHEMES) { if (!registeredSchemes.contains(scheme)) { builder.add(new HadoopFileSystem(scheme, configuration)); } } } return builder.build(); }  @Override Iterable<FileSystem<?>> fromOptions(@Nonnull PipelineOptions options);  }
@Test public void testServiceLoader() { HadoopFileSystemOptions options = PipelineOptionsFactory.as(HadoopFileSystemOptions.class); options.setHdfsConfiguration(ImmutableList.of(configuration)); for (FileSystemRegistrar registrar : Lists.newArrayList(ServiceLoader.load(FileSystemRegistrar.class).iterator())) { if (registrar instanceof HadoopFileSystemRegistrar) { Iterable<FileSystem<?>> fileSystems = registrar.fromOptions(options); assertEquals( hdfsClusterBaseUri.getScheme(), ((HadoopFileSystem) Iterables.getOnlyElement(fileSystems)).getScheme()); return; } } fail("Expected to find " + HadoopFileSystemRegistrar.class); }