What causes "The record returned by the source cannot be null" in apache/pulsar?

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

apache/pulsar · error · IllegalArgumentException

The record returned by the source cannot be null

Error message

The record returned by the source cannot be null

What it means

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.

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

Example fix

// before
public Record<String> read() {
    String msg = queue.poll();
    return msg == null ? null : new StringRecord(msg);
}
// after
public Record<String> read() throws Exception {
    String msg = queue.take(); // blocks until a record is available
    return new StringRecord(msg);
}

Defensive patterns

Strategy: validation

Validate before calling

if (source instanceof MySource) {
    Record<String> rec = ((MySource) source).read();
    if (rec == null) throw new IllegalStateException("Source returned null record");
}

Type guard

boolean isValidRecord(Record<?> r) { return r != null; }

Prevention

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

When it happens

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

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

Related errors

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/dcdcc7950212937b. Report an issue: GitHub.

Appendix: source

Thrown at pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java:565

        Record<?> record;
        if (componentType == FunctionDetails.ComponentType.SOURCE) {
            Thread.currentThread().setContextClassLoader(componentClassLoader);
        }
        try {
            record = this.source.read();
        } catch (Exception e) {
            if (stats != null) {
                stats.incrSourceExceptions(e);
            }
            log.error().exception(e).log("Encountered exception in source read");
            throw e;
        } finally {
            Thread.currentThread().setContextClassLoader(instanceClassLoader);
        }

        // check record is valid
        if (record == null) {
            throw new IllegalArgumentException("The record returned by the source cannot be null");
        }
        // Eagerly access the value here so a malformed/poison message surfaces with enough
        // context (message id, topic, key, schema version) to be located and skipped, instead
        // of bubbling up as an opaque crash that names no message.
        try {
            if (record.getValue() == null) {
                throw new IllegalArgumentException("The value in the record returned by the source cannot be null");
            }
        } catch (Exception e) {
            logInputValueDecodeFailure(record, e);
            throw e;
        }
        return record;
    }

    private void logInputValueDecodeFailure(Record<?> record, Exception e) {
        log.warn()
                .attr("topic", record.getTopicName().orElse(null))

View on GitHub (pinned to 820761864e)