{"record":{"id":"de59ce16a2bf630e","repo":"apache/beam","slug":"combine-groupedvalues-requires-its-input-to-use-kvcoder","errorCode":null,"errorMessage":"Combine.GroupedValues requires its input to use KvCoder","messagePattern":"Combine\\.GroupedValues 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/Combine.java","lineNumber":2290,"sourceCode":"    }\n\n    /**\n     * Returns the {@link CombineFn} bound to its coders.\n     *\n     * <p>For internal use.\n     */\n    public AppliedCombineFn<? super K, ? super InputT, ?, OutputT> getAppliedFn(\n        CoderRegistry registry,\n        Coder<? extends KV<K, ? extends Iterable<InputT>>> inputCoder,\n        WindowingStrategy<?, ?> windowingStrategy) {\n      KvCoder<K, InputT> kvCoder = getKvCoder(inputCoder);\n      return AppliedCombineFn.withInputCoder(fn, registry, kvCoder, sideInputs, windowingStrategy);\n    }\n\n    private KvCoder<K, InputT> getKvCoder(\n        Coder<? extends KV<K, ? extends Iterable<InputT>>> inputCoder) {\n      if (!(inputCoder instanceof KvCoder)) {\n        throw new IllegalStateException(\"Combine.GroupedValues requires its input to use KvCoder\");\n      }\n      @SuppressWarnings({\"unchecked\", \"rawtypes\"})\n      KvCoder<K, ? extends Iterable<InputT>> kvCoder = (KvCoder) inputCoder;\n      Coder<K> keyCoder = kvCoder.getKeyCoder();\n      Coder<? extends Iterable<InputT>> kvValueCoder = kvCoder.getValueCoder();\n      if (!(kvValueCoder instanceof IterableCoder)) {\n        throw new IllegalStateException(\n            \"Combine.GroupedValues requires its input values to use \" + \"IterableCoder\");\n      }\n      @SuppressWarnings(\"unchecked\")\n      IterableCoder<InputT> inputValuesCoder = (IterableCoder<InputT>) kvValueCoder;\n      Coder<InputT> inputValueCoder = inputValuesCoder.getElemCoder();\n      return KvCoder.of(keyCoder, inputValueCoder);\n    }\n\n    @Override\n    public void populateDisplayData(DisplayData.Builder builder) {\n      super.populateDisplayData(builder);","sourceCodeStart":2272,"sourceCodeEnd":2308,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java#L2272-L2308","documentation":"Combine.GroupedValues validates at pipeline-construction time that its input PCollection's coder is a KvCoder, because it must know key/value coders to combine grouped values. When the input coder is not a KvCoder (e.g. the coder was set manually or inference produced a generic coder), Beam throws IllegalStateException. This is an internal contract between GroupByKey (which produces KvCoder inputs) and GroupedValues.","triggerScenarios":"Applying Combine.GroupedValues to a PCollection<KV<K, Iterable<V>>> whose coder was explicitly set to a non-KvCoder (e.g. via PCollection.setCoder with a custom coder), or building the input without going through GroupByKey so coder inference does not yield KvCoder.","commonSituations":"Manual coder overrides after a GroupByKey; custom sources producing KV coder-less collections; pipelines where schema/coder inference silently picked the wrong coder type.","solutions":["Ensure the input comes from GroupByKey (or a transform that yields KvCoder) rather than a manually-coded collection.","Remove any explicit setCoder(...) on the input that replaces the KvCoder.","If using a custom coder, wrap the key/value coders with KvCoder.of(keyCoder, valueCoder) so the input is a KvCoder.","Verify with pipeline.getCoderRegistry() / pCollection.getCoder() that the input coder is a KvCoder before applying GroupedValues."],"exampleFix":"// before\nPCollection<KV<String, Iterable<Integer>>> grouped = ...;\ngrouped.setCoder(ListCoder.of(...)); // not KvCoder\ngrouped.apply(Combine.groupedValues(sumFn));\n// after\nPCollection<KV<String, Iterable<Integer>>> grouped = keyed.apply(GroupByKey.create()); // yields KvCoder\nPCollection<KV<String, Integer>> summed = grouped.apply(Combine.groupedValues(sumFn));","handlingStrategy":"type-guard","validationCode":"if (!(grouped.getCoder() instanceof KvCoder)) { throw new IllegalStateException(\"GroupedValues input must have KvCoder; got \" + grouped.getCoder()); }","typeGuard":"boolean hasKvCoder(PCollection<?> pc) { return pc.getCoder() instanceof KvCoder; }","tryCatchPattern":"try { grouped.apply(Combine.groupedValues(fn)); } catch (IllegalStateException e) { if (e.getMessage().contains(\"KvCoder\")) { /* drop manual setCoder or wrap with KvCoder.of */ } throw e; }","preventionTips":["Derive GroupedValues inputs from GroupByKey rather than hand-built collections","Avoid setCoder overrides on KV collections feeding GroupedValues","Log/inspect pc.getCoder() before applying coder-sensitive transforms"],"tags":["java","apache-beam","combine","coder"],"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-20T03:17:13.778Z"}