Update to Beam 2.16.0 API

mattcasters · Nov 15, 2019 · 081cf37 · 081cf37
1 parent ad90060
commit 081cf37
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 6 deletions.
diff --git a/pom.xml b/pom.xml
@@ -22,7 +22,7 @@
     <pentaho-kettle.version>8.2.0.0-342</pentaho-kettle.version>
     <mockito-all.version>1.9.5</mockito-all.version>
     <junit.version>4.4</junit.version>
-    <apache-beam-version>2.15.0</apache-beam-version>
+    <apache-beam-version>2.16.0</apache-beam-version>
     <kettle-beam-version>1.0.0-SNAPSHOT</kettle-beam-version>
     <spark-version>2.4.3</spark-version>
   </properties>

diff --git a/src/main/java/org/kettle/beam/pipeline/KettleBeamPipelineExecutor.java b/src/main/java/org/kettle/beam/pipeline/KettleBeamPipelineExecutor.java
@@ -551,7 +551,7 @@ private void configureFlinkOptions( BeamJobConfig config, FlinkPipelineOptions o
       try {
         CheckpointingMode mode = CheckpointingMode.valueOf( modeString);
         if ( mode != null ) {
-          options.setCheckpointingMode( mode );
+          options.setCheckpointingMode( modeString );
         }
       } catch(Exception e) {
         throw new IOException( "Unable to parse flink check pointing mode '"+modeString+"'", e );
@@ -673,12 +673,12 @@ private void configureFlinkOptions( BeamJobConfig config, FlinkPipelineOptions o
     // Flink mode for data exchange of batch pipelines. Reference {@link org.apache.flink.api.common.ExecutionMode}.
     // Set this to BATCH_FORCED if pipelines get blocked, see https://issues.apache.org/jira/browse/FLINK-10672")
     if (StringUtils.isNotEmpty( config.getFlinkExecutionModeForBatch() )) {
-      String str = space.environmentSubstitute( config.getFlinkExecutionModeForBatch() );
-      ExecutionMode value = ExecutionMode.valueOf( str );
+      String modeString = space.environmentSubstitute( config.getFlinkExecutionModeForBatch() );
+      ExecutionMode mode = ExecutionMode.valueOf( modeString );
       try {
-        options.setExecutionModeForBatch( value );
+        options.setExecutionModeForBatch( modeString );
       } catch(Exception e) {
-        throw new IOException( "Unable to parse flink execution mode for batch '"+str+"'", e );
+        throw new IOException( "Unable to parse flink execution mode for batch '"+modeString+"'", e );
       }
     }
   }