apache/beam · error · IllegalArgumentException

Failed to parse record into proto: {row}

Error message

Failed to parse record into proto: {row}

What it means

ChangeStreamRecordMapper.toChangeStreamRecordJson parses a JSON row read from Spanner change stream metadata into a protobuf Value. If parser.merge fails (InvalidProtocolBufferException) the row is not valid JSON/proto-JSON, so it cannot be mapped into a ChangeStreamRecord and an IllegalArgumentException is thrown. This is an internal data-format invariant: rows read from Spanner must be valid proto-JSON.

Source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/mapper/ChangeStreamRecordMapper.java:479

            .filter(this::isNonNullHeartbeatRecord)
            .map(struct -> toHeartbeatRecord(partition, struct, resultSetMetadata));

    final Stream<ChildPartitionsRecord> childPartitionsRecords =
        row.getStructList(CHILD_PARTITIONS_RECORD_COLUMN).stream()
            .filter(this::isNonNullChildPartitionsRecord)
            .map(struct -> toChildPartitionsRecord(partition, struct, resultSetMetadata));

    return Stream.concat(
        Stream.concat(dataChangeRecords, heartbeatRecords), childPartitionsRecords);
  }

  ChangeStreamRecord toChangeStreamRecordJson(
      PartitionMetadata partition, String row, ChangeStreamResultSetMetadata resultSetMetadata) {
    Value.Builder valueBuilder = Value.newBuilder();
    try {
      this.parser.merge(row, valueBuilder);
    } catch (InvalidProtocolBufferException exc) {
      throw new IllegalArgumentException("Failed to parse record into proto: " + row);
    }
    Value value = valueBuilder.build();
    if (isNonNullDataChangeRecordJson(value)) {
      return toDataChangeRecordJson(partition, value, resultSetMetadata);
    } else if (isNonNullHeartbeatRecordJson(value)) {
      return toHeartbeatRecordJson(partition, value, resultSetMetadata);
    } else if (isNonNullChildPartitionsRecordJson(value)) {
      return toChildPartitionsRecordJson(partition, value, resultSetMetadata);
    } else {
      throw new IllegalArgumentException("Unknown change stream record type " + row);
    }
  }

  private boolean isNonNullDataChangeRecord(Struct row) {
    return !row.isNull(COMMIT_TIMESTAMP_COLUMN);
  }

  private boolean isNonNullDataChangeRecordJson(Value row) {

View on GitHub (pinned to 12126d8942)

Solutions

1. Validate the row is well-formed JSON (e.g. with a JSON parser) before feeding it to the mapper.
2. Check the source of the row string: confirm it was produced by the same Spanner change stream format the mapper expects.
3. Verify Beam google-cloud-platform SDK version matches the data written by your pipeline version.
4. Wrap toChangeStreamRecords in try-catch to log and skip/transform the malformed row instead of failing the pipeline.
5. Inspect the offending row printed in the message for truncation, encoding, or manual modification.

Example fix

// before
ChangeStreamRecord record = mapper.toChangeStreamRecordJson(partition, rawRow, metadata);

// after
new ObjectMapper().readTree(rawRow); // throws early with a clear parse error if invalid
ChangeStreamRecord record = mapper.toChangeStreamRecordJson(partition, rawRow, metadata);

boolean isProtoJsonRow(String row) {
  try {
    com.google.protobuf.util.JsonFormat.parser().ignoringUnknownFields().merge(row,
        com.google.protobuf.Value.newBuilder());
    return true;
  } catch (Exception e) {
    return false;
  }
}

try {
  record = mapper.toChangeStreamRecordJson(partition, row, metadata);
} catch (IllegalArgumentException e) {
  LOG.error("Malformed change-stream row: %s", e.getMessage());
  // skip, DLQ, or rethrow depending on pipeline policy
}

When it happens

Trigger: Calling toChangeStreamRecordJson (via toChangeStreamRecords) with a `row` string that is not valid JSON or does not conform to the expected proto-JSON structure, causing parser.merge(row, valueBuilder) to throw InvalidProtocolBufferException.

Common situations: Corrupted or manually edited rows in Spanner change stream metadata tables; reading rows written by an incompatible Spanner/connector version; unit-test fixtures with malformed JSON; character encoding issues corrupting the stored JSON.

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

• Failed to parse DirectedReadOptions from string: + directedR
• Failed to print type: {columnMetadataProto.getType().toStrin
• Failed to print type: {modStructValue}
• Failed to print type: {row}
• Failed to print mod: {row}

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