zhisheng17/flink-learning · error · IllegalArgumentException

Unknown value for CONSUMER_OFFSET_RESET_TO.

Error message

Unknown value for CONSUMER_OFFSET_RESET_TO.

What it means

RocketMQSource.getMessageQueueOffset switches on the CONSUMER_OFFSET_RESET_TO property to decide where to start consuming (latest, earliest, or a timestamp). An unrecognized value falls into the default branch and throws IllegalArgumentException('Unknown value for CONSUMER_OFFSET_RESET_TO.').

Source

Thrown at flink-learning-connectors/flink-learning-connectors-rocketmq/src/main/java/com/zhisheng/connectors/rocketmq/RocketMQSource.java:213

            offset = restoredOffsets.get(mq);
        }
        if (offset == null) {
            offset = consumer.fetchConsumeOffset(mq, false);
            if (offset < 0) {
                String initialOffset = props.getProperty(RocketMQConfig.CONSUMER_OFFSET_RESET_TO, CONSUMER_OFFSET_LATEST);
                switch (initialOffset) {
                    case CONSUMER_OFFSET_EARLIEST:
                        offset = consumer.minOffset(mq);
                        break;
                    case CONSUMER_OFFSET_LATEST:
                        offset = consumer.maxOffset(mq);
                        break;
                    case CONSUMER_OFFSET_TIMESTAMP:
                        offset = consumer.searchOffset(mq, getLong(props,
                                RocketMQConfig.CONSUMER_OFFSET_FROM_TIMESTAMP, System.currentTimeMillis()));
                        break;
                    default:
                        throw new IllegalArgumentException("Unknown value for CONSUMER_OFFSET_RESET_TO.");
                }
            }
        }
        offsetTable.put(mq, offset);
        return offsetTable.get(mq);
    }

    private void putMessageQueueOffset(MessageQueue mq, long offset) throws MQClientException {
        offsetTable.put(mq, offset);
        consumer.updateConsumeOffset(mq, offset);
    }

    @Override
    public void cancel() {
        LOG.debug("cancel ...");
        runningChecker.setRunning(false);

        if (pullConsumerScheduleService != null) {

View on GitHub (pinned to d731cee761)

Solutions

1. Set CONSUMER_OFFSET_RESET_TO to exactly one of the supported constants: CONSUMER_OFFSET_LATEST, CONSUMER_OFFSET_EARLIEST, or CONSUMER_OFFSET_TIMESTAMP
2. Remove the property entirely to use the default behavior
3. Compare your config value against the RocketMQConfig constant definitions

Example fix

// before
props.put(RocketMQConfig.CONSUMER_OFFSET_RESET_TO, "earliest");
// after
props.put(RocketMQConfig.CONSUMER_OFFSET_RESET_TO, RocketMQConfig.CONSUMER_OFFSET_EARLIEST);

String v = props.getProperty(RocketMQConfig.CONSUMER_OFFSET_RESET_TO);
Set<String> allowed = Set.of(
    RocketMQConfig.CONSUMER_OFFSET_LATEST,
    RocketMQConfig.CONSUMER_OFFSET_EARLIEST,
    RocketMQConfig.CONSUMER_OFFSET_TIMESTAMP);
if (v != null && !allowed.contains(v)) {
    throw new IllegalArgumentException("CONSUMER_OFFSET_RESET_TO must be one of " + allowed + ", got: " + v);
}

try {
    source.run(ctx);
} catch (IllegalArgumentException e) {
    if (e.getMessage() != null && e.getMessage().contains("CONSUMER_OFFSET_RESET_TO")) {
        props.remove(RocketMQConfig.CONSUMER_OFFSET_RESET_TO); // fall back to default
    } else throw e;
}

When it happens

Trigger: Setting RocketMQConfig.CONSUMER_OFFSET_RESET_TO in the consumer properties to any string other than the supported constants (CONSUMER_OFFSET_LATEST / EARLIEST / TIMESTAMP), e.g. a typo like 'lastest' or 'LATEST'.

Common situations: Config typos or case mismatches when copying offset-reset settings from Kafka consumer configs (auto.offset.reset=earliest) into RocketMQ connector properties.

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

• invalid elasticsearch hosts format
• invalid elasticsearch hosts format
• This ${database} database does not exist!
• Invalid host/port configuration. Host: ${host} Port: ${port}
• Unsupported client type - cannot happen

AI-assisted analysis of zhisheng17/flink-learning@d731cee761 (2026-09-06). Data as JSON: /api/errors/a4183c74c4306aab. Report an issue: GitHub.