apache/beam · error · RuntimeException

Failed to parse the proto bytes to ChangeStreamRecord proto

Error message

Failed to parse the proto bytes to ChangeStreamRecord proto

What it means

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.

Source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/ChangeStreamResultSet.java:151

  }

  /**
   * Returns the change stream record at the current pointer by parsing the bytes column. It also
   * updates the timestamp at which the record was read.
   *
   * <p>Should only be used for PostgreSQL databases when the change stream record is delivered as
   * proto bytes.
   *
   * @return a change stream record as a proto or null
   */
  public com.google.spanner.v1.ChangeStreamRecord getBytes(int index) {
    recordReadAt = Timestamp.now();
    try {
      // Use getBytes(0) for the BYTES column returned by read_proto_bytes_ TVF
      return com.google.spanner.v1.ChangeStreamRecord.parseFrom(
          resultSet.getBytes(index).toByteArray());
    } catch (InvalidProtocolBufferException e) {
      throw new RuntimeException("Failed to parse the proto bytes to ChangeStreamRecord proto", e);
    }
  }

  /** Returns true if the result set at the current pointer contain only one bytes change record. */
  public boolean isProtoBytesChangeRecord() {
    return resultSet.getColumnCount() == 1
        && !resultSet.isNull(0)
        && resultSet.getColumnType(0).getCode() == com.google.cloud.spanner.Type.Code.BYTES;
  }

  /**
   * Returns the record at the current pointer as {@link JsonB}. It also updates the timestamp at
   * which the record was read.
   *
   * <p>If {@link ChangeStreamResultSet#next()} was not called or if it was called but there are no
   * more records in the stream, null will be returned.
   *
   * <p>Should only be used for PostgreSQL databases.

View on GitHub (pinned to 12126d8942)

Solutions

1. Upgrade the Beam Spanner connector to match your Spanner change stream TVF/proto format version.
2. Verify the query targets the read_proto_bytes_ TVF BYTES column (index 0) and that isProtoBytesChangeRecord() is checked before calling getBytes.
3. Re-read the affected change stream partition; if corruption persists, reset the partition from a known-good start timestamp.
4. 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.

Example fix

// before
if (resultSet.getColumnCount() > 0) { record = resultSet.getBytes(0); ... }
// after
if (isProtoBytesChangeRecord()) {
  record = changeStreamResultSet.getBytes(0); // safe: column is a proto BYTES payload
}

if (!changeStreamResultSet.isProtoBytesChangeRecord()) {
  throw new IllegalStateException("Row is not a proto-bytes change record");
}

try { record = resultSet.getBytes(0); } catch (RuntimeException e) {
  log.error("Corrupt change stream proto bytes", e);
  // skip row or reset partition
}

When it happens

Trigger: 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.

Common situations: 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.

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.

• Parsing and encoding errors: unexpected token, malformed input — why parsers reject input and how to find the real culprit.

Related errors

• Could not decode bytes as message
• Unexpected type_info: +protoFieldType.getTypeInfoCase()
• Unable to infer data schema from configuration proto.
• Could not parse Pub/Sub message
• Could not decode Pubsub message

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