title: Dataflow Splittable ReadFn not using multiple workers
body: <p>I have a particularly simple Dataflow pipeline where I want to read a file and output its parsed records to Avro. This works in most cases, except where the source file is particularly large (20+ GB) which causes me to OOM even with particularly large memory machines. I am pretty sure this happens because the non-splittable source is read in its entirety by Beam, so I implemented a splittable DoFn&lt;FileIO.ReadableFile, GenericRecord&gt;</p>
  
  <p>This <em>functionally</em> works in that the pipeline now succeeds, which seems to validate my assumption that the single large batch from a non-splittable file is the cause. However, this does not seem to spread the work across multiple workers. I tried the following:</p>
  
  <ol>
  <li>Disabled autoscaling (autoscalingAlgorithm=NONE) and set numWorkers to 10. This had the same throughput as numWorkers 1</li>
  <li>Left autoscaling on with a high maxWorkers. This went briefly up to 2, and then came back down to 1</li>
  <li>Added a shuffle (Reshuffle.viaRandomKey) after the DoFn, but before the Avro write</li>
  </ol>
  
  <p>Any ideas? The exact code is difficult to share because of company policy, but overall is pretty simple. I implemented the following:</p>
  
  <pre class="lang-java prettyprint-override"><code>public class SplittableReadFn extends DoFn&lt;FileIO.ReadableFile, GenericRecord&gt; {
  // ...
  @ProcessElement
  public void process(final ProcessContext c, final OffsetRangeTracker tracker) {
     final FileIO.ReadableFile file = c.element();
     // Followed by something like
     ReadableByteStream in = file.open()
     in.seek(tracker.from())
     Parser parser = new Parser(in)
     while (parser.next()) {
       if (parser.getOffset() &gt; tracker.to()) {
          break
       }
       tracker.tryClaim(parser.getOffset())
       c.output(parser.item())
     }
     tracker.markDone()
  }
  
  @GetInitialRestriction
  public OffsetRange getInitialRestriction(final FileIO.ReadableFile file) {
     return new Offset(0, getSize(file) - 1);
  }
  
  @SplitRestriction
  public void splitRestriction(final FileIO.ReadableFile file, final OffsetRange restriction, final DoFn.OutputReceiver&lt;OffsetRange&gt; receiver) {
      // chunkRange for test purposes just breaks into at most 500MB chunks
      for (final OffsetRange chunk: chunkRange(restriction)) {
         receiver.output(chunk);
      }
  }
  </code></pre>