Fix TSAN bug by swapping ForkJoinPool with ScheduledExecutorService

Author: sjvanrossum

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

@@ -55,6 +55,7 @@
 import org.apache.beam.sdk.io.fs.MoveOptions.StandardMoveOptions;
 import org.apache.beam.sdk.io.fs.ResolveOptions.StandardResolveOptions;
 import org.apache.beam.sdk.io.fs.ResourceId;
+import org.apache.beam.sdk.options.ExecutorOptions;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.options.ValueProvider;
 import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;

sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java

@@ -29,6 +29,7 @@
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadLocalRandom;
 import org.apache.beam.sdk.annotations.Internal;
 import org.apache.beam.sdk.coders.CannotProvideCoderException;
@@ -1199,6 +1200,7 @@ public WriteShardsIntoTempFilesFn(Coder<UserT> inputCoder) {
     private transient List<CompletionStage<Void>> closeFutures = new ArrayList<>();
     private transient List<KV<Instant, FileResult<DestinationT>>> deferredOutput =
         new ArrayList<>();
+    private transient ScheduledExecutorService executorService;
 
     // Ensure that transient fields are initialized.
     private void readObject(java.io.ObjectInputStream in)
@@ -1208,6 +1210,11 @@ private void readObject(java.io.ObjectInputStream in)
       deferredOutput = new ArrayList<>();
     }
 
+    @Setup
+    public void setup(PipelineOptions options) {
+      executorService = options.as(ExecutorOptions.class).getScheduledExecutorService();
+    }
+
     @ProcessElement
     public void processElement(
         ProcessContext c, BoundedWindow window, MultiOutputReceiver outputReceiver)
@@ -1285,7 +1292,8 @@ private void closeWriterInBackground(Writer<DestinationT, OutputT> writer) {
                   writer.cleanup();
                   throw e;
                 }
-              }));
+              },
+              executorService));
     }
 
     @FinishBundle