{"record":{"id":"c7688538d9f9b9f0","repo":"apache/pulsar","slug":"the-value-in-the-record-returned-by-the-source-can","errorCode":null,"errorMessage":"The value in the record returned by the source cannot be null","messagePattern":"The value in the record returned by the source cannot be null","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java","lineNumber":572,"sourceCode":"            if (stats != null) {\n                stats.incrSourceExceptions(e);\n            }\n            log.error().exception(e).log(\"Encountered exception in source read\");\n            throw e;\n        } finally {\n            Thread.currentThread().setContextClassLoader(instanceClassLoader);\n        }\n\n        // check record is valid\n        if (record == null) {\n            throw new IllegalArgumentException(\"The record returned by the source cannot be null\");\n        }\n        // Eagerly access the value here so a malformed/poison message surfaces with enough\n        // context (message id, topic, key, schema version) to be located and skipped, instead\n        // of bubbling up as an opaque crash that names no message.\n        try {\n            if (record.getValue() == null) {\n                throw new IllegalArgumentException(\"The value in the record returned by the source cannot be null\");\n            }\n        } catch (Exception e) {\n            logInputValueDecodeFailure(record, e);\n            throw e;\n        }\n        return record;\n    }\n\n    private void logInputValueDecodeFailure(Record<?> record, Exception e) {\n        log.warn()\n                .attr(\"topic\", record.getTopicName().orElse(null))\n                .attr(\"messageId\", record.getMessage().map(m -> String.valueOf(m.getMessageId())).orElse(null))\n                .attr(\"partitionKey\", record.getKey().orElse(null))\n                .attr(\"schemaVersion\", record.getMessage()\n                        .map(Message::getSchemaVersion)\n                        .map(sv -> HexFormat.of().formatHex(sv))\n                        .orElse(null))\n                .exception(e)","sourceCodeStart":554,"sourceCodeEnd":590,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java#L554-L590","documentation":"The Source returned a Record whose getValue() is null. readInput eagerly resolves the value so a malformed or poison message fails with full context (message id, topic, key, schema version) via logInputValueDecodeFailure, then rethrows. The function runtime cannot route a null value through the function, so it throws IllegalArgumentException.","triggerScenarios":"Record.getValue() deserialization fails and the Record implementation swallows the exception returning null; a Source constructs a Record with a null payload; a schema decode of a malformed message yields null.","commonSituations":"Poison messages on the input topic (invalid protobuf/avro bytes); custom Record implementations that catch decode exceptions and return null; Sources built over untyped stores where the payload is genuinely missing.","solutions":["Fix the Source or Record implementation to throw a meaningful exception (or skip the record) instead of returning null from getValue()","Check the logInputValueDecodeFailure log output for topic/message-id/schema-version to locate and delete or repair the poison message","If using auto schema (AUTO_CONSUME), verify the topic's schema matches the declared function input schema; a mismatch can decode to null","Validate payloads at the producer side so null/empty values never reach the input topic"],"exampleFix":"// before\npublic String getValue() {\n    try { return decode(bytes); } catch (Exception e) { return null; }\n}\n// after\npublic String getValue() {\n    try { return decode(bytes); } catch (Exception e) {\n        throw new RuntimeException(\"Cannot decode message \" + getId().get(), e);\n    }\n}","handlingStrategy":"validation","validationCode":"Record<String> rec = source.read();\nif (rec == null || rec.getValue() == null) {\n    throw new IllegalStateException(\"null/null-value record from source\");\n}","typeGuard":"boolean hasValue(Record<?> r) { return r != null && r.getValue() != null; }","tryCatchPattern":"try { ctx.newOutputMessage(...).value(rec.getValue()).send(); }\ncatch (IllegalArgumentException e) { log.error(\"null record value\", e); /* skip */ }","preventionTips":["Let getValue() throw with message context instead of returning null","Match the function's declared schema to the topic's actual schema","Quarantine/delete poison messages identified via topic and message id in logs","Validate producer payloads upstream to prevent null/empty values"],"tags":["java","pulsar-functions","record","null-value","poison-message"],"backgroundTag":"null-record-value","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}