Prevent concurrent JobServer shutdown issue

apache · Dec 18, 2018 · 1736b9c · 1736b9c
1 parent 12bf1d3
commit 1736b9c
Showing 1 changed file with 7 additions and 3 deletions.
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkJobServerDriver.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkJobServerDriver.java
@@ -49,8 +49,8 @@ public class FlinkJobServerDriver implements Runnable {
   @VisibleForTesting ServerConfiguration configuration;
   private final ServerFactory jobServerFactory;
   private final ServerFactory artifactServerFactory;
-  private GrpcFnServer<InMemoryJobService> jobServer;
-  private GrpcFnServer<BeamFileSystemArtifactStagingService> artifactStagingServer;
+  private volatile GrpcFnServer<InMemoryJobService> jobServer;
+  private volatile GrpcFnServer<BeamFileSystemArtifactStagingService> artifactStagingServer;
 
   /** Configuration for the jobServer. */
   public static class ServerConfiguration {
@@ -184,12 +184,16 @@ public void run() {
     }
   }
 
+  // This method is executed by TestPortableRunner via Reflection
   public String start() throws IOException {
     jobServer = createJobServer();
     return jobServer.getApiServiceDescriptor().getUrl();
   }
 
-  public void stop() {
+  // This method is executed by TestPortableRunner via Reflection
+  // Needs to be synchronized to prevent concurrency issues in testing shutdown
+  @SuppressWarnings("WeakerAccess")
+  public synchronized void stop() {
     if (jobServer != null) {
       try {
         jobServer.close();