apache/beam · error · IllegalStateException

Could not decode bytes as message

Error message

Could not decode bytes as message

What it means

RowMessages.bytesToRowFn wraps protobuf decoding: the raw bytes are handed to a fromBytes function (e.g. message.parseFrom). Any failure there is rethrown as an IllegalStateException with 'Could not decode bytes as message'.

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/RowMessages.java:70

  private static final class BytesToRowFn<T> extends SimpleFunction<byte[], Row> {

    private final ProcessFunction<byte[], ? extends T> fromBytesFn;
    private final SerializableFunction<T, Row> toRowFn;

    private BytesToRowFn(
        ProcessFunction<byte[], ? extends T> fromBytesFn, SerializableFunction<T, Row> toRowFn) {
      this.fromBytesFn = fromBytesFn;
      this.toRowFn = toRowFn;
    }

    @Override
    public Row apply(byte[] bytes) {
      final T message;
      try {
        message = fromBytesFn.apply(bytes);
      } catch (Exception e) {
        throw new IllegalStateException("Could not decode bytes as message", e);
      }
      return toRowFn.apply(message);
    }
  }

  public static <T> SimpleFunction<Row, byte[]> rowToBytesFn(
      SchemaProvider schemaProvider,
      TypeDescriptor<T> typeDescriptor,
      ProcessFunction<? super T, byte[]> toBytesFn) {
    final Schema schema = checkArgumentNotNull(schemaProvider.schemaFor(typeDescriptor));
    final SerializableFunction<Row, T> fromRowFn =
        checkArgumentNotNull(schemaProvider.fromRowFunction(typeDescriptor));
    toBytesFn = checkArgumentNotNull(toBytesFn);
    return new RowToBytesFn<>(schema, fromRowFn, toBytesFn);
  }

  public static <T> SimpleFunction<Row, byte[]> rowToBytesFn(
      SchemaProvider schemaProvider, TypeDescriptor<T> typeDescriptor, Coder<? super T> coder) {

View on GitHub (pinned to 12126d8942)

Solutions

1. Verify the byte source actually contains serialized messages of the expected proto type and version.
2. Check that writer and reader use compatible .proto definitions (reserve/never reuse field numbers, avoid breaking changes).
3. Validate/inspect bytes with message.getParser().parseFrom in a test to see the underlying InvalidProtocolBufferException.

Example fix

// before
PCollection<Row> rows = bytes.apply(MapElements.via(RowMessages.bytesToRowFn(Msg.parser())));

// after
PCollection<Row> rows = bytes
  .apply("ValidateProto", MapElements.into(TypeDescriptor.of(byte[].class)).via(b -> {
     try { Msg.getDescriptor(); return b; } catch (Exception e) { throw new IllegalArgumentException("bad proto bytes", e); }
  }))
  .apply(MapElements.via(RowMessages.bytesToRowFn(Msg.parser())));

// Java: pre-validate bytes parse as the expected message
public static boolean isValidMsg(byte[] b) {
  try { Msg.parseFrom(b); return true; } catch (Exception e) { return false; }
}

// Java
try {
  Row row = bytesToRowFn.apply(bytes);
} catch (IllegalStateException e) {
  // inspect e.getCause() (InvalidProtocolBufferException); route to dead-letter PCollection
}

When it happens

Trigger: Converting byte[] elements via ProtoCoder-derived functions when the bytes are not a valid serialization of the expected protobuf message type.

Common situations: Reading from a topic/file written with a different proto schema or entirely non-protobuf data; schema evolved incompatibly between writer and reader; corrupt payload.

Understand the failure class

Background: "cannot parse invalid wire-format data", "cannot unmarshal", "failed unmarshalling": protobuf unmarshal errors explained — this error's family across 10 libraries.

Related errors

• Could not encode message as bytes
• Failed to decode Schema due to an error decoding Field proto
• Encountered UNSPECIFIED AtomicType
• Unexpected type_info: +protoFieldType.getTypeInfoCase()
• FieldTypeValue has no value but the field cannot be null.

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