apache/beam · warning

Error while parsing the element

Error message

Error while parsing the element

What it means

Warning logged per record when applying the configured valueMapper to a consumed Kafka message throws and error handling is enabled. Instead of failing the DoFn, the record is routed to the ERROR_TAG output with an error record containing the error schema; if handleErrors is false, a RuntimeException is thrown instead.

Solutions

1. Inspect the wrapped exception in the ERROR_TAG output's error message to identify the parse failure cause.
2. Update the value mapper / deserialization schema to match the actual topic payload format and version.
3. Handle null (tombstone) records explicitly in the mapper before parsing.
4. If the element should fail the pipeline instead, set handleErrors=false / remove error handling so the underlying RuntimeException propagates.
5. Verify you are reading the correct topic with the correct schema registry subject/version.

Example fix

// before
mappedRow = jsonMapper.apply(msg); // throws on null tombstone
// after
if (msg.value() == null) {
  return; // skip tombstone
}
mappedRow = jsonMapper.apply(msg);

// Pre-validate the mapper against a sample record before building the pipeline
sampleRecord = readSampleFromTopic();
valueMapper.apply(sampleRecord); // fails fast on schema mismatch

if (msg.value() == null) return null; // tombstone record guard before mapping

try {
  mappedRow = valueMapper.apply(msg);
} catch (Exception e) {
  log.error("Failed to map Kafka record on topic {} partition {} offset {}", msg.topic(), msg.partition(), msg.offset(), e);
  // route to error output or dead-letter
}

When it happens

Trigger: KafkaReadSchemaTransformProvider's process() receives a ConsumerRecord whose payload fails valueMapper.apply(msg) — e.g. Avro/Proto/JSON deserialization mismatch, wrong schema, corrupt data, or a null value from a tombstone record — while errorHandling is configured.

Common situations: Producer upgraded to a new schema version not matching the consumer's mapper; malformed JSON on the topic; tombstone records (null values) hitting a mapper that does not handle nulls; wrong topic consumed with different data format.

Related errors

• : closing producer after unrecoverable error. The work…
• : consumer thread is interrupted
• consumerPollingTimeout should be > 0.
• Couldn't infer Coder from
• Encoded value for the offset consumer config key

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

Appendix: source

Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaReadSchemaTransformProvider.java:367

        Schema errorSchema,
        boolean handleErrors) {
      this.errorCounter = Metrics.counter(KafkaReadSchemaTransformProvider.class, name);
      this.valueMapper = valueMapper;
      this.handleErrors = handleErrors;
      this.errorSchema = errorSchema;
    }

    @ProcessElement
    public void process(@DoFn.Element byte[] msg, MultiOutputReceiver receiver) {
      Row mappedRow = null;
      try {
        mappedRow = valueMapper.apply(msg);
      } catch (Exception e) {
        if (!handleErrors) {
          throw new RuntimeException(e);
        }
        errorsInBundle += 1;
        LOG.warn("Error while parsing the element", e);
        receiver.get(ERROR_TAG).output(ErrorHandling.errorRecord(errorSchema, msg, e));
      }
      if (mappedRow != null) {
        receiver.get(OUTPUT_TAG).output(mappedRow);
      }
    }

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

  private static class ConsumerFactoryWithGcsTrustStores
      implements SerializableFunction<Map<String, Object>, Consumer<byte[], byte[]>> {

    @Override

View on GitHub (pinned to 12126d8942)