apache/pulsar · error · java.lang.IllegalArgumentException

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

Error message

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

What it means

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.

Source

Thrown at pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java:292

    @Override
    public Message<T> receive() throws PulsarClientException {
        return receiveQueue.take();
    }

    @Override
    public Message<T> receive(Duration timeout) throws PulsarClientException {
        return receiveQueue.poll(timeout);
    }

    @Override
    public Messages<T> receiveMulti(int maxNumMessages, Duration timeout) throws PulsarClientException {
        return new MessagesV5<>(receiveQueue.receiveMulti(maxNumMessages, timeout));
    }

    @Override
    public void acknowledgeCumulative(MessageId messageId) {
        if (!(messageId instanceof MessageIdV5 id)) {
            throw new IllegalArgumentException("Expected MessageIdV5, got: " + messageId.getClass());
        }

        // Ack each segment up to the position recorded in the vector
        for (var entry : id.positionVector().entrySet()) {
            ackSegmentUpTo(entry.getKey(), entry.getValue(), null);
        }
    }

    @Override
    public void acknowledgeCumulative(MessageId messageId, Transaction txn) {
        if (!(messageId instanceof MessageIdV5 id)) {
            throw new IllegalArgumentException("Expected MessageIdV5, got: " + messageId.getClass());
        }
        var v4Txn = TransactionV5.unwrap(txn);
        for (var entry : id.positionVector().entrySet()) {
            ackSegmentUpTo(entry.getKey(), entry.getValue(), v4Txn);
        }
    }

View on GitHub (pinned to 820761864e)

Solutions

1. Ack only with the MessageIdV5 instances this consumer produced
2. Convert/upgrade persisted ids to the MessageIdV5 format (with a valid positionVector)
3. Add an instanceof MessageIdV5 guard before cumulative acks
4. Do not cross-wire ids between ScalableStreamConsumer instances

Example fix

// before
streamConsumer.acknowledgeCumulative(legacyId);
// after
if (legacyId instanceof MessageIdV5 id) {
    streamConsumer.acknowledgeCumulative(id);
} else {
    throw new IllegalArgumentException("cumulative ack requires MessageIdV5");
}

if (!(messageId instanceof MessageIdV5 id) || id.positionVector() == null) {
    throw new IllegalArgumentException("cumulative ack requires MessageIdV5 with positionVector");
}

static boolean isCumulativeCapable(MessageId id) {
    return id instanceof MessageIdV5 v && v.positionVector() != null;
}

try {
    streamConsumer.acknowledgeCumulative(messageId);
} catch (IllegalArgumentException e) {
    log.error("cumulative ack requires MessageIdV5", e);
}

When it happens

Trigger: Calling acknowledgeCumulative() with a v4 MessageIdImpl/TopicMessageIdImpl or an id from another consumer/technology; cumulative acks in code shared between v4 and v5 consumers.

Common situations: 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.

Related errors

• Expected MessageIdV5, got: ${messageId.getClass()}
• Expected MessageIdV5, got: ${messageId.getClass()}
• Expected MessageIdV5, got: + messageId.getClass()
• Class ${entryFilterClass} does not implement entry filter in
• Class ${className} does not implement additional servlet int

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