{"record":{"id":"34dffad8baf2af73","repo":"apache/beam","slug":"could-not-decode-bytes-as-message","errorCode":null,"errorMessage":"Could not decode bytes as message","messagePattern":"Could not decode bytes as message","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/RowMessages.java","lineNumber":70,"sourceCode":"\n  private static final class BytesToRowFn<T> extends SimpleFunction<byte[], Row> {\n\n    private final ProcessFunction<byte[], ? extends T> fromBytesFn;\n    private final SerializableFunction<T, Row> toRowFn;\n\n    private BytesToRowFn(\n        ProcessFunction<byte[], ? extends T> fromBytesFn, SerializableFunction<T, Row> toRowFn) {\n      this.fromBytesFn = fromBytesFn;\n      this.toRowFn = toRowFn;\n    }\n\n    @Override\n    public Row apply(byte[] bytes) {\n      final T message;\n      try {\n        message = fromBytesFn.apply(bytes);\n      } catch (Exception e) {\n        throw new IllegalStateException(\"Could not decode bytes as message\", e);\n      }\n      return toRowFn.apply(message);\n    }\n  }\n\n  public static <T> SimpleFunction<Row, byte[]> rowToBytesFn(\n      SchemaProvider schemaProvider,\n      TypeDescriptor<T> typeDescriptor,\n      ProcessFunction<? super T, byte[]> toBytesFn) {\n    final Schema schema = checkArgumentNotNull(schemaProvider.schemaFor(typeDescriptor));\n    final SerializableFunction<Row, T> fromRowFn =\n        checkArgumentNotNull(schemaProvider.fromRowFunction(typeDescriptor));\n    toBytesFn = checkArgumentNotNull(toBytesFn);\n    return new RowToBytesFn<>(schema, fromRowFn, toBytesFn);\n  }\n\n  public static <T> SimpleFunction<Row, byte[]> rowToBytesFn(\n      SchemaProvider schemaProvider, TypeDescriptor<T> typeDescriptor, Coder<? super T> coder) {","sourceCodeStart":52,"sourceCodeEnd":88,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/RowMessages.java#L52-L88","documentation":"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'.","triggerScenarios":"Converting byte[] elements via ProtoCoder-derived functions when the bytes are not a valid serialization of the expected protobuf message type.","commonSituations":"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.","solutions":["Verify the byte source actually contains serialized messages of the expected proto type and version.","Check that writer and reader use compatible .proto definitions (reserve/never reuse field numbers, avoid breaking changes).","Validate/inspect bytes with message.getParser().parseFrom in a test to see the underlying InvalidProtocolBufferException."],"exampleFix":"// before\nPCollection<Row> rows = bytes.apply(MapElements.via(RowMessages.bytesToRowFn(Msg.parser())));\n\n// after\nPCollection<Row> rows = bytes\n  .apply(\"ValidateProto\", MapElements.into(TypeDescriptor.of(byte[].class)).via(b -> {\n     try { Msg.getDescriptor(); return b; } catch (Exception e) { throw new IllegalArgumentException(\"bad proto bytes\", e); }\n  }))\n  .apply(MapElements.via(RowMessages.bytesToRowFn(Msg.parser())));","handlingStrategy":"try-catch","validationCode":"// Java: pre-validate bytes parse as the expected message\npublic static boolean isValidMsg(byte[] b) {\n  try { Msg.parseFrom(b); return true; } catch (Exception e) { return false; }\n}","typeGuard":null,"tryCatchPattern":"// Java\ntry {\n  Row row = bytesToRowFn.apply(bytes);\n} catch (IllegalStateException e) {\n  // inspect e.getCause() (InvalidProtocolBufferException); route to dead-letter PCollection\n}","preventionTips":["Keep writer and reader proto schemas compatible; never reuse field numbers.","Dead-letter unparseable byte payloads instead of failing the pipeline."],"tags":["java","protobuf","deserialization","schema"],"backgroundTag":"protobuf-unmarshal-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}