apache/pulsar · error · RuntimeException

SourceRecord class type must be PulsarRecord

Error message

SourceRecord class type must be PulsarRecord

What it means

AbstractSinkRecord.cumulativeAck() delegates acknowledgment to the wrapped sourceRecord only when it is an instance of PulsarRecord (a record backed by a Pulsar reader/consumer that supports cumulative acks). For other Record implementations (e.g. records from other sources) cumulative acknowledgment is impossible, so it throws RuntimeException.

Source

Thrown at pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/AbstractSinkRecord.java:66

    @Override
    public Optional<String> getTopicName() {
        return sourceRecord.getTopicName();
    }

    @Override
    public void ack() {
        sourceRecord.ack();
    }

    /**
     * Some sink sometimes wants to control the ack type.
     */
    public void cumulativeAck() {
        if (sourceRecord instanceof PulsarRecord<?> pulsarRecord) {
            pulsarRecord.cumulativeAck();
        } else {
            throw new RuntimeException("SourceRecord class type must be PulsarRecord");
        }
    }

    /**
     * Some sink sometimes wants to control the ack type.
     */
    public void individualAck() {
        if (sourceRecord instanceof PulsarRecord<?> pulsarRecord) {
            pulsarRecord.individualAck();
        } else {
            throw new RuntimeException("SourceRecord class type must be PulsarRecord");
        }
    }

    @Override
    public void fail() {
        sourceRecord.fail();
    }

View on GitHub (pinned to 820761864e)

Solutions

1. Only call cumulativeAck() when the record is Pulsar-backed; check `sourceRecord instanceof PulsarRecord` first (guard is already built in — use it before calling)
2. Use per-record ack() instead, which is supported for all record types
3. If cumulative semantics are required, ensure the pipeline sources data directly from Pulsar

Example fix

// before
record.cumulativeAck(); // RuntimeException for non-Pulsar records
// after
if (record.getSourceRecord() instanceof PulsarRecord) {
    record.cumulativeAck();
} else {
    record.ack();
}

if (!(record.getSourceRecord() instanceof PulsarRecord)) {
    // use plain ack() instead of cumulativeAck()
}

boolean canCumulativeAck(Record<?> r) { return r instanceof PulsarRecord; }

try {
    record.cumulativeAck();
} catch (RuntimeException e) {
    record.ack(); // per-record fallback
}

When it happens

Trigger: Calling cumulativeAck() on a sink record whose sourceRecord came from a non-Pulsar source (e.g. Kafka-source records or synthetic records produced by a function) instead of PulsarRecord.

Common situations: Sinks shared across source types, IoT/Flink-derived records passed through functions, or generic sink code that assumes Pulsar-backed records.

Related errors

• Sink does not implement correct interface
• Component cannot get state store
• not implemented
• Schema should not be null.
• The record returned by the source cannot be null

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