{"record":{"id":"faeefb5b4bdd3a3b","repo":"apache/beam","slug":"error-while-parsing-the-datachangerecord","errorCode":null,"errorMessage":"Error while parsing the DataChangeRecord","messagePattern":"Error while parsing the DataChangeRecord","errorType":"console","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java","lineNumber":291,"sourceCode":"                rowBuilder.withFieldValue(\n                    pkEntry.getKey().toLowerCase(),\n                    stringToParsedValue(\n                        internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),\n                        pkEntry.getValue()));\n          }\n          receiver\n              .get(OUTPUT_TAG)\n              .outputWithTimestamp(\n                  Row.withSchema(tableChangeRecordSchema)\n                      .addValue(record.getModType().toString())\n                      .addValue(record.getCommitTimestamp().toString())\n                      .addValue(Long.parseLong(record.getRecordSequence()))\n                      .addValue(rowBuilder.build())\n                      .build(),\n                  timestamp);\n        } catch (Exception e) {\n          errorsInBundle += 1;\n          LOG.warn(\"Error while parsing the DataChangeRecord\", e);\n          String recordString = \"Key:\" + mod.getKeysJson() + \" Value:\" + mod.getNewValuesJson();\n          receiver\n              .get(ERROR_TAG)\n              .output(Row.withSchema(ERROR_SCHEMA).addValues(e.toString(), recordString).build());\n        }\n      }\n    }\n\n    @FinishBundle\n    public void finish(FinishBundleContext c) {\n      errorCounter.inc(errorsInBundle);\n      errorsInBundle = 0L;\n    }\n  }\n\n  private static final HashMap<String, SpannerSchema> TABLE_SCHEMAS = new HashMap<>();\n\n  private static Schema getTableSchema(SpannerChangestreamsReadConfiguration config) {","sourceCodeStart":273,"sourceCodeEnd":309,"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/SpannerChangestreamsReadSchemaTransformProvider.java#L273-L309","documentation":"A DataChangeRecord from a Spanner change stream could not be parsed/converted into a Mutation row (e.g., Long.parseLong on the record sequence failed or the row builder rejected the values). The counter errorsInBundle is incremented, the exception is logged, and the record plus error text is emitted to the ERROR_TAG output instead of failing the pipeline.","triggerScenarios":"Any Exception while building the row from a DataChangeRecord in SpannerChangestreamsReadSchemaTransformProvider.process: unparseable record sequence, column type mismatch between the change stream payload and the target row schema, null values in non-nullable fields, malformed keys/new-values JSON.","commonSituations":"Spanner table schema changed after the read pipeline was configured (added/retyped columns), DataBoost/change stream records with unexpected types, decimal/numeric values overflowing Long fields.","solutions":["Consume the ERROR_TAG PCollection of the schema transform to retrieve e.toString() and the key/value JSON of failed records","Compare the Spanner table schema against the expected schema in the transform and reconfigure/update it after ALTER TABLE changes","Fix data-level issues (e.g., numeric values that don't fit Long) by widening the sink schema","Replay the failed records from Spanner once the schema is corrected"],"exampleFix":"// before\n.addValue(Long.parseLong(record.getRecordSequence()))\n// after\n.addValue(new BigDecimal(record.getRecordSequence())) // or validate digits before parse","handlingStrategy":"try-catch","validationCode":"// guard parse-ability before emitting the row\ntry { new BigDecimal(record.getRecordSequence()); } catch (NumberFormatException e) { sendToErrorTag(record, e); return; }\nvalidateRowAgainstSchema(rowBuilder.build(), ERROR_SCHEMA);","typeGuard":null,"tryCatchPattern":"try {\n  mutation = Mutation.newInsertBuilder(...)\n      .addValue(Long.parseLong(record.getRecordSequence()))\n      .addValue(rowBuilder.build()).build();\n} catch (Exception e) {\n  errorsInBundle += 1;\n  receiver.get(ERROR_TAG).output(Row.withSchema(ERROR_SCHEMA)\n      .addValues(e.toString(), \"Key:\" + mod.getKeysJson() + \" Value:\" + mod.getNewValuesJson()).build());\n}","preventionTips":["Always consume the ERROR_TAG output and persist failed records for replay","Re-run schema compatibility checks after any ALTER TABLE on the source Spanner table","Use column types wide enough (NUMERIC/BigDecimal) for change stream values","Monitor errorsInBundle; alert when it exceeds a small fraction of the bundle"],"tags":["gcp","spanner","change-stream","dead-letter","schema"],"backgroundTag":"schema-validation-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}