Consolidate SamzaPipelineOptions and SamzaPortablePipelineOptions

linkedin · May 19, 2022 · 79738cc · 79738cc
1 parent 1b0df38
commit 79738cc
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 35 deletions.
diff --git a/runners/samza/src/main/java/org/apache/beam/runners/samza/SamzaJobInvoker.java b/runners/samza/src/main/java/org/apache/beam/runners/samza/SamzaJobInvoker.java
@@ -26,6 +26,7 @@
 import org.apache.beam.runners.jobsubmission.JobInvoker;
 import org.apache.beam.runners.jobsubmission.PortablePipelineJarCreator;
 import org.apache.beam.runners.jobsubmission.PortablePipelineRunner;
+import org.apache.beam.sdk.options.PortablePipelineOptions;
 import org.apache.beam.vendor.grpc.v1p43p2.com.google.protobuf.Struct;
 import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Strings;
 import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.ListeningExecutorService;
@@ -59,11 +60,12 @@ protected JobInvocation invokeWithExecutor(
       @Nullable String retrievalToken,
       ListeningExecutorService executorService) {
     LOG.trace("Parsing pipeline options");
-    final SamzaPortablePipelineOptions samzaOptions =
-        PipelineOptionsTranslation.fromProto(options).as(SamzaPortablePipelineOptions.class);
+    final SamzaPipelineOptions samzaOptions =
+        PipelineOptionsTranslation.fromProto(options).as(SamzaPipelineOptions.class);
+    final PortablePipelineOptions portableOptions = samzaOptions.as(PortablePipelineOptions.class);
 
     final PortablePipelineRunner pipelineRunner;
-    if (Strings.isNullOrEmpty(samzaOptions.getOutputExecutablePath())) {
+    if (Strings.isNullOrEmpty(portableOptions.getOutputExecutablePath())) {
       pipelineRunner = new SamzaPipelineRunner(samzaOptions);
     } else {
       /*

diff --git a/runners/samza/src/main/java/org/apache/beam/runners/samza/SamzaPipelineOptions.java b/runners/samza/src/main/java/org/apache/beam/runners/samza/SamzaPipelineOptions.java
@@ -56,6 +56,13 @@ public interface SamzaPipelineOptions extends PipelineOptions {
 
   void setJobInstance(String instance);
 
+  @Description(
+      "The file path for the local file system token. If not set (by default), then the runner would"
+          + " not use secure server factory.")
+  String getFsTokenPath();
+
+  void setFsTokenPath(String path);
+
   @Description(
       "Samza application execution environment."
           + "See {@link org.apache.beam.runners.samza.SamzaExecutionEnvironment} for detailed environment descriptions.")

diff --git a/runners/samza/src/main/java/org/apache/beam/runners/samza/SamzaPortablePipelineOptions.java b/runners/samza/src/main/java/org/apache/beam/runners/samza/SamzaPortablePipelineOptions.java