{"record":{"id":"a9c93f34c7a950ff","repo":"apache/seatunnel","slug":"tables-configs-d-start-mode-end-timestamp-mus","errorCode":null,"errorMessage":"tables_configs[%d]: 'start_mode.end_timestamp' must be >= 0, got: %d","messagePattern":"tables_configs\\[(.+?)\\]: 'start_mode\\.end_timestamp' must be >= 0, got: (.+?)","errorType":"validation","errorClass":"OptionValidationException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java","lineNumber":222,"sourceCode":"                ReadonlyConfig entryConfig = ReadonlyConfig.fromMap(entries.get(i));\n                StartMode startMode =\n                        entryConfig.getOptional(KafkaSourceOptions.START_MODE).orElse(null);\n                if (startMode == StartMode.TIMESTAMP) {\n                    Long ts = entryConfig.get(KafkaSourceOptions.START_MODE_TIMESTAMP);\n                    if (ts == null || ts < 0) {\n                        throw new OptionValidationException(\n                                \"tables_configs[%d]: 'start_mode.timestamp' must be >= 0, got: %d\",\n                                i, ts);\n                    }\n                    if (entries.get(i).containsKey(KafkaSourceOptions.START_MODE_OFFSETS.key())) {\n                        throw new OptionValidationException(\n                                \"tables_configs[%d]: 'start_mode.offsets' is only valid \"\n                                        + \"when start_mode=SPECIFIC_OFFSETS\",\n                                i);\n                    }\n                    Long endTs = entryConfig.get(KafkaSourceOptions.START_MODE_END_TIMESTAMP);\n                    if (endTs != null && endTs < 0) {\n                        throw new OptionValidationException(\n                                \"tables_configs[%d]: 'start_mode.end_timestamp' must be >= 0, got: %d\",\n                                i, endTs);\n                    }\n                } else if (startMode == StartMode.SPECIFIC_OFFSETS) {\n                    Map<String, Long> offsets =\n                            entryConfig.get(KafkaSourceOptions.START_MODE_OFFSETS);\n                    if (offsets == null || offsets.isEmpty()) {\n                        throw new OptionValidationException(\n                                \"tables_configs[%d]: 'start_mode.offsets' must not be empty \"\n                                        + \"when start_mode=SPECIFIC_OFFSETS\",\n                                i);\n                    }\n                    if (entries.get(i).containsKey(KafkaSourceOptions.START_MODE_TIMESTAMP.key())) {\n                        throw new OptionValidationException(\n                                \"tables_configs[%d]: 'start_mode.timestamp' is only valid \"\n                                        + \"when start_mode=TIMESTAMP\",\n                                i);\n                    }","sourceCodeStart":204,"sourceCodeEnd":240,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java#L204-L240","documentation":"When a tables_configs entry uses start_mode=TIMESTAMP, the optional start_mode.end_timestamp must be non-negative. evaluate() throws OptionValidationException when it is negative, mirroring the check on start_mode.timestamp.","triggerScenarios":"A tables_configs entry with start_mode=TIMESTAMP and start_mode.end_timestamp set to a negative number; detected during KafkaSourceFactory.evaluate.","commonSituations":"Negative sentinel values used to mean 'no end'; hand-edited multi-table configs; timestamps accidentally written in seconds as negative from arithmetic errors.","solutions":["Set start_mode.end_timestamp to a non-negative epoch-millis value, or remove it entirely if unbounded.","Fix negative sentinel placeholders in table configs.","Validate all TIMESTAMP entries; the message names the offending index."],"exampleFix":"// before\nstart_mode.end_timestamp = -5\n// after\nstart_mode.end_timestamp = 1704067200000  // or remove the option","handlingStrategy":"validation","validationCode":"Long endTs = entry.get(\"start_mode.end_timestamp\");\nif (\"TIMESTAMP\".equals(entry.get(\"start_mode\")) && endTs != null && endTs < 0) throw new IllegalArgumentException(\"invalid start_mode.end_timestamp\");","typeGuard":null,"tryCatchPattern":"try { factory.evaluate(ctx); } catch (OptionValidationException e) { log.error(\"Fix end_timestamp per: {}\", e.getMessage()); }","preventionTips":["Remove negative sentinels; omit the option for unbounded reads.","Sanity-check all timestamps are non-negative epoch millis."],"tags":["kafka","config","validation","tables-configs"],"backgroundTag":"invalid-config-value","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}