title: Beam TextIO write NullPointerException because destination is null
body: <p>I am using the Apache Beam DirectRunner, and I have defined a pipeline as follows:</p>
  
  <pre><code>val p = Pipeline.create(options)
  p.apply(Create.of("/tmp/dc/foo.txt"))
          .apply(FileLoader())
          .apply(SaveLineToRedis())
          .apply(AddToRedisIndex())
          .apply(MatchTransform())
          .apply(GroupByKey.create())
          .apply(TextIO.writeCustomType&lt;KV&lt;String, Iterable&lt;SimpleMatcherResult&gt;&gt;&gt;().to("/tmp/bar"))
  </code></pre>
  
  <p>The writing out fails with:</p>
  
  <pre><code>13:45:47.247 [direct-runner-worker] INFO  org.apache.beam.sdk.io.WriteFiles - Opening writer 83c36e3f-7e1f-406c-a9c6-f3ab4bac1cb7 for window org.apache.beam.sdk.transforms.windowing.GlobalWindow@29caf222 pane PaneInfo{isFirst=true, isLast=true, timing=ON_TIME, index=0, onTimeIndex=0} destination null
  Exception in thread "main" org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.NullPointerException
          at org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFinish(DirectRunner.java:342)
          at org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFinish(DirectRunner.java:312)
          at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:206)
          at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:62)
          at org.apache.beam.sdk.Pipeline.run(Pipeline.java:311)
          at de.techmatrix.dc.matcher.MainKt.main(Main.kt:12)
  Caused by: java.lang.NullPointerException
          at org.apache.beam.sdk.io.DynamicFileDestinations$ConstantFilenamePolicy.formatRecord(DynamicFileDestinations.java:49)
          at org.apache.beam.sdk.io.WriteFiles$WriteShardsIntoTempFilesFn.processElement(WriteFiles.java:718)
  </code></pre>
  
  <p>The destination is returned by the <code>DynamicFileDestinations</code> as <code>null</code>:</p>
  
  <pre><code>public Void getDestination(UserT element) {
    return (Void) null;
  }
  </code></pre>
  
  <p><strong>Update:</strong> This works with <code>FileIO</code>:</p>
  
  <pre><code>        .apply(FileIO.writeDynamic&lt;String, KV&lt;String, Iterable&lt;SimpleMatcherResult&gt;&gt;&gt;()
                  .by { it.value.first().matchedKey }.withDestinationCoder(StringUtf8Coder.of())
                  .via(Contextful.fn({ mapper.writeValueAsString(it) }), TextIO.sink())
                  .to("/tmp/bar")
                  .withNaming{ _ -&gt; defaultNaming("matches", "txt")})
  </code></pre>
  
  <p>Could someone explain why?</p>