[BEAM-59] Register standard FileSystems wherever we register IOChanne…

…lFactories Additionally, drop an unnecessary use of `GcsOptions` in `PipelineRunner`.
apache · Apr 18, 2017 · 88cc651 · 88cc651
1 parent 36e4355
commit 88cc651
Show file tree

Hide file tree

Showing 4 changed files with 9 additions and 4 deletions.
diff --git a/.../main/java/org/apache/beam/runners/flink/translation/utils/SerializedPipelineOptions.java b/.../main/java/org/apache/beam/runners/flink/translation/utils/SerializedPipelineOptions.java
@@ -24,6 +24,7 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.Serializable;
+import org.apache.beam.sdk.io.FileSystems;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.util.IOChannelUtils;
 
@@ -55,6 +56,7 @@ public PipelineOptions getPipelineOptions() {
         pipelineOptions = new ObjectMapper().readValue(serializedOptions, PipelineOptions.class);
 
         IOChannelUtils.registerIOFactoriesAllowOverride(pipelineOptions);
+        FileSystems.setDefaultConfigInWorkers(pipelineOptions);
       } catch (IOException e) {
         throw new RuntimeException("Couldn't deserialize the PipelineOptions.", e);
       }

diff --git a/...rs/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkRuntimeContext.java b/...rs/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkRuntimeContext.java
@@ -29,6 +29,7 @@
 import org.apache.beam.sdk.coders.CannotProvideCoderException;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.CoderRegistry;
+import org.apache.beam.sdk.io.FileSystems;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.transforms.Aggregator;
 import org.apache.beam.sdk.transforms.Combine;
@@ -130,6 +131,7 @@ static PipelineOptions getOrInit(String serializedPipelineOptions) {
         }
         // register IO factories.
         IOChannelUtils.registerIOFactoriesAllowOverride(pipelineOptions);
+        FileSystems.setDefaultConfigInWorkers(pipelineOptions);
       }
       return pipelineOptions;
     }

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PipelineRunner.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PipelineRunner.java
@@ -21,9 +21,8 @@
 
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.PipelineResult;
-import org.apache.beam.sdk.options.GcsOptions;
+import org.apache.beam.sdk.io.FileSystems;
 import org.apache.beam.sdk.options.PipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptionsValidator;
 import org.apache.beam.sdk.util.IOChannelUtils;
 import org.apache.beam.sdk.util.InstanceBuilder;
 
@@ -41,11 +40,11 @@ public abstract class PipelineRunner<ResultT extends PipelineResult> {
    * @return The newly created runner.
    */
   public static PipelineRunner<? extends PipelineResult> fromOptions(PipelineOptions options) {
-    GcsOptions gcsOptions = PipelineOptionsValidator.validate(GcsOptions.class, options);
     checkNotNull(options);
 
     // (Re-)register standard IO factories. Clobbers any prior credentials.
-    IOChannelUtils.registerIOFactoriesAllowOverride(gcsOptions);
+    IOChannelUtils.registerIOFactoriesAllowOverride(options);
+    FileSystems.setDefaultConfigInWorkers(options);
 
     @SuppressWarnings("unchecked")
     PipelineRunner<? extends PipelineResult> result =

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipeline.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipeline.java
@@ -40,6 +40,7 @@
 import javax.annotation.Nullable;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.io.FileSystems;
 import org.apache.beam.sdk.options.ApplicationNameOptions;
 import org.apache.beam.sdk.options.GcpOptions;
 import org.apache.beam.sdk.options.PipelineOptions;
@@ -404,6 +405,7 @@ public static PipelineOptions testingPipelineOptions() {
       options.setStableUniqueNames(CheckEnabled.ERROR);
 
       IOChannelUtils.registerIOFactoriesAllowOverride(options);
+      FileSystems.setDefaultConfigInWorkers(options);
       return options;
     } catch (IOException e) {
       throw new RuntimeException(