title: Why Apache beam can't infer the default coder when using KV<String, String>?
body: <p>I'm implementing the <a href="https://github.com/apache/beam/blob/master/examples/java/src/main/java/org/apache/beam/examples/cookbook/CombinePerKeyExamples.java" rel="nofollow noreferrer">CombinePerKeyExample</a> using a subclass of <code>CombineFn</code>  instead of using an implementation of <code>SerializableFunction</code>. </p>
  
  <pre class="lang-java prettyprint-override"><code>package me.examples;
  
  import org.apache.beam.sdk.coders.AvroCoder;
  import org.apache.beam.sdk.coders.DefaultCoder;
  import org.apache.beam.sdk.transforms.Combine.CombineFn;
  
  import java.util.HashSet;
  import java.util.Set;
  
  public class ConcatWordsCombineFn extends CombineFn&lt;String, ConcatWordsCombineFn.Accumulator, String&gt; {
      @DefaultCoder(AvroCoder.class)
      public static class Accumulator{
          HashSet&lt;String&gt; plays;
      }
  
      @Override
      public Accumulator createAccumulator(){
          Accumulator accumulator = new Accumulator();
          accumulator.plays = new HashSet&lt;&gt;();
          return accumulator;
      }
  
      @Override
      public Accumulator addInput(Accumulator accumulator, String input){
          accumulator.plays.add(input);
          return accumulator;
      }
  
      @Override
      public Accumulator mergeAccumulators(Iterable&lt;Accumulator&gt; accumulators){
          Accumulator mergeAccumulator = new Accumulator();
          mergeAccumulator.plays = new HashSet&lt;&gt;();
  
          for(Accumulator accumulator: accumulators){
              mergeAccumulator.plays.addAll(accumulator.plays);
          }
  
          return mergeAccumulator;
      }
  
      @Override
      public String extractOutput(Accumulator accumulator){
          return String.join(",", accumulator.plays);
      }
  }
  
  </code></pre>
  
  <p>The pipeline is composed of a <code>ReadFromBigQuery</code>, <code>ExtractAllPlaysOfWords</code> (code below) and <code>WriteToBigQuery</code> </p>
  
  <pre class="lang-java prettyprint-override"><code>package me.examples;
  
  import com.google.api.services.bigquery.model.TableRow;
  import org.apache.beam.sdk.coders.KvCoder;
  import org.apache.beam.sdk.coders.StringUtf8Coder;
  import org.apache.beam.sdk.transforms.Combine;
  import org.apache.beam.sdk.transforms.PTransform;
  import org.apache.beam.sdk.transforms.ParDo;
  import org.apache.beam.sdk.values.KV;
  import org.apache.beam.sdk.values.PCollection;
  
  public class PlaysForWord extends PTransform&lt;PCollection&lt;TableRow&gt;, PCollection&lt;TableRow&gt;&gt; {
  
  
      @Override
      public PCollection&lt;TableRow&gt; expand(PCollection&lt;TableRow&gt; input) {
  
              PCollection&lt;KV&lt;String, String&gt;&gt; largeWords = input.apply("ExtractLargeWords", ParDo.of(new ExtractLargeWordsFn()));
  
              //PCollection&lt;KV&lt;String, String&gt;&gt; wordNPlays = largeWords.apply("CombinePlays", Combine.perKey(new ConcatWordsCombineFunction()));
              //using CombineFn instead
              PCollection&lt;KV&lt;String, String&gt;&gt; wordNPlays = largeWords.apply("CombinePlays",Combine.perKey(new ConcatWordsCombineFn()));
              wordNPlays.setCoder(KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()));
              PCollection&lt;TableRow&gt; rows = wordNPlays.apply("FormatToRow", ParDo.of(new FormatShakespeareOutputFn()));
              return rows;
      }
  }
  
  </code></pre>
  
  <p>If I'm not adding this line in the code above</p>
  
  <pre class="lang-java prettyprint-override"><code> wordNPlays.setCoder(KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()));
  </code></pre>
  
  <p>I'm having an exception</p>
  
  <pre><code>Exception in thread "main" java.lang.IllegalStateException: Unable to return a default Coder for ExtractAllPlaysOfWords/CombinePlays/Combine.GroupedValues/ParDo(Anonymous)/ParMultiDo(Anonymous).output [PCollection]. Correct one of the following root causes:
            No Coder has been manually specified;  you may do so using .setCoder().
            Inferring a Coder from the CoderRegistry failed: Cannot provide coder for parameterized type org.apache.beam.sdk.values.KV&lt;K, OutputT&gt;: Unable to provide a Coder for K.
            Building a Coder using a registered CoderProvider failed.
            See suppressed exceptions for detailed failures.
            Using the default output Coder from the producing PTransform failed: PTransform.getOutputCoder called.
                  at org.apache.beam.vendor.guava.v20_0.com.google.common.base.Preconditions.checkState(Preconditions.java:444)
                  at org.apache.beam.sdk.values.PCollection.getCoder(PCollection.java:278)
                  at org.apache.beam.sdk.values.PCollection.finishSpecifying(PCollection.java:115)
                  at org.apache.beam.sdk.runners.TransformHierarchy.finishSpecifyingInput(TransformHierarchy.java:191)
                  at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:536)
                  at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:488)
                  at org.apache.beam.sdk.values.PCollection.apply(PCollection.java:370)
                  at me.examples.PlaysForWord.expand(PlaysForWord.java:21)
                  at me.examples.PlaysForWord.expand(PlaysForWord.java:10)
                  at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:537)
                  at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:488)
                  at org.apache.beam.sdk.values.PCollection.apply(PCollection.java:370)
                  at me.examples.Main.main(Main.java:41)
  </code></pre>
  
  <p>From the stacktrace, I think the pipeline is not able to get a coder for the <code>String</code> type of the KV obejct. Why is that ? Isn't supposed to be a "known" type for Apache Beam. Why is it working without specifying the coder when using the subclass of <code>SerializableFunction</code> in the <code>Combine.perKey</code>?</p>
  
  <p>In addition to that, when I tried to get the default coder for String from the coder registry I get <code>StringUTF8Coder</code></p>
  
  <pre class="lang-java prettyprint-override"><code>Coder coder = null;
         try {
             coder = pipeline.getCoderRegistry().getCoder(String.class);
             logger.info("coder is " + coder);
         } catch (Exception e){
             logger.info("exception "+ e.getMessage() +"\n coder is " + coder );
         }
  /*result
  INFO: coder is StringUtf8Coder
  */
  </code></pre>
  
  <p>I used Apache Beam 2.12.0 and run it on Google Dataflow </p>