Stop the context and update the state BEFORE throwing the exception.

apache · Mar 6, 2017 · 4d1222f · 4d1222f
1 parent d71207a
commit 4d1222f
Showing 1 changed file with 4 additions and 4 deletions.
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineResult.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineResult.java
@@ -184,13 +184,13 @@ protected void stop() {
       // been thrown during the "grace period".
       try {
         javaStreamingContext.awaitTermination(0);
+        SparkContextFactory.stopSparkContext(javaSparkContext);
+        if (Objects.equals(state, State.RUNNING)) {
+          state = State.STOPPED;
+        }
       } catch (Exception e) {
         throw beamExceptionFrom(e);
       }
-      SparkContextFactory.stopSparkContext(javaSparkContext);
-      if (Objects.equals(state, State.RUNNING)) {
-        state = State.STOPPED;
-      }
     }
 
     @Override