apache/rocketmq · warning · MQClientException

Fetch consume offset from broker exception

Error message

Fetch consume offset from broker exception

What it means

DefaultLitePullConsumerImpl.committed(MessageQueue) calls the offsetStore with MEMORY_FIRST_THEN_STORE; when the store lookup returns -2 (OFFSET_OVERFLOW / no readable offset — neither memory nor broker had a committed offset for the queue), it throws this MQClientException. It means the consumer group has never committed an offset for that queue anywhere.

Solutions

1. Poll at least once (letting auto-commit run) before querying committed()
2. Treat this error as 'no commit yet' and fall back to minOffset/maxOffset or consumer launch configuration (fromWhere)
3. Ensure autoCommit is enabled or call commitSync() before querying

Example fix

// before
long off = consumer.committed(queue);

// after
long off;
try {
    off = consumer.committed(queue);
} catch (MQClientException e) {
    off = consumer.minOffset(queue); // no commit yet: start from earliest
}

try {
    offset = consumer.committed(q);
} catch (MQClientException e) {
    offset = consumer.minOffset(q); // group never committed for this queue
}

When it happens

Trigger: Calling committed(q) right after start() before any message has been pulled and auto-committed; a brand-new consumer group (CONSUME_FROM_LAST_OFFSET never materialized) on a queue that was just assigned.

Common situations: New group on an existing topic; checking committed offset for a queue after seek() cleared committed state; offset store wiped (new client instance with remote offsets not yet persisted).

Related errors

• offset < 0
• QUERY_NOT_FOUND
• Seek offset illegal, seek offset =
• SYSTEM_ERROR
• allocateMessageQueueStrategy is null

AI-assisted analysis of apache/rocketmq@293f588571 (2026-08-14). Data as JSON: /api/errors/3079a953b43f7d14. Report an issue: GitHub.

Appendix: source

Thrown at client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java:831

    private void submitConsumeRequest(ConsumeRequest consumeRequest) {
        try {
            consumeRequestCache.put(consumeRequest);
        } catch (InterruptedException e) {
            log.error("Submit consumeRequest error", e);
        }
    }

    private long fetchConsumeOffset(MessageQueue messageQueue) throws MQClientException {
        checkServiceState();
        long offset = this.rebalanceImpl.computePullFromWhereWithException(messageQueue);
        return offset;
    }

    public long committed(MessageQueue messageQueue) throws MQClientException {
        checkServiceState();
        long offset = this.offsetStore.readOffset(messageQueue, ReadOffsetType.MEMORY_FIRST_THEN_STORE);
        if (offset == -2) {
            throw new MQClientException("Fetch consume offset from broker exception", null);
        }
        return offset;
    }

    private void clearMessageQueueInCache(MessageQueue messageQueue) {
        ProcessQueue processQueue = assignedMessageQueue.getProcessQueue(messageQueue);
        if (processQueue != null) {
            processQueue.clear();
        }
        Iterator<ConsumeRequest> iter = consumeRequestCache.iterator();
        while (iter.hasNext()) {
            if (iter.next().getMessageQueue().equals(messageQueue)) {
                iter.remove();
            }
        }
    }

    private long nextPullOffset(MessageQueue messageQueue) throws MQClientException {

View on GitHub (pinned to 293f588571)