{"record":{"id":"bc7d88ad1eb83d9e","repo":"apache/pulsar","slug":"expected-messageidv5-got-messageid-getclass-bc7d88","errorCode":null,"errorMessage":"Expected MessageIdV5, got: ${messageId.getClass()}","messagePattern":"Expected MessageIdV5, got: (.+?)","errorType":"exception","errorClass":"java.lang.IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java","lineNumber":292,"sourceCode":"    @Override\n    public Message<T> receive() throws PulsarClientException {\n        return receiveQueue.take();\n    }\n\n    @Override\n    public Message<T> receive(Duration timeout) throws PulsarClientException {\n        return receiveQueue.poll(timeout);\n    }\n\n    @Override\n    public Messages<T> receiveMulti(int maxNumMessages, Duration timeout) throws PulsarClientException {\n        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    }","sourceCodeStart":274,"sourceCodeEnd":310,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java#L274-L310","documentation":"ScalableStreamConsumer.acknowledgeCumulative(MessageId) interprets a MessageIdV5's positionVector to ack every segment up to the recorded positions. Only MessageIdV5 carries this vector, so any other MessageId type throws IllegalArgumentException before any ack is sent. Also called internally by ackUpToVector.","triggerScenarios":"Calling acknowledgeCumulative() with a v4 MessageIdImpl/TopicMessageIdImpl or an id from another consumer/technology; cumulative acks in code shared between v4 and v5 consumers.","commonSituations":"Mixing clients during v4→v5 migration; storing ids in a database with the v4 serializer and re-acking later with the v5 consumer; generic message-processing pipelines typed on the base MessageId.","solutions":["Ack only with the MessageIdV5 instances this consumer produced","Convert/upgrade persisted ids to the MessageIdV5 format (with a valid positionVector)","Add an instanceof MessageIdV5 guard before cumulative acks","Do not cross-wire ids between ScalableStreamConsumer instances"],"exampleFix":"// before\nstreamConsumer.acknowledgeCumulative(legacyId);\n// after\nif (legacyId instanceof MessageIdV5 id) {\n    streamConsumer.acknowledgeCumulative(id);\n} else {\n    throw new IllegalArgumentException(\"cumulative ack requires MessageIdV5\");\n}","handlingStrategy":"type-guard","validationCode":"if (!(messageId instanceof MessageIdV5 id) || id.positionVector() == null) {\n    throw new IllegalArgumentException(\"cumulative ack requires MessageIdV5 with positionVector\");\n}","typeGuard":"static boolean isCumulativeCapable(MessageId id) {\n    return id instanceof MessageIdV5 v && v.positionVector() != null;\n}","tryCatchPattern":"try {\n    streamConsumer.acknowledgeCumulative(messageId);\n} catch (IllegalArgumentException e) {\n    log.error(\"cumulative ack requires MessageIdV5\", e);\n}","preventionTips":["Cumulative-ack only with ids from this stream consumer","Persist ids with the v5 serializer including positionVector","Guard shared ack utilities with instanceof MessageIdV5","Keep v4 and v5 ack paths separate during migration"],"tags":["pulsar","messageid","cumulative-ack","type-mismatch"],"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-14T00:17:10.932Z"}