apache/seatunnel · error · IllegalArgumentException

start.mode.timestamp must not be greater than the current…

Error message

start.mode.timestamp must not be greater than the current time

What it means

RocketMqSourceConfig.parseTableConfig validates start.mode.timestamp during table config parsing. When start.mode = timestamp and the configured timestamp is in the future (greater than System.currentTimeMillis()), it throws IllegalArgumentException, since consuming from a future timestamp is not meaningful and the value can't be resolved declaratively.

Solutions

Set start.mode.timestamp to a past or current epoch-millisecond value
Confirm the value is in milliseconds, not seconds or a date string
Clamp/guard the value at generation time: Math.min(ts, System.currentTimeMillis())
Use start.mode = earliest/latest if a fixed past timestamp isn't required

Example fix

// before
start_mode = "timestamp"
start_mode_timestamp = 9999999999999
// after
start_mode = "timestamp"
start_mode_timestamp = 1757472000000

Defensive patterns

Strategy: validation

Validate before calling

long ts = Long.parseLong(cfg.get("start_mode_timestamp"));
if (ts > System.currentTimeMillis()) throw new IllegalArgumentException("start.mode.timestamp is in the future");

Try / catch

try { new RocketMqSourceConfig(tableConfig); } catch (IllegalArgumentException e) { if (e.getMessage().contains("start.mode.timestamp")) { /* use past timestamp or start.mode=latest */ } throw e; }

Prevention

Always express timestamps in epoch milliseconds
Clamp generated timestamps to the current clock at config write time
Avoid hardcoding 'now' timestamps; prefer start.mode = latest

When it happens

Trigger: start.mode = CONSUME_FROM_TIMESTAMP with start.mode.timestamp > current wall-clock time; parseTableConfig runs at source config construction, failing the job at submission/planning time.

Common situations: Timestamp in milliseconds supplied as seconds (or vice versa) producing a huge value; timestamp computed from a future event or misconfigured template; clock skew between config generator and runtime.

Understand the failure class

Background: "Invalid value" and "allowed values are" config errors: what your library rejected and how to fix it — this error's family across 41 libraries.

Related errors

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/9ae61e6f6fc06b94. Report an issue: GitHub.

Appendix: source

Thrown at seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/source/RocketMqSourceConfig.java:136

                            catalogTable.getOptions(),
                            catalogTable.getPartitionKeys(),
                            catalogTable.getComment());
        }
        DeserializationSchema<SeaTunnelRow> deserializationSchema =
                buildDeserialization(tableConfig, catalogTable);
        List<String> tags = parseTags(tableConfig);
        StartMode startMode =
                tableConfig.getOptional(RocketMqSourceOptions.START_MODE).orElse(null);
        Long startTimestamp = null;

        if (startMode != null) {
            switch (startMode) {
                case CONSUME_FROM_TIMESTAMP:
                    startTimestamp = tableConfig.get(RocketMqSourceOptions.START_MODE_TIMESTAMP);
                    // Runtime check: cannot be declarative (depends on current time)
                    long currentTimestamp = System.currentTimeMillis();
                    if (startTimestamp > currentTimestamp) {
                        throw new IllegalArgumentException(
                                "start.mode.timestamp must not be greater than the current time");
                    }
                    break;
                case CONSUME_FROM_SPECIFIC_OFFSETS:
                    Map<String, Long> offsetConfigMap =
                            tableConfig.get(RocketMqSourceOptions.START_MODE_OFFSETS);
                    Map<MessageQueue, Long> specificOffsets = metadata.getSpecificStartOffsets();
                    if (specificOffsets == null) {
                        specificOffsets = new HashMap<>();
                        metadata.setSpecificStartOffsets(specificOffsets);
                    }
                    for (Map.Entry<String, Long> entry : offsetConfigMap.entrySet()) {
                        int splitIndex = entry.getKey().lastIndexOf("-");
                        String topicName = entry.getKey().substring(0, splitIndex);
                        int queueId = Integer.parseInt(entry.getKey().substring(splitIndex + 1));
                        specificOffsets.put(
                                new MessageQueue(topicName, null, queueId), entry.getValue());
                    }

View on GitHub (pinned to cf67b549a7)