{"record":{"id":"dcdcc7950212937b","repo":"apache/pulsar","slug":"the-record-returned-by-the-source-cannot-be-null","errorCode":null,"errorMessage":"The record returned by the source cannot be null","messagePattern":"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":565,"sourceCode":"        Record<?> record;\n        if (componentType == FunctionDetails.ComponentType.SOURCE) {\n            Thread.currentThread().setContextClassLoader(componentClassLoader);\n        }\n        try {\n            record = this.source.read();\n        } catch (Exception e) {\n            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))","sourceCodeStart":547,"sourceCodeEnd":583,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java#L547-L583","documentation":"A Pulsar Function's Source implementation returned null from its read() method. JavaInstanceRunnable.readInput fetches the next Record from the configured Source and immediately validates it, because the function instance cannot process a null Record — it has no id, topic, or payload. The library throws IllegalArgumentException to fail fast at the input boundary rather than crashing later deep inside the function with no useful context.","triggerScenarios":"A custom Source class's read() returns null instead of blocking or throwing (e.g. developer treats 'no data available' as null). Also occurs when a Source wrapper/decorator (batching, filtering) returns null after consuming an element from an internal queue.","commonSituations":"Writing a custom Source that polls a database or external API and returns null when the query yields nothing; adapting a legacy collector that signals end-of-stream with null; misusing a bounded queue where poll() returns null on timeout.","solutions":["Fix the Source's read() implementation to block for data or return a valid Record instead of null; use a blocking queue take() or sleep-and-retry loop when no data is available","If the Source legitimately has no data, return a Record only when one exists — loop internally in read() until a Record can be constructed","Check any intermediate Source wrappers to ensure they never strip or drop records and return null","Inspect instance logs for the exact Source class name and review its read() code path for null-return branches"],"exampleFix":"// before\npublic Record<String> read() {\n    String msg = queue.poll();\n    return msg == null ? null : new StringRecord(msg);\n}\n// after\npublic Record<String> read() throws Exception {\n    String msg = queue.take(); // blocks until a record is available\n    return new StringRecord(msg);\n}","handlingStrategy":"validation","validationCode":"if (source instanceof MySource) {\n    Record<String> rec = ((MySource) source).read();\n    if (rec == null) throw new IllegalStateException(\"Source returned null record\");\n}","typeGuard":"boolean isValidRecord(Record<?> r) { return r != null; }","tryCatchPattern":null,"preventionTips":["Never return null from Source.read(); block or retry internally","Use blocking queue take() instead of poll() when waiting for data","Unit-test your Source with an empty-input scenario","Wrap third-party adapters so null markers are converted to exceptions"],"tags":["java","pulsar-functions","source","null-check"],"backgroundTag":"source-returned-null-record","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}