apache/pulsar · error · java.lang.IllegalArgumentException

Expected MessageIdV5, got: ${messageId.getClass()}

Error message

Expected MessageIdV5, got: ${messageId.getClass()}

What it means

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.

Source

Thrown at pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java:330

        fanOutCumulativeAck(messageId, (sc, vector) -> sc.ackUpToVector(vector));
    }

    @Override
    public void acknowledgeCumulative(MessageId messageId, Transaction txn) {
        // Transactions on multi-topic are best-effort across per-topic consumers — each
        // per-topic ack is independently transactional. See note in the design doc.
        fanOutCumulativeAck(messageId, (sc, vector) -> sc.ackUpToVector(vector));
    }

    /**
     * For a cumulative ack on a multi-topic message, look up its multi-topic vector
     * and invoke the per-topic ack on every parent topic.
     */
    private void fanOutCumulativeAck(MessageId messageId,
                                     java.util.function.BiConsumer<ScalableStreamConsumer<T>,
                                             Map<Long, org.apache.pulsar.client.api.MessageId>> action) {
        if (!(messageId instanceof MessageIdV5 id)) {
            throw new IllegalArgumentException("Expected MessageIdV5, got: " + messageId.getClass());
        }
        Map<String, Map<Long, org.apache.pulsar.client.api.MessageId>> vector = id.multiTopicVector();
        if (vector == null) {
            throw new IllegalStateException("MessageIdV5 missing multi-topic vector — was the"
                    + " message delivered through a multi-topic stream consumer?");
        }
        for (var entry : vector.entrySet()) {
            PerTopic<T> state = perTopic.get(entry.getKey());
            if (state == null) {
                // Topic left the matching set since this message was enqueued: we've
                // detached it and no longer ack removed topics, so skip its slice.
                continue;
            }
            action.accept(state.consumer, entry.getValue());
        }
    }

    @Override

View on GitHub (pinned to 820761864e)

Solutions

1. Pass only ids returned by this multi-topic stream consumer
2. Check instanceof MessageIdV5 before acknowledgeCumulative
3. Re-receive messages instead of reusing ids from other consumers

Example fix

// before
streamConsumer.acknowledgeCumulative(foreignId);
// after
if (foreignId instanceof MessageIdV5 id) {
    streamConsumer.acknowledgeCumulative(id);
}

static boolean isV5Id(MessageId id) { return id instanceof MessageIdV5; }

try { consumer.acknowledgeCumulative(id); } catch (IllegalArgumentException e) { log.error("cumulative ack needs MessageIdV5", e); }

When it happens

Trigger: Calling acknowledgeCumulative(messageId) with an id from a single-topic consumer, a MessageIdV3, or a manually built id.

Common situations: Mixing ids between consumer types; using ids loaded from external storage produced by another client version.

Related errors

• Expected MessageIdV5, got: ${messageId.getClass()}
• MessageIdV5 missing parent topic — was the message delivered
• MessageIdV5 missing multi-topic vector — was the message del
• Expected MessageIdV5, got: ${messageId.getClass()}
• Class ${entryFilterClass} does not implement entry filter in

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