apache/beam · error · IllegalStateException

GroupByEncryptedKey requires its input to use KvCoder

Error message

GroupByEncryptedKey requires its input to use KvCoder

What it means

GroupByEncryptedKey needs the key coder to inspect and verify key encoding (determinism check, key encryption). If the input PCollection<KV<K,V>> coder is not a KvCoder, it cannot obtain the key coder and expand() throws IllegalStateException.

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByEncryptedKey.java:105

   * @param gbk The custom GBK transform to use in the middle of the GBEK.
   * @param <K> The type of the keys in the input PCollection.
   * @param <V> The type of the values in the input PCollection.
   * @return A {@link GroupByEncryptedKey} transform.
   */
  public static <K, V> GroupByEncryptedKey<K, V> createWithCustomGbk(
      Secret hmacKey,
      PTransform<
              PCollection<KV<byte[], KV<byte[], byte[]>>>,
              PCollection<KV<byte[], Iterable<KV<byte[], byte[]>>>>>
          gbk) {
    return new GroupByEncryptedKey<>(hmacKey, gbk);
  }

  @Override
  public PCollection<KV<K, Iterable<V>>> expand(PCollection<KV<K, V>> input) {
    Coder<KV<K, V>> inputCoder = input.getCoder();
    if (!(inputCoder instanceof KvCoder)) {
      throw new IllegalStateException("GroupByEncryptedKey requires its input to use KvCoder");
    }
    KvCoder<K, V> inputKvCoder = (KvCoder<K, V>) inputCoder;
    Coder<K> keyCoder = inputKvCoder.getKeyCoder();

    try {
      keyCoder.verifyDeterministic();
    } catch (NonDeterministicException e) {
      throw new IllegalStateException(
          "the keyCoder of a GroupByEncryptedKey must be deterministic", e);
    }

    Coder<V> valueCoder = inputKvCoder.getValueCoder();

    PCollection<KV<byte[], Iterable<KV<byte[], byte[]>>>> grouped =
        input
            .apply(
                "EncryptMessage",
                ParDo.of(new EncryptMessage<>(this.hmacKey, keyCoder, valueCoder)))

View on GitHub (pinned to 12126d8942)

Solutions

1. Ensure the input is a standard KV PCollection with inferred KvCoder (remove explicit setCoder calls)
2. Explicitly set KvCoder: input.setCoder(KvCoder.of(keyCoder, valueCoder)) before applying GroupByEncryptedKey
3. Verify the upstream transform produces KV<K,V> with proper generics so coder inference yields KvCoder

Example fix

// before
input.setCoder(new MyCustomCoder<KV<K,V>>());
input.apply(GroupByEncryptedKey.create());
// after
input.setCoder(KvCoder.of(keyCoder, valueCoder));
input.apply(GroupByEncryptedKey.create());

// Verify KvCoder before applying GroupByEncryptedKey
if (!(input.getCoder() instanceof KvCoder)) {
  throw new IllegalStateException("input must use KvCoder");
}

boolean hasKvCoder(PCollection<KV<K,V>> in) {
  return in.getCoder() instanceof KvCoder;
}

try { out = input.apply(GroupByEncryptedKey.create()); }
catch (IllegalStateException e) { // set KvCoder and retry
  input.setCoder(KvCoder.of(keyCoder, valueCoder));
  out = input.apply(GroupByEncryptedKey.create());
}

When it happens

Trigger: Applying GroupByEncryptedKey to a PCollection whose coder was overridden with a non-KvCoder, or whose upstream transform inferred something other than KvCoder for the KV elements.

Common situations: Manual setCoder with a custom coder; custom sources or transforms emitting KV with a bespoke coder; raw KV types erasing the coder inference.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• the GroupByKey requires its output coder to be %s but found
• cannot encode a null Integer
• cannot encode a null Long
• cannot encode a null Short
• cannot encode a null BitSet

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