apache/beam · error · IllegalStateException

Combine.GroupedValues requires its input to use KvCoder

Error message

Combine.GroupedValues requires its input to use KvCoder

What it means

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.

Solutions

1. Ensure the input comes from GroupByKey (or a transform that yields KvCoder) rather than a manually-coded collection.
2. Remove any explicit setCoder(...) on the input that replaces the KvCoder.
3. If using a custom coder, wrap the key/value coders with KvCoder.of(keyCoder, valueCoder) so the input is a KvCoder.
4. Verify with pipeline.getCoderRegistry() / pCollection.getCoder() that the input coder is a KvCoder before applying GroupedValues.

Example fix

// before
PCollection<KV<String, Iterable<Integer>>> grouped = ...;
grouped.setCoder(ListCoder.of(...)); // not KvCoder
grouped.apply(Combine.groupedValues(sumFn));
// after
PCollection<KV<String, Iterable<Integer>>> grouped = keyed.apply(GroupByKey.create()); // yields KvCoder
PCollection<KV<String, Integer>> summed = grouped.apply(Combine.groupedValues(sumFn));

if (!(grouped.getCoder() instanceof KvCoder)) { throw new IllegalStateException("GroupedValues input must have KvCoder; got " + grouped.getCoder()); }

boolean hasKvCoder(PCollection<?> pc) { return pc.getCoder() instanceof KvCoder; }

try { grouped.apply(Combine.groupedValues(fn)); } catch (IllegalStateException e) { if (e.getMessage().contains("KvCoder")) { /* drop manual setCoder or wrap with KvCoder.of */ } throw e; }

When it happens

Trigger: 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.

Common situations: Manual coder overrides after a GroupByKey; custom sources producing KV coder-less collections; pipelines where schema/coder inference silently picked the wrong coder type.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

• Combine.GroupedValues requires its input values to use…
• ApproximateUnique.PerKey requires its input to use KvCoder
• AvroCoder for GenericRecord requires a schema
• cannot encode a null BitSet
• cannot encode a null byte[]

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/de59ce16a2bf630e. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java:2290

    }

    /**
     * Returns the {@link CombineFn} bound to its coders.
     *
     * <p>For internal use.
     */
    public AppliedCombineFn<? super K, ? super InputT, ?, OutputT> getAppliedFn(
        CoderRegistry registry,
        Coder<? extends KV<K, ? extends Iterable<InputT>>> inputCoder,
        WindowingStrategy<?, ?> windowingStrategy) {
      KvCoder<K, InputT> kvCoder = getKvCoder(inputCoder);
      return AppliedCombineFn.withInputCoder(fn, registry, kvCoder, sideInputs, windowingStrategy);
    }

    private KvCoder<K, InputT> getKvCoder(
        Coder<? extends KV<K, ? extends Iterable<InputT>>> inputCoder) {
      if (!(inputCoder instanceof KvCoder)) {
        throw new IllegalStateException("Combine.GroupedValues requires its input to use KvCoder");
      }
      @SuppressWarnings({"unchecked", "rawtypes"})
      KvCoder<K, ? extends Iterable<InputT>> kvCoder = (KvCoder) inputCoder;
      Coder<K> keyCoder = kvCoder.getKeyCoder();
      Coder<? extends Iterable<InputT>> kvValueCoder = kvCoder.getValueCoder();
      if (!(kvValueCoder instanceof IterableCoder)) {
        throw new IllegalStateException(
            "Combine.GroupedValues requires its input values to use " + "IterableCoder");
      }
      @SuppressWarnings("unchecked")
      IterableCoder<InputT> inputValuesCoder = (IterableCoder<InputT>) kvValueCoder;
      Coder<InputT> inputValueCoder = inputValuesCoder.getElemCoder();
      return KvCoder.of(keyCoder, inputValueCoder);
    }

    @Override
    public void populateDisplayData(DisplayData.Builder builder) {
      super.populateDisplayData(builder);

View on GitHub (pinned to 12126d8942)