{"record":{"id":"7ac518e5aafec81a","repo":"apache/beam","slug":"groupbykey-requires-its-input-to-use-kvcoder","errorCode":null,"errorMessage":"GroupByKey requires its input to use KvCoder","messagePattern":"GroupByKey requires its input to use KvCoder","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java","lineNumber":298,"sourceCode":"\n    // This primitive operation groups by the combination of key and window,\n    // merging windows as needed, using the windows assigned to the\n    // key/value input elements and the window merge operation of the\n    // window function associated with the input PCollection.\n    return PCollection.createPrimitiveOutputInternal(\n        input.getPipeline(),\n        updateWindowingStrategy(input.getWindowingStrategy()),\n        input.isBounded(),\n        getOutputKvCoder(input.getCoder()));\n  }\n\n  /**\n   * Returns the {@code Coder} of the input to this transform, which should be a {@code KvCoder}.\n   */\n  @SuppressWarnings(\"unchecked\")\n  static <K, V> KvCoder<K, V> getInputKvCoder(Coder<?> inputCoder) {\n    if (!(inputCoder instanceof KvCoder)) {\n      throw new IllegalStateException(\"GroupByKey requires its input to use KvCoder\");\n    }\n    return (KvCoder<K, V>) inputCoder;\n  }\n\n  /////////////////////////////////////////////////////////////////////////////\n\n  /**\n   * Returns the {@code Coder} of the keys of the input to this transform, which is also used as the\n   * {@code Coder} of the keys of the output of this transform.\n   */\n  public static <K, V> Coder<K> getKeyCoder(Coder<KV<K, V>> inputCoder) {\n    return GroupByKey.<K, V>getInputKvCoder(inputCoder).getKeyCoder();\n  }\n\n  /** Returns the {@code Coder} of the values of the input to this transform. */\n  public static <K, V> Coder<V> getInputValueCoder(Coder<KV<K, V>> inputCoder) {\n    return GroupByKey.<K, V>getInputKvCoder(inputCoder).getValueCoder();\n  }","sourceCodeStart":280,"sourceCodeEnd":316,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java#L280-L316","documentation":"GroupByKey's coder handling assumes the input PCollection's coder is a KvCoder<K,V> so it can extract key/value coders. getInputKvCoder() type-checks the coder and throws IllegalStateException if it is any other Coder, since grouping is meaningless without KV-shaped coders.","triggerScenarios":"Applying GroupByKey to a PCollection whose coder was set (or inferred) to something other than KvCoder — e.g. input.setCoder(SomeCustomCoder) on a KV-typed collection, an in-memory/create() source with a mismatched coder, or an intermediate transform that replaced the coder.","commonSituations":"Overriding coders on KV collections for debugging; using PCollectionList/PBegin glue that lost the KvCoder; consuming from a source that set a generic SerializableCoder for KV values; grouping after a transform that returned a different logical type but kept KV generics.","solutions":["Remove the erroneous setCoder() call so Beam infers KvCoder from the KV generics","Explicitly set the correct coder: input.setCoder(KvCoder.of(keyCoder, valueCoder))","If the element type genuinely isn't KV<K,V>, add a MapElements step producing KV before GroupByKey","Check upstream transforms (e.g. custom sources) for coder overrides that replaced the KvCoder"],"exampleFix":"// before\nkvCollection.setCoder(SerializableCoder.of(KV.class));\nkvCollection.apply(GroupByKey.create());\n// after\nkvCollection.setCoder(KvCoder.of(StringUtf8Coder.of(), valueCoder));\nkvCollection.apply(GroupByKey.create());","handlingStrategy":"type-guard","validationCode":"if (!(input.getCoder() instanceof KvCoder)) {\n  input.setCoder(KvCoder.of(keyCoder, valueCoder));\n}","typeGuard":"static <K,V> boolean hasKvCoder(PCollection<KV<K,V>> pc) {\n  return pc.getCoder() instanceof KvCoder<?, ?>;\n}","tryCatchPattern":"try {\n  grouped = input.apply(GroupByKey.create());\n} catch (IllegalStateException e) {\n  if (e.getMessage().contains(\"requires its input to use KvCoder\")) { /* fix input coder */ }\n  else throw e;\n}","preventionTips":["Don't override coders on KV collections casually","Set coders via KvCoder.of(...) only","Check upstream custom sources for coder substitutions"],"tags":["apache-beam","java","coder","groupbykey","type-mismatch"],"backgroundTag":"incompatible-source-type","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}