apache/beam · error

Error while parsing the DataChangeRecord

Error message

Error while parsing the DataChangeRecord

What it means

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.

Solutions

1. Consume the ERROR_TAG PCollection of the schema transform to retrieve e.toString() and the key/value JSON of failed records
2. Compare the Spanner table schema against the expected schema in the transform and reconfigure/update it after ALTER TABLE changes
3. Fix data-level issues (e.g., numeric values that don't fit Long) by widening the sink schema
4. Replay the failed records from Spanner once the schema is corrected

Example fix

// before
.addValue(Long.parseLong(record.getRecordSequence()))
// after
.addValue(new BigDecimal(record.getRecordSequence())) // or validate digits before parse

// guard parse-ability before emitting the row
try { new BigDecimal(record.getRecordSequence()); } catch (NumberFormatException e) { sendToErrorTag(record, e); return; }
validateRowAgainstSchema(rowBuilder.build(), ERROR_SCHEMA);

try {
  mutation = Mutation.newInsertBuilder(...)
      .addValue(Long.parseLong(record.getRecordSequence()))
      .addValue(rowBuilder.build()).build();
} catch (Exception e) {
  errorsInBundle += 1;
  receiver.get(ERROR_TAG).output(Row.withSchema(ERROR_SCHEMA)
      .addValues(e.toString(), "Key:" + mod.getKeysJson() + " Value:" + mod.getNewValuesJson()).build());
}

When it happens

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

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

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• Failed to fetch partition_mode for change stream
• Cannot find Spanner table.
• Duplicate column names
• Error processing struct to row
• Exception while trying to retrieve schema

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

Appendix: source

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

                rowBuilder.withFieldValue(
                    pkEntry.getKey().toLowerCase(),
                    stringToParsedValue(
                        internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),
                        pkEntry.getValue()));
          }
          receiver
              .get(OUTPUT_TAG)
              .outputWithTimestamp(
                  Row.withSchema(tableChangeRecordSchema)
                      .addValue(record.getModType().toString())
                      .addValue(record.getCommitTimestamp().toString())
                      .addValue(Long.parseLong(record.getRecordSequence()))
                      .addValue(rowBuilder.build())
                      .build(),
                  timestamp);
        } catch (Exception e) {
          errorsInBundle += 1;
          LOG.warn("Error while parsing the DataChangeRecord", e);
          String recordString = "Key:" + mod.getKeysJson() + " Value:" + mod.getNewValuesJson();
          receiver
              .get(ERROR_TAG)
              .output(Row.withSchema(ERROR_SCHEMA).addValues(e.toString(), recordString).build());
        }
      }
    }

    @FinishBundle
    public void finish(FinishBundleContext c) {
      errorCounter.inc(errorsInBundle);
      errorsInBundle = 0L;
    }
  }

  private static final HashMap<String, SpannerSchema> TABLE_SCHEMAS = new HashMap<>();

  private static Schema getTableSchema(SpannerChangestreamsReadConfiguration config) {

View on GitHub (pinned to 12126d8942)