[beam] branch master updated: Do not stage dataflow worker jar when use runner_v2.

[boyuanz]

This is an automated email from the ASF dual-hosted git repository.

boyuanz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new a31e004  Do not stage dataflow worker jar when use runner_v2.
     new 11be7aa  Merge pull request #14100 from Do not stage dataflow worker 
jar when use runner_v2.
a31e004 is described below

commit a31e00433449fdbdf63a053cedec60d060d5ab9c
Author: Boyuan Zhang <boyu...@google.com>
AuthorDate: Thu Feb 25 17:24:07 2021 -0800

    Do not stage dataflow worker jar when use runner_v2.
---
 .../main/java/org/apache/beam/runners/dataflow/DataflowRunner.java    | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git 
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
 
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
index 97a4329..f03b40c 100644
--- 
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
+++ 
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
@@ -901,7 +901,7 @@ public class DataflowRunner extends 
PipelineRunner<DataflowPipelineJob> {
     String windmillBinary =
         
options.as(DataflowPipelineDebugOptions.class).getOverrideWindmillBinary();
     String dataflowWorkerJar = options.getDataflowWorkerJar();
-    if (dataflowWorkerJar != null && !dataflowWorkerJar.isEmpty()) {
+    if (dataflowWorkerJar != null && !dataflowWorkerJar.isEmpty() && 
!useUnifiedWorker(options)) {
       // Put the user specified worker jar at the start of the classpath, to 
be consistent with the
       // built in worker order.
       pathsToStageBuilder.add("dataflow-worker.jar=" + dataflowWorkerJar);
@@ -992,7 +992,7 @@ public class DataflowRunner extends 
PipelineRunner<DataflowPipelineJob> {
         options.getStager().stageToFile(serializedProtoPipeline, 
PIPELINE_FILE_NAME);
     dataflowOptions.setPipelineUrl(stagedPipeline.getLocation());
 
-    if (!isNullOrEmpty(dataflowOptions.getDataflowWorkerJar())) {
+    if (!isNullOrEmpty(dataflowOptions.getDataflowWorkerJar()) && 
!useUnifiedWorker(options)) {
       List<String> experiments =
           dataflowOptions.getExperiments() == null
               ? new ArrayList<>()