Move tempLocation to PipelineOptions.

runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/FlinkGroupAlsoByWindowWrapper.java

@@ -476,6 +476,15 @@ public CheckEnabled getStableUniqueNames() {
           @Override
           public void setStableUniqueNames(CheckEnabled enabled) {
           }
+
+          @Override
+          public String getTempLocation() {
+            return null;
+          }
+
+          @Override
+          public void setTempLocation(String tempLocation) {
+          }
         };
       }
       return options;
@@ -628,4 +637,4 @@ public void restoreState(StreamTaskState taskState, long recoveryTimestamp) thro
     // restore the timerInternals.
     this.timerInternals.restoreTimerInternals(reader, inputKvCoder, windowCoder);
   }
-}
+}

sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineOptions.java

@@ -34,8 +34,6 @@ public interface DataflowPipelineOptions extends
     GcsOptions, StreamingOptions, CloudDebuggerOptions, DataflowWorkerLoggingOptions,
     DataflowProfilingOptions {
 
-  static final String DATAFLOW_STORAGE_LOCATION = "Dataflow Storage Location";
-
   @Description("Project id. Required when running a Dataflow in the cloud. "
       + "See https://cloud.google.com/storage/docs/projects for further details.")
   @Override
@@ -45,37 +43,19 @@ public interface DataflowPipelineOptions extends
   @Override
   void setProject(String value);
 
-  /**
-   * GCS path for temporary files, e.g. gs://bucket/object
-   *
-   * <p>Must be a valid Cloud Storage URL, beginning with the prefix "gs://"
-   *
-   * <p>At least one of {@link #getTempLocation()} or {@link #getStagingLocation()} must be set. If
-   * {@link #getTempLocation()} is not set, then the Dataflow pipeline defaults to using
-   * {@link #getStagingLocation()}.
-   */
-  @Description("GCS path for temporary files, eg \"gs://bucket/object\". "
-      + "Must be a valid Cloud Storage URL, beginning with the prefix \"gs://\". "
-      + "At least one of tempLocation or stagingLocation must be set. If tempLocation is unset, "
-      + "defaults to using stagingLocation.")
-  @Validation.Required(groups = {DATAFLOW_STORAGE_LOCATION})
-  String getTempLocation();
-  void setTempLocation(String value);
-
   /**
    * GCS path for staging local files, e.g. gs://bucket/object
    *
    * <p>Must be a valid Cloud Storage URL, beginning with the prefix "gs://"
    *
-   * <p>At least one of {@link #getTempLocation()} or {@link #getStagingLocation()} must be set. If
-   * {@link #getTempLocation()} is not set, then the Dataflow pipeline defaults to using
-   * {@link #getStagingLocation()}.
+   * <p>At least one of {@link PipelineOptions#getTempLocation()} or {@link #getStagingLocation()}
+   * must be set. If {@link #getStagingLocation()} is not set, then the Dataflow
+   * pipeline defaults to using {@link PipelineOptions#getTempLocation()}.
    */
   @Description("GCS path for staging local files, e.g. \"gs://bucket/object\". "
       + "Must be a valid Cloud Storage URL, beginning with the prefix \"gs://\". "
       + "At least one of stagingLocation or tempLocation must be set. If stagingLocation is unset, "
       + "defaults to using tempLocation.")
-  @Validation.Required(groups = {DATAFLOW_STORAGE_LOCATION})
   String getStagingLocation();
   void setStagingLocation(String value);
 

sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PipelineOptions.java

@@ -246,4 +246,18 @@ public static enum CheckEnabled {
   @Default.Enum("WARNING")
   CheckEnabled getStableUniqueNames();
   void setStableUniqueNames(CheckEnabled enabled);
+
+  /**
+   * A pipeline level default location for storing temporary files.
+   *
+   * <p>This can be a path of any file system.
+   *
+   * <p>{@link #getTempLocation()} can be used as a default location in other
+   * {@link PipelineOptions}.
+   *
+   * <p>If it is unset, {@link PipelineRunner} can override it.
+   */
+  @Description("A pipeline level default location for storing temporary files.")
+  String getTempLocation();
+  void setTempLocation(String value);
 }

sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/DataflowPipelineRunner.java

@@ -251,6 +251,10 @@ public static DataflowPipelineRunner fromOptions(PipelineOptions options) {
     }
 
     PathValidator validator = dataflowOptions.getPathValidator();
+    Preconditions.checkArgument(!(Strings.isNullOrEmpty(dataflowOptions.getTempLocation())
+        && Strings.isNullOrEmpty(dataflowOptions.getStagingLocation())),
+        "Missing required value: at least one of tempLocation or stagingLocation must be set.");
+
     if (dataflowOptions.getStagingLocation() != null) {
       validator.validateOutputFilePrefixSupported(dataflowOptions.getStagingLocation());
     }

sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/runners/DataflowPipelineRunnerTest.java

@@ -649,10 +649,8 @@ public void testNoStagingLocationAndNoTempLocationFails() {
     options.setProject("foo-project");
 
     thrown.expect(IllegalArgumentException.class);
-    thrown.expectMessage("Missing required value for group");
-    thrown.expectMessage(DataflowPipelineOptions.DATAFLOW_STORAGE_LOCATION);
-    thrown.expectMessage("getStagingLocation");
-    thrown.expectMessage("getTempLocation");
+    thrown.expectMessage(
+        "Missing required value: at least one of tempLocation or stagingLocation must be set.");
 
     DataflowPipelineRunner.fromOptions(options);
   }