Add Documentation Hint for Template Job Creation in DataflowRunner (#34204)

Author: gayatrikate04

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java

@@ -201,6 +201,17 @@
  *
  * <p>Please see <a href="https://cloud.google.com/dataflow/security-and-permissions">Google Cloud
  * Dataflow Security and Permissions</a> for more details.
+ *
+ * <p>DataflowRunner now supports creating job templates using the {@code --templateLocation}
+ * option. If this option is set, the runner will generate a template instead of running the
+ * pipeline immediately.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * --runner=DataflowRunner
+ * --templateLocation=gs://your-bucket/templates/my-template
+ * }</pre>
  */
 @SuppressWarnings({
   "rawtypes", // TODO(https://github.com/apache/beam/issues/20447)
@@ -595,6 +606,7 @@ protected DataflowRunner(DataflowPipelineOptions options) {
 
   private static class AlwaysCreateViaRead<T>
       implements PTransformOverrideFactory<PBegin, PCollection<T>, Create.Values<T>> {
+
     @Override
     public PTransformOverrideFactory.PTransformReplacement<PBegin, PCollection<T>>
         getReplacementTransform(
@@ -775,7 +787,7 @@ private List<PTransformOverride> getOverrides(boolean streaming) {
         PTransformOverride.of(
             PTransformMatchers.requiresStableInputParDoMulti(),
             RequiresStableInputParDoOverrides.multiOutputOverrideFactory()));
-    */
+         */
     overridesBuilder
         .add(
             PTransformOverride.of(
@@ -949,7 +961,7 @@ public Map<PCollection<?>, ReplacementOutput> mapOutputs(
       The PCollectionView itself must have the same tag since that tag may have been embedded in serialized DoFns
       previously and cannot easily be rewired. The PCollection may differ, so we rewire it, even if the rewiring
       is a noop.
-      */
+             */
       return ReplacementOutputs.singleton(outputs, newOutput);
     }
   }
@@ -1179,6 +1191,7 @@ private List<RunnerApi.ArtifactInformation> getDefaultArtifacts() {
   @VisibleForTesting
   static boolean isMultiLanguagePipeline(Pipeline pipeline) {
     class IsMultiLanguageVisitor extends PipelineVisitor.Defaults {
+
       private boolean isMultiLanguage = false;
 
       private void performMultiLanguageTest(Node node) {
@@ -1656,6 +1669,7 @@ private static EnvironmentInfo getEnvironmentInfoFromEnvironmentId(
 
   @AutoValue
   abstract static class EnvironmentInfo {
+
     static EnvironmentInfo create(
         String environmentId, String containerUrl, List<String> capabilities) {
       return new AutoValue_DataflowRunner_EnvironmentInfo(
@@ -1954,7 +1968,6 @@ void recordViewUsesNonDeterministicKeyCoder(PTransform<?, ?> ptransform) {
   // ================================================================================
   // PubsubIO translations
   // ================================================================================
-
   private static class StreamingPubsubIOReadOverrideFactory
       implements PTransformOverrideFactory<
           PBegin, PCollection<PubsubMessage>, PubsubUnboundedSource> {
@@ -2113,6 +2126,7 @@ protected String getKindString() {
   }
 
   private static class StreamingPubsubSinkTranslators {
+
     /** Rewrite {@link StreamingPubsubIOWrite} to the appropriate internal node. */
     static class StreamingPubsubIOWriteTranslator
         implements TransformTranslator<StreamingPubsubIOWrite> {
@@ -2166,9 +2180,9 @@ private static void translate(
   }
 
   // ================================================================================
-
   private static class SingleOutputExpandableTransformTranslator
       implements TransformTranslator<External.SingleOutputExpandableTransform> {
+
     @Override
     public void translate(
         External.SingleOutputExpandableTransform transform, TranslationContext context) {
@@ -2186,6 +2200,7 @@ public void translate(
 
   private static class MultiOutputExpandableTransformTranslator
       implements TransformTranslator<External.MultiOutputExpandableTransform> {
+
     @Override
     public void translate(
         External.MultiOutputExpandableTransform transform, TranslationContext context) {
@@ -2734,6 +2749,7 @@ static void verifyStateSupportForWindowingStrategy(WindowingStrategy strategy) {
    */
   private static class DataflowPayloadTranslator
       implements TransformPayloadTranslator<PTransform<?, ?>> {
+
     @Override
     public String getUrn(PTransform transform) {
       return "dataflow_stub:" + transform.getClass().getName();
@@ -2758,6 +2774,7 @@ public RunnerApi.FunctionSpec translate(
   })
   @AutoService(TransformPayloadTranslatorRegistrar.class)
   public static class DataflowTransformTranslator implements TransformPayloadTranslatorRegistrar {
+
     @Override
     public Map<? extends Class<? extends PTransform>, ? extends TransformPayloadTranslator>
         getTransformPayloadTranslators() {

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineOptions.java

@@ -105,6 +105,19 @@ public interface DataflowPipelineOptions
           + "Must either be local or Cloud Storage.")
   String getTemplateLocation();
 
+  /**
+   * Sets the Cloud Storage path where the Dataflow template will be stored. Required for creating
+   * Flex Templates or Classic Templates.
+   *
+   * <p>Example:
+   *
+   * <pre>{@code
+   * DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
+   * options.setTemplateLocation("gs://your-bucket/templates/my-template");
+   * }</pre>
+   *
+   * @param value Cloud Storage path for storing the Dataflow template.
+   */
   void setTemplateLocation(String value);
 
   /**
@@ -181,10 +194,8 @@ public interface DataflowPipelineOptions
   enum FlexResourceSchedulingGoal {
     /** No goal specified. */
     UNSPECIFIED,
-
     /** Optimize for lower execution time. */
     SPEED_OPTIMIZED,
-
     /** Optimize for lower cost. */
     COST_OPTIMIZED,
   }
@@ -198,6 +209,7 @@ enum FlexResourceSchedulingGoal {
 
   /** Returns a default staging location under {@link GcpOptions#getGcpTempLocation}. */
   class StagingLocationFactory implements DefaultValueFactory<String> {
+
     private static final Logger LOG = LoggerFactory.getLogger(StagingLocationFactory.class);
 
     @Override