title: SpannerIO java.lang.IllegalStateException: Sorter should be null here
body: <p>I am trying to write to Spanner from a DataFlow streaming job by using</p>
  
  <pre><code>&lt;dependency&gt;
    &lt;groupId&gt;org.apache.beam&lt;/groupId&gt;
    &lt;artifactId&gt;beam-sdks-java-io-google-cloud-platform&lt;/artifactId&gt;
    &lt;version&gt;2.18.0&lt;/version&gt;
  &lt;/dependency&gt;
  </code></pre>
  
  <p>After mapping the data to <code>PCollection&lt;Mutation&gt;</code> I am writing them to Spanner via <code>SpannerIO.write</code></p>
  
  <pre><code>Pipeline pipeline = Pipeline.create(options);
  
  PCollection&lt;Mutation&gt; mutations = pipeline.apply...
  
  mutations.apply("WriteMutations", SpannerIO.write()
                  .withInstanceId(INSTANCE_ID)
                  .withDatabaseId(DATABASE_ID)
  );
  
  pipeline.run();
  </code></pre>
  
  <p>However, it throws</p>
  
  <pre><code>java.lang.IllegalStateException: Sorter should be null here
          at org.apache.beam.sdk.io.gcp.spanner.SpannerIO$GatherBundleAndSortFn.startBundle (SpannerIO.java:1080)
  </code></pre>
  
  <p>What would be the cause of this exception?</p>
  
  <p>The following pipeline produces the exception. I test it with 20 workers but it looks like it's independent of the data load.</p>
  
  <pre><code>import com.google.cloud.spanner.Mutation;
  
  import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
  import org.apache.beam.sdk.Pipeline;
  import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
  import org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage;
  import org.apache.beam.sdk.io.gcp.spanner.SpannerIO;
  import org.apache.beam.sdk.options.Description;
  import org.apache.beam.sdk.options.PipelineOptionsFactory;
  import org.apache.beam.sdk.transforms.DoFn;
  import org.apache.beam.sdk.transforms.ParDo;
  import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
  import org.apache.beam.sdk.transforms.windowing.GlobalWindows;
  import org.apache.beam.sdk.transforms.windowing.Repeatedly;
  import org.apache.beam.sdk.transforms.windowing.Window;
  import org.joda.time.Duration;
  
  import java.util.UUID;
  
  public final class TestPipeline {
  
      private static final Duration WINDOW_DURATION = Duration.standardSeconds(1);
      private static final String DATABASE_ID = "test";
      private static final String INSTANCE_ID = "test-spanner";
      private static final String TEST_TABLE = "test";
  
      public static void main(String[] args) {
          TestPipelineOptions options = PipelineOptionsFactory
                  .fromArgs(args)
                  .withValidation()
                  .as(TestPipelineOptions.class);
  
          Pipeline pipeline = Pipeline.create(options);
  
          pipeline
                  .apply("Read pubsub", PubsubIO.readMessagesWithAttributes()
                          .fromSubscription(options.getInputSubscription()))
                  .apply("Parse message", ParDo.of(new ProcessMessage()))
                  .apply("Windowing", Window.&lt;Mutation&gt;into(new GlobalWindows())
                          .triggering(Repeatedly.forever(
                                  AfterProcessingTime.pastFirstElementInPane()
                                          .plusDelayOf(WINDOW_DURATION)))
                          .withAllowedLateness(Duration.ZERO)
                          .discardingFiredPanes())
                  .apply("Write mutations", SpannerIO.write()
                          .withInstanceId(INSTANCE_ID)
                          .withDatabaseId(DATABASE_ID)
                  );
  
          pipeline.run();
      }
  
      private static class ProcessMessage extends DoFn&lt;PubsubMessage, Mutation&gt; {
  
          @ProcessElement
          public void processElement(@Element final PubsubMessage message,
                                     final OutputReceiver&lt;Mutation&gt; out) {
              out.output(Mutation.newInsertOrUpdateBuilder(TEST_TABLE)
                      .set("id").to(UUID.randomUUID().toString())
                      .set("string").to("test")
                      .set("count").to(Long.MAX_VALUE)
                      .build()
              );
          }
      }
  
      interface TestPipelineOptions extends DataflowPipelineOptions {
  
          void setInputSubscription(String inputSubscription);
  
          @Description("Google Pubsub subscription id.")
          String getInputSubscription();
      }
  
  }
  </code></pre>
  
  <p>Table <code>CREATE TABLE test (id STRING(50) NOT NULL, string STRING(50) NOT NULL, count INT64) PRIMARY KEY (id);</code></p>