{"record":{"id":"301ab52d3ff1512d","repo":"apache/beam","slug":"failed-to-parse-record-into-proto-row","errorCode":null,"errorMessage":"Failed to parse record into proto: {row}","messagePattern":"Failed to parse record into proto: (.+?)","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/mapper/ChangeStreamRecordMapper.java","lineNumber":479,"sourceCode":"            .filter(this::isNonNullHeartbeatRecord)\n            .map(struct -> toHeartbeatRecord(partition, struct, resultSetMetadata));\n\n    final Stream<ChildPartitionsRecord> childPartitionsRecords =\n        row.getStructList(CHILD_PARTITIONS_RECORD_COLUMN).stream()\n            .filter(this::isNonNullChildPartitionsRecord)\n            .map(struct -> toChildPartitionsRecord(partition, struct, resultSetMetadata));\n\n    return Stream.concat(\n        Stream.concat(dataChangeRecords, heartbeatRecords), childPartitionsRecords);\n  }\n\n  ChangeStreamRecord toChangeStreamRecordJson(\n      PartitionMetadata partition, String row, ChangeStreamResultSetMetadata resultSetMetadata) {\n    Value.Builder valueBuilder = Value.newBuilder();\n    try {\n      this.parser.merge(row, valueBuilder);\n    } catch (InvalidProtocolBufferException exc) {\n      throw new IllegalArgumentException(\"Failed to parse record into proto: \" + row);\n    }\n    Value value = valueBuilder.build();\n    if (isNonNullDataChangeRecordJson(value)) {\n      return toDataChangeRecordJson(partition, value, resultSetMetadata);\n    } else if (isNonNullHeartbeatRecordJson(value)) {\n      return toHeartbeatRecordJson(partition, value, resultSetMetadata);\n    } else if (isNonNullChildPartitionsRecordJson(value)) {\n      return toChildPartitionsRecordJson(partition, value, resultSetMetadata);\n    } else {\n      throw new IllegalArgumentException(\"Unknown change stream record type \" + row);\n    }\n  }\n\n  private boolean isNonNullDataChangeRecord(Struct row) {\n    return !row.isNull(COMMIT_TIMESTAMP_COLUMN);\n  }\n\n  private boolean isNonNullDataChangeRecordJson(Value row) {","sourceCodeStart":461,"sourceCodeEnd":497,"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/mapper/ChangeStreamRecordMapper.java#L461-L497","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Validate the row is well-formed JSON (e.g. with a JSON parser) before feeding it to the mapper.","Check the source of the row string: confirm it was produced by the same Spanner change stream format the mapper expects.","Verify Beam google-cloud-platform SDK version matches the data written by your pipeline version.","Wrap toChangeStreamRecords in try-catch to log and skip/transform the malformed row instead of failing the pipeline.","Inspect the offending row printed in the message for truncation, encoding, or manual modification."],"exampleFix":"// before\nChangeStreamRecord record = mapper.toChangeStreamRecordJson(partition, rawRow, metadata);\n\n// after\nnew ObjectMapper().readTree(rawRow); // throws early with a clear parse error if invalid\nChangeStreamRecord record = mapper.toChangeStreamRecordJson(partition, rawRow, metadata);","handlingStrategy":"validation","validationCode":"boolean isProtoJsonRow(String row) {\n  try {\n    com.google.protobuf.util.JsonFormat.parser().ignoringUnknownFields().merge(row,\n        com.google.protobuf.Value.newBuilder());\n    return true;\n  } catch (Exception e) {\n    return false;\n  }\n}","typeGuard":null,"tryCatchPattern":"try {\n  record = mapper.toChangeStreamRecordJson(partition, row, metadata);\n} catch (IllegalArgumentException e) {\n  LOG.error(\"Malformed change-stream row: %s\", e.getMessage());\n  // skip, DLQ, or rethrow depending on pipeline policy\n}","preventionTips":["Only feed mapper rows read directly from Spanner change streams — never hand-edited strings.","Validate rows as JSON in tests before running mappers on fixtures.","Keep the Beam google-cloud-platform SDK version in sync across read/write paths.","Alert on this exception to catch data-format drift early."],"tags":["java","spanner","json","protobuf","data-mapping"],"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"}