{"record":{"id":"c394165ba066278c","repo":"apache/pulsar","slug":"expected-messageidv5-got-messageid-getclass-c39416","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/MultiTopicStreamConsumer.java","lineNumber":330,"sourceCode":"        fanOutCumulativeAck(messageId, (sc, vector) -> sc.ackUpToVector(vector));\n    }\n\n    @Override\n    public void acknowledgeCumulative(MessageId messageId, Transaction txn) {\n        // Transactions on multi-topic are best-effort across per-topic consumers — each\n        // per-topic ack is independently transactional. See note in the design doc.\n        fanOutCumulativeAck(messageId, (sc, vector) -> sc.ackUpToVector(vector));\n    }\n\n    /**\n     * For a cumulative ack on a multi-topic message, look up its multi-topic vector\n     * and invoke the per-topic ack on every parent topic.\n     */\n    private void fanOutCumulativeAck(MessageId messageId,\n                                     java.util.function.BiConsumer<ScalableStreamConsumer<T>,\n                                             Map<Long, org.apache.pulsar.client.api.MessageId>> action) {\n        if (!(messageId instanceof MessageIdV5 id)) {\n            throw new IllegalArgumentException(\"Expected MessageIdV5, got: \" + messageId.getClass());\n        }\n        Map<String, Map<Long, org.apache.pulsar.client.api.MessageId>> vector = id.multiTopicVector();\n        if (vector == null) {\n            throw new IllegalStateException(\"MessageIdV5 missing multi-topic vector — was the\"\n                    + \" message delivered through a multi-topic stream consumer?\");\n        }\n        for (var entry : vector.entrySet()) {\n            PerTopic<T> state = perTopic.get(entry.getKey());\n            if (state == null) {\n                // Topic left the matching set since this message was enqueued: we've\n                // detached it and no longer ack removed topics, so skip its slice.\n                continue;\n            }\n            action.accept(state.consumer, entry.getValue());\n        }\n    }\n\n    @Override","sourceCodeStart":312,"sourceCodeEnd":348,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java#L312-L348","documentation":"Thrown by MultiTopicStreamConsumer.fanOutCumulativeAck when the given MessageId is not a MessageIdV5. Cumulative acks on a multi-topic stream consumer need the per-topic id vector carried only by MessageIdV5; foreign id types cannot be fanned out to per-topic cumulative acks.","triggerScenarios":"Calling acknowledgeCumulative(messageId) with an id from a single-topic consumer, a MessageIdV3, or a manually built id.","commonSituations":"Mixing ids between consumer types; using ids loaded from external storage produced by another client version.","solutions":["Pass only ids returned by this multi-topic stream consumer","Check instanceof MessageIdV5 before acknowledgeCumulative","Re-receive messages instead of reusing ids from other consumers"],"exampleFix":"// before\nstreamConsumer.acknowledgeCumulative(foreignId);\n// after\nif (foreignId instanceof MessageIdV5 id) {\n    streamConsumer.acknowledgeCumulative(id);\n}","handlingStrategy":"type-guard","validationCode":null,"typeGuard":"static boolean isV5Id(MessageId id) { return id instanceof MessageIdV5; }","tryCatchPattern":"try { consumer.acknowledgeCumulative(id); } catch (IllegalArgumentException e) { log.error(\"cumulative ack needs MessageIdV5\", e); }","preventionTips":["Use only ids from the same stream consumer","Never reuse ids across consumer types"],"tags":["pulsar","client","ack","type-mismatch","cumulative-ack"],"backgroundTag":"message-id-type-mismatch","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"}