{"record":{"id":"a4183c74c4306aab","repo":"zhisheng17/flink-learning","slug":"unknown-value-for-consumer-offset-reset-to","errorCode":null,"errorMessage":"Unknown value for CONSUMER_OFFSET_RESET_TO.","messagePattern":"Unknown value for CONSUMER_OFFSET_RESET_TO\\.","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink-learning-connectors/flink-learning-connectors-rocketmq/src/main/java/com/zhisheng/connectors/rocketmq/RocketMQSource.java","lineNumber":213,"sourceCode":"            offset = restoredOffsets.get(mq);\n        }\n        if (offset == null) {\n            offset = consumer.fetchConsumeOffset(mq, false);\n            if (offset < 0) {\n                String initialOffset = props.getProperty(RocketMQConfig.CONSUMER_OFFSET_RESET_TO, CONSUMER_OFFSET_LATEST);\n                switch (initialOffset) {\n                    case CONSUMER_OFFSET_EARLIEST:\n                        offset = consumer.minOffset(mq);\n                        break;\n                    case CONSUMER_OFFSET_LATEST:\n                        offset = consumer.maxOffset(mq);\n                        break;\n                    case CONSUMER_OFFSET_TIMESTAMP:\n                        offset = consumer.searchOffset(mq, getLong(props,\n                                RocketMQConfig.CONSUMER_OFFSET_FROM_TIMESTAMP, System.currentTimeMillis()));\n                        break;\n                    default:\n                        throw new IllegalArgumentException(\"Unknown value for CONSUMER_OFFSET_RESET_TO.\");\n                }\n            }\n        }\n        offsetTable.put(mq, offset);\n        return offsetTable.get(mq);\n    }\n\n    private void putMessageQueueOffset(MessageQueue mq, long offset) throws MQClientException {\n        offsetTable.put(mq, offset);\n        consumer.updateConsumeOffset(mq, offset);\n    }\n\n    @Override\n    public void cancel() {\n        LOG.debug(\"cancel ...\");\n        runningChecker.setRunning(false);\n\n        if (pullConsumerScheduleService != null) {","sourceCodeStart":195,"sourceCodeEnd":231,"githubUrl":"https://github.com/zhisheng17/flink-learning/blob/d731cee7618021be56d132cc925102ffff8d75e6/flink-learning-connectors/flink-learning-connectors-rocketmq/src/main/java/com/zhisheng/connectors/rocketmq/RocketMQSource.java#L195-L231","documentation":"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.').","triggerScenarios":"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'.","commonSituations":"Config typos or case mismatches when copying offset-reset settings from Kafka consumer configs (auto.offset.reset=earliest) into RocketMQ connector properties.","solutions":["Set CONSUMER_OFFSET_RESET_TO to exactly one of the supported constants: CONSUMER_OFFSET_LATEST, CONSUMER_OFFSET_EARLIEST, or CONSUMER_OFFSET_TIMESTAMP","Remove the property entirely to use the default behavior","Compare your config value against the RocketMQConfig constant definitions"],"exampleFix":"// before\nprops.put(RocketMQConfig.CONSUMER_OFFSET_RESET_TO, \"earliest\");\n// after\nprops.put(RocketMQConfig.CONSUMER_OFFSET_RESET_TO, RocketMQConfig.CONSUMER_OFFSET_EARLIEST);","handlingStrategy":"validation","validationCode":"String v = props.getProperty(RocketMQConfig.CONSUMER_OFFSET_RESET_TO);\nSet<String> allowed = Set.of(\n    RocketMQConfig.CONSUMER_OFFSET_LATEST,\n    RocketMQConfig.CONSUMER_OFFSET_EARLIEST,\n    RocketMQConfig.CONSUMER_OFFSET_TIMESTAMP);\nif (v != null && !allowed.contains(v)) {\n    throw new IllegalArgumentException(\"CONSUMER_OFFSET_RESET_TO must be one of \" + allowed + \", got: \" + v);\n}","typeGuard":null,"tryCatchPattern":"try {\n    source.run(ctx);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage() != null && e.getMessage().contains(\"CONSUMER_OFFSET_RESET_TO\")) {\n        props.remove(RocketMQConfig.CONSUMER_OFFSET_RESET_TO); // fall back to default\n    } else throw e;\n}","preventionTips":["Use the RocketMQConfig constants, never raw strings, when setting the property","Remember this is not Kafka's auto.offset.reset syntax — do not copy 'earliest'/'latest' lowercase values","Validate consumer properties at job-submission time"],"tags":["rocketmq","config","enum","flink","kafka-migration"],"backgroundTag":"invalid-enum-value","analyzedSha":"d731cee7618021be56d132cc925102ffff8d75e6","analyzedAt":"2026-09-06T05:35:08.496Z","contentChangedAt":"2026-09-06T05:35:08.496Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}