{"record":{"id":"8c3978cdcaf384e9","repo":"apache/beam","slug":"pcollection-does-not-use-a-kvcoder","errorCode":null,"errorMessage":"PCollection does not use a KvCoder","messagePattern":"PCollection does not use a KvCoder","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/CoGroupByKey.java","lineNumber":139,"sourceCode":"    PCollection<KV<K, CoGbkResult>> result =\n        groupedTable.apply(\n            \"ConstructCoGbkResultFn\", ParDo.of(new ConstructCoGbkResultFn<>(tupleTags)));\n    result.setCoder(KvCoder.of(keyCoder, CoGbkResultCoder.of(tupleTags, unionCoder)));\n\n    return result;\n  }\n\n  //////////////////////////////////////////////////////////////////////////////\n\n  /**\n   * Returns the value coder for the given PCollection. Assumes that the value coder is an instance\n   * of {@code KvCoder<K, V>}.\n   */\n  private <V> Coder<V> getValueCoder(PCollection<KV<K, V>> pCollection) {\n    // Assumes that the PCollection uses a KvCoder.\n    Coder<?> entryCoder = pCollection.getCoder();\n    if (!(entryCoder instanceof KvCoder<?, ?>)) {\n      throw new IllegalArgumentException(\"PCollection does not use a KvCoder\");\n    }\n    @SuppressWarnings(\"unchecked\")\n    KvCoder<K, V> coder = (KvCoder<K, V>) entryCoder;\n    return coder.getValueCoder();\n  }\n\n  /**\n   * Returns a UnionTable for the given input PCollection, using the given union index and the given\n   * unionTableEncoder.\n   */\n  private <V> PCollection<KV<K, RawUnionValue>> makeUnionTable(\n      final int index,\n      PCollection<KV<K, V>> pCollection,\n      KvCoder<K, RawUnionValue> unionTableEncoder) {\n\n    return pCollection\n        .apply(\"MakeUnionTable\" + index, ParDo.of(new ConstructUnionTableFn<>(index)))\n        .setCoder(unionTableEncoder);","sourceCodeStart":121,"sourceCodeEnd":157,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/CoGroupByKey.java#L121-L157","documentation":"CoGroupByKey needs the value coder of each input to build the union coder, and it assumes every input PCollection's coder is a KvCoder. If a PCollection's coder is not a KvCoder instance, the assumption is violated and an IllegalArgumentException is thrown.","triggerScenarios":"Passing a PCollection<KV<K,V>> whose coder was not inferred as KvCoder, e.g. after setCoder() with a non-KvCoder or a custom source that produced an unspecified coder.","commonSituations":"Custom I/O or PTransforms that set explicit wrong coders; using coders.setCoder with the wrong type; older Beam versions where coder inference differed.","solutions":["Ensure each input PCollection has a KvCoder via KvCoder.of(keyCoder, valueCoder)","Call pc.setCoder(KvCoder.of(...)) explicitly on inputs before the join","Verify coder inference upstream — the transform producing the KV should set a KvCoder"],"exampleFix":"// before\npc.apply(CoGroupByKey.create()); // coder not KvCoder\n// after\npc.setCoder(KvCoder.of(keyCoder, valueCoder));\npc.apply(CoGroupByKey.create());","handlingStrategy":"type-guard","validationCode":"if (!(pc.getCoder() instanceof KvCoder)) { throw new IllegalArgumentException(\"Input must have a KvCoder\"); }","typeGuard":"Coder<?> c = pc.getCoder();\nif (c instanceof KvCoder<?, ?> kvCoder) { /* use kvCoder */ }","tryCatchPattern":"try { coder = getValueCoder(pc); } catch (IllegalArgumentException e) { pc.setCoder(KvCoder.of(defaultKeyCoder, defaultValueCoder)); }","preventionTips":["Always set explicit KvCoder on KV PCollections entering joins","Verify coder inference of upstream transforms","Avoid setCoder with non-KvCoder on KV collections"],"tags":["beam","kvcoder","coder","cogroupbykey"],"backgroundTag":"type-mismatch","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"}