{"record":{"id":"8662beb56d7bb081","repo":"apache/beam","slug":"failed-to-parse-the-proto-bytes-to-changestreamrecord-proto","errorCode":null,"errorMessage":"Failed to parse the proto bytes to ChangeStreamRecord proto","messagePattern":"Failed to parse the proto bytes to ChangeStreamRecord proto","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/ChangeStreamResultSet.java","lineNumber":151,"sourceCode":"  }\n\n  /**\n   * Returns the change stream record at the current pointer by parsing the bytes column. It also\n   * updates the timestamp at which the record was read.\n   *\n   * <p>Should only be used for PostgreSQL databases when the change stream record is delivered as\n   * proto bytes.\n   *\n   * @return a change stream record as a proto or null\n   */\n  public com.google.spanner.v1.ChangeStreamRecord getBytes(int index) {\n    recordReadAt = Timestamp.now();\n    try {\n      // Use getBytes(0) for the BYTES column returned by read_proto_bytes_ TVF\n      return com.google.spanner.v1.ChangeStreamRecord.parseFrom(\n          resultSet.getBytes(index).toByteArray());\n    } catch (InvalidProtocolBufferException e) {\n      throw new RuntimeException(\"Failed to parse the proto bytes to ChangeStreamRecord proto\", e);\n    }\n  }\n\n  /** Returns true if the result set at the current pointer contain only one bytes change record. */\n  public boolean isProtoBytesChangeRecord() {\n    return resultSet.getColumnCount() == 1\n        && !resultSet.isNull(0)\n        && resultSet.getColumnType(0).getCode() == com.google.cloud.spanner.Type.Code.BYTES;\n  }\n\n  /**\n   * Returns the record at the current pointer as {@link JsonB}. It also updates the timestamp at\n   * which the record was read.\n   *\n   * <p>If {@link ChangeStreamResultSet#next()} was not called or if it was called but there are no\n   * more records in the stream, null will be returned.\n   *\n   * <p>Should only be used for PostgreSQL databases.","sourceCodeStart":133,"sourceCodeEnd":169,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/ChangeStreamResultSet.java#L133-L169","documentation":"ChangeStreamResultSet.getBytes parses the single BYTES column returned by the read_proto_bytes_ TVF into a ChangeStreamRecord proto via parseFrom. If the bytes are not a valid protobuf payload (InvalidProtocolBufferException), it wraps the failure in this RuntimeException.","triggerScenarios":"The BYTES column read from the change stream TVF does not contain a serialized ChangeStreamRecord — corrupt/truncated row data, reading the wrong column index, or a protocol/version mismatch between the connector and the Spanner TVF output.","commonSituations":"Connector version older than the TVF output format (Spanner changed the change stream serialization); reading a non-proto result set as if it were proto bytes; data corruption or manual edits to metadata tables.","solutions":["Upgrade the Beam Spanner connector to match your Spanner change stream TVF/proto format version.","Verify the query targets the read_proto_bytes_ TVF BYTES column (index 0) and that isProtoBytesChangeRecord() is checked before calling getBytes.","Re-read the affected change stream partition; if corruption persists, reset the partition from a known-good start timestamp.","Inspect the raw bytes for a known protobuf header; if the format differs, the TVF and client proto definitions are out of sync — update google.spanner.v1 protos."],"exampleFix":"// before\nif (resultSet.getColumnCount() > 0) { record = resultSet.getBytes(0); ... }\n// after\nif (isProtoBytesChangeRecord()) {\n  record = changeStreamResultSet.getBytes(0); // safe: column is a proto BYTES payload\n}","handlingStrategy":"try-catch","validationCode":"if (!changeStreamResultSet.isProtoBytesChangeRecord()) {\n  throw new IllegalStateException(\"Row is not a proto-bytes change record\");\n}","typeGuard":null,"tryCatchPattern":"try { record = resultSet.getBytes(0); } catch (RuntimeException e) {\n  log.error(\"Corrupt change stream proto bytes\", e);\n  // skip row or reset partition\n}","preventionTips":["Keep connector protos in sync with the Spanner TVF output version.","Check isProtoBytesChangeRecord() before parsing bytes.","Never manually modify change stream metadata tables."],"tags":["spanner","protobuf","deserialization","java"],"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-14T21:17:11.552Z"}