{"record":{"id":"3aaac93d8f846fc5","repo":"apache/pulsar","slug":"sourcerecord-class-type-must-be-pulsarrecord","errorCode":null,"errorMessage":"SourceRecord class type must be PulsarRecord","messagePattern":"SourceRecord class type must be PulsarRecord","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/AbstractSinkRecord.java","lineNumber":66,"sourceCode":"\n    @Override\n    public Optional<String> getTopicName() {\n        return sourceRecord.getTopicName();\n    }\n\n    @Override\n    public void ack() {\n        sourceRecord.ack();\n    }\n\n    /**\n     * Some sink sometimes wants to control the ack type.\n     */\n    public void cumulativeAck() {\n        if (sourceRecord instanceof PulsarRecord<?> pulsarRecord) {\n            pulsarRecord.cumulativeAck();\n        } else {\n            throw new RuntimeException(\"SourceRecord class type must be PulsarRecord\");\n        }\n    }\n\n    /**\n     * Some sink sometimes wants to control the ack type.\n     */\n    public void individualAck() {\n        if (sourceRecord instanceof PulsarRecord<?> pulsarRecord) {\n            pulsarRecord.individualAck();\n        } else {\n            throw new RuntimeException(\"SourceRecord class type must be PulsarRecord\");\n        }\n    }\n\n    @Override\n    public void fail() {\n        sourceRecord.fail();\n    }","sourceCodeStart":48,"sourceCodeEnd":84,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/AbstractSinkRecord.java#L48-L84","documentation":"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.","triggerScenarios":"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.","commonSituations":"Sinks shared across source types, IoT/Flink-derived records passed through functions, or generic sink code that assumes Pulsar-backed records.","solutions":["Only call cumulativeAck() when the record is Pulsar-backed; check `sourceRecord instanceof PulsarRecord` first (guard is already built in — use it before calling)","Use per-record ack() instead, which is supported for all record types","If cumulative semantics are required, ensure the pipeline sources data directly from Pulsar"],"exampleFix":"// before\nrecord.cumulativeAck(); // RuntimeException for non-Pulsar records\n// after\nif (record.getSourceRecord() instanceof PulsarRecord) {\n    record.cumulativeAck();\n} else {\n    record.ack();\n}","handlingStrategy":"type-guard","validationCode":"if (!(record.getSourceRecord() instanceof PulsarRecord)) {\n    // use plain ack() instead of cumulativeAck()\n}","typeGuard":"boolean canCumulativeAck(Record<?> r) { return r instanceof PulsarRecord; }","tryCatchPattern":"try {\n    record.cumulativeAck();\n} catch (RuntimeException e) {\n    record.ack(); // per-record fallback\n}","preventionTips":["Check sourceRecord type before requesting cumulative semantics","Use plain ack() for non-Pulsar-backed records","Keep sink code source-type agnostic; gate ack strategy on record type"],"tags":["java","pulsar-functions","sink","ack"],"backgroundTag":"unsupported-record-type","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"}