{"record":{"id":"7bd31bb981bf7e16","repo":"apache/beam","slug":"error-while-parsing-the-element-7bd31b","errorCode":null,"errorMessage":"Error while parsing the element","messagePattern":"Error while parsing the element","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaReadSchemaTransformProvider.java","lineNumber":367,"sourceCode":"        Schema errorSchema,\n        boolean handleErrors) {\n      this.errorCounter = Metrics.counter(KafkaReadSchemaTransformProvider.class, name);\n      this.valueMapper = valueMapper;\n      this.handleErrors = handleErrors;\n      this.errorSchema = errorSchema;\n    }\n\n    @ProcessElement\n    public void process(@DoFn.Element byte[] msg, MultiOutputReceiver receiver) {\n      Row mappedRow = null;\n      try {\n        mappedRow = valueMapper.apply(msg);\n      } catch (Exception e) {\n        if (!handleErrors) {\n          throw new RuntimeException(e);\n        }\n        errorsInBundle += 1;\n        LOG.warn(\"Error while parsing the element\", e);\n        receiver.get(ERROR_TAG).output(ErrorHandling.errorRecord(errorSchema, msg, e));\n      }\n      if (mappedRow != null) {\n        receiver.get(OUTPUT_TAG).output(mappedRow);\n      }\n    }\n\n    @FinishBundle\n    public void finish(FinishBundleContext c) {\n      errorCounter.inc(errorsInBundle);\n      errorsInBundle = 0L;\n    }\n  }\n\n  private static class ConsumerFactoryWithGcsTrustStores\n      implements SerializableFunction<Map<String, Object>, Consumer<byte[], byte[]>> {\n\n    @Override","sourceCodeStart":349,"sourceCodeEnd":385,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaReadSchemaTransformProvider.java#L349-L385","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Inspect the wrapped exception in the ERROR_TAG output's error message to identify the parse failure cause.","Update the value mapper / deserialization schema to match the actual topic payload format and version.","Handle null (tombstone) records explicitly in the mapper before parsing.","If the element should fail the pipeline instead, set handleErrors=false / remove error handling so the underlying RuntimeException propagates.","Verify you are reading the correct topic with the correct schema registry subject/version."],"exampleFix":"// before\nmappedRow = jsonMapper.apply(msg); // throws on null tombstone\n// after\nif (msg.value() == null) {\n  return; // skip tombstone\n}\nmappedRow = jsonMapper.apply(msg);","handlingStrategy":"try-catch","validationCode":"// Pre-validate the mapper against a sample record before building the pipeline\nsampleRecord = readSampleFromTopic();\nvalueMapper.apply(sampleRecord); // fails fast on schema mismatch","typeGuard":"if (msg.value() == null) return null; // tombstone record guard before mapping","tryCatchPattern":"try {\n  mappedRow = valueMapper.apply(msg);\n} catch (Exception e) {\n  log.error(\"Failed to map Kafka record on topic {} partition {} offset {}\", msg.topic(), msg.partition(), msg.offset(), e);\n  // route to error output or dead-letter\n}","preventionTips":["Handle null tombstone values in every mapper.","Version-check consumer schema (Avro/Proto) against the producer's schema registry subject.","Always configure the ERROR_TAG output and monitor its volume as a data-quality signal.","Test mappers with representative records from the live topic before deploying."],"tags":["kafka","deserialization","error-handling","beam-io"],"backgroundTag":"payload-parse-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"}