{"record":{"id":"aeed9f83b02c1a61","repo":"apache/pulsar","slug":"expected-messageidv5-got-messageid-getclass-aeed9f","errorCode":null,"errorMessage":"Expected MessageIdV5, got: + messageId.getClass()","messagePattern":"Expected MessageIdV5, got: \\+ messageId\\.getClass\\(\\)","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java","lineNumber":304,"sourceCode":"        return new MessagesV5<>(receiveQueue.receiveMulti(maxNumMessages, timeout));\n    }\n\n    @Override\n    public void acknowledgeCumulative(MessageId messageId) {\n        if (!(messageId instanceof MessageIdV5 id)) {\n            throw new IllegalArgumentException(\"Expected MessageIdV5, got: \" + messageId.getClass());\n        }\n\n        // Ack each segment up to the position recorded in the vector\n        for (var entry : id.positionVector().entrySet()) {\n            ackSegmentUpTo(entry.getKey(), entry.getValue(), null);\n        }\n    }\n\n    @Override\n    public void acknowledgeCumulative(MessageId messageId, Transaction txn) {\n        if (!(messageId instanceof MessageIdV5 id)) {\n            throw new IllegalArgumentException(\"Expected MessageIdV5, got: \" + messageId.getClass());\n        }\n        var v4Txn = TransactionV5.unwrap(txn);\n        for (var entry : id.positionVector().entrySet()) {\n            ackSegmentUpTo(entry.getKey(), entry.getValue(), v4Txn);\n        }\n    }\n\n    /**\n     * Ack one segment up to the given position. Whole-segment (Exclusive) consumers ack\n     * cumulatively. PIP-486 bucket-shared segments are Key_Shared underneath, where cumulative acks\n     * are not permitted — the ack is translated into individually acking every delivered-but-unacked\n     * id up to the position (exactly the messages this consumer received: its buckets' share).\n     */\n    private void ackSegmentUpTo(long segmentId, org.apache.pulsar.client.api.MessageId position,\n                                org.apache.pulsar.client.api.transaction.Transaction v4Txn) {\n        // A draining consumer takes precedence: during a release the segment's slot in\n        // segmentConsumers is vacated (or already holds the re-subscribe chain), but the acks that\n        // complete the drain must still reach the consumer being drained.","sourceCodeStart":286,"sourceCodeEnd":322,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java#L286-L322","documentation":"Transactional cumulative ack on ScalableStreamConsumer requires MessageIdV5 because its positionVector drives per-segment acks inside the (unwrapped) v4 transaction. Other MessageId implementations have no vector, so IllegalArgumentException is thrown synchronously and the transaction is unaffected.","triggerScenarios":"acknowledgeCumulative(messageId, txn) called with a non-v5 id: an id from a v4 consumer, a deserialized v4 id, or an id produced by a different stream consumer, while a transaction is open.","commonSituations":"Transaction spanning messages from v4 and v5 consumers; replaying recorded ids from an old storage format inside a new transaction; framework code typed on the generic MessageId API.","solutions":["Use MessageIdV5 ids exclusively inside transactions on this consumer","Convert persisted ids to MessageIdV5 (reconstructing positionVector) before acking","Guard transactional ack code with instanceof MessageIdV5 checks","Keep v4 and v5 transactional consumers on separate code paths"],"exampleFix":"// before\nstreamConsumer.acknowledgeCumulative(v4Id, txn);\n// after\nif (v4Id instanceof MessageIdV5 id) {\n    streamConsumer.acknowledgeCumulative(id, txn);\n} else {\n    txn.abort();\n    throw new IllegalArgumentException(\"need MessageIdV5 with positionVector\");\n}","handlingStrategy":"type-guard","validationCode":"if (!(messageId instanceof MessageIdV5 id) || id.positionVector() == null) {\n    throw new IllegalArgumentException(\"transactional cumulative ack requires MessageIdV5\");\n}","typeGuard":"static boolean isCumulativeCapable(MessageId id) {\n    return id instanceof MessageIdV5 v && v.positionVector() != null;\n}","tryCatchPattern":"try {\n    streamConsumer.acknowledgeCumulative(messageId, txn);\n} catch (IllegalArgumentException e) {\n    txn.abort();\n    log.error(\"transactional cumulative ack type error\", e);\n}","preventionTips":["Use only v5 ids in transactions on this consumer","Abort transactions when ack validation fails","Rebuild positionVector when converting legacy ids","Isolate v4/v5 transactional code paths"],"tags":["pulsar","messageid","transaction","cumulative-ack"],"backgroundTag":"wrong-messageid-type","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"}