apache/beam · error · CoderException

Cannot decode object from input stream.

Error message

Cannot decode object from input stream.

What it means

KryoCoder.decode reads a serialized object from the input stream using Kryo's readClassAndObject. If Kryo encounters malformed, truncated, or incompatible data in the stream it throws a KryoException, which is wrapped in a Beam CoderException with this message. It means the bytes on the wire cannot be decoded into an object by the configured Kryo instance.

Solutions

1. Ensure the same KryoCoder/KryoOptions (registrations, serializers) is used on both the encoding and decoding side of the PCollection.
2. Verify the input stream contains data actually encoded by KryoCoder.encode (readClassAndObject/writeClassAndObject pair).
3. Check for stream truncation or corruption in the source (e.g. shuffled/serialized bytes in the runner).
4. Explicitly set the coder with PCollection.setCoder(KryoCoder.of(options)) if type erasure caused coder inference to pick the wrong coder.

Example fix

// before: relying on inferred coder after changing Kryo options
PCollection<MyType> out = in.apply(...);
// after: pin the coder so encode/decode use the same Kryo configuration
PCollection<MyType> out = in.apply(...).setCoder(KryoCoder.of(KryoOptions.withRegistrations(regs)));

if (bytes == null || bytes.length == 0) throw new IllegalArgumentException("Empty payload for Kryo decode");

try { T obj = coder.decode(stream); } catch (CoderException e) { LOG.error("Kryo decode failed; check coder/config match", e); throw new UnrecoverableException(e); }

When it happens

Trigger: Decoding a PCollection element whose bytes were not written by the matching KryoCoder (encode/decode coder mismatch), a truncated or corrupted input stream, or reading data encoded with a different Kryo registration/serialization configuration than the one used at decode time.

Common situations: Changing KryoOptions (registrationRequired, custom serializers, registrations) between the job that encoded data and the one decoding it; re-reading old files/checkpoints written by a previous serializer version; incorrect coder inference causing a mismatched coder on a PCollection.

Related errors

• cannot encode a null BitSet
• cannot encode a null byte[]
• cannot encode a null Integer
• cannot encode a null Long
• cannot encode a null Short

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

Appendix: source

Thrown at sdks/java/extensions/kryo/src/main/java/org/apache/beam/sdk/extensions/kryo/KryoCoder.java:231

        if (message.startsWith("Class is not registered")) {
          throw new CoderException(message);
        }
      }
      throw e;
    }
  }

  @Override
  public T decode(InputStream inStream) throws IOException {
    final KryoState kryoState = KryoState.get(this);
    final InputChunked inputChunked = kryoState.getInputChunked();
    inputChunked.setInputStream(inStream);
    try {
      @SuppressWarnings("unchecked")
      final T instance = (T) kryoState.getKryo().readClassAndObject(inputChunked);
      return instance;
    } catch (KryoException e) {
      throw new CoderException("Cannot decode object from input stream.", e);
    }
  }

  @Override
  public void verifyDeterministic() throws NonDeterministicException {
    // noop
  }

  /**
   * Create a new {@link KryoCoder} instance with the user provided registrar.
   *
   * @param registrar registrar to append to list of already registered registrars.
   * @return new kryo coder
   */
  public KryoCoder<T> withRegistrar(KryoRegistrar registrar) {
    final List<KryoRegistrar> newRegistrars = new ArrayList<>(registrars);
    registrars.add(registrar);
    return new KryoCoder<>(options, newRegistrars);

View on GitHub (pinned to 12126d8942)