{"record":{"id":"7acaca4c489ac545","repo":"apache/seatunnel","slug":"tables-configs-d-start-mode-offsets-is-only-v","errorCode":null,"errorMessage":"tables_configs[%d]: 'start_mode.offsets' is only valid when start_mode=SPECIFIC_OFFSETS","messagePattern":"tables_configs\\[(.+?)\\]: 'start_mode\\.offsets' is only valid when start_mode=SPECIFIC_OFFSETS","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":215,"sourceCode":"        @Override\n        public boolean evaluate(ReadonlyConfig config, List<Map<String, Object>> entries)\n                throws OptionValidationException {\n            if (entries == null || entries.isEmpty()) {\n                return true;\n            }\n            for (int i = 0; i < entries.size(); i++) {\n                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);","sourceCodeStart":197,"sourceCodeEnd":233,"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#L197-L233","documentation":"In tables_configs, start_mode.offsets is only meaningful when start_mode=SPECIFIC_OFFSETS. evaluate() rejects entries that set start_mode=TIMESTAMP but also supply start_mode.offsets, since the two start strategies conflict.","triggerScenarios":"A tables_configs entry combining start_mode=TIMESTAMP with a start_mode.offsets map present (even empty); detected in KafkaSourceFactory.evaluate at job build time.","commonSituations":"Merging/reusing per-table config blocks that accumulate stale options; template configs carrying offsets plus a later-switched timestamp mode.","solutions":["Remove start_mode.offsets from entries where start_mode=TIMESTAMP.","Change start_mode to SPECIFIC_OFFSETS if you actually want explicit offsets.","Audit each tables_configs entry for mutually exclusive options before submission."],"exampleFix":"// before\nstart_mode = TIMESTAMP\nstart_mode.timestamp = 1704067200000\nstart_mode.offsets = {partition-0: 100}\n// after\nstart_mode = TIMESTAMP\nstart_mode.timestamp = 1704067200000","handlingStrategy":"validation","validationCode":"if (!\"SPECIFIC_OFFSETS\".equals(entry.get(\"start_mode\")) && entry.containsKey(\"start_mode.offsets\")) { throw new IllegalArgumentException(\"start_mode.offsets requires start_mode=SPECIFIC_OFFSETS\"); }","typeGuard":null,"tryCatchPattern":"try { factory.evaluate(ctx); } catch (OptionValidationException e) { stripConflictingOptions(ctx); reevaluate(ctx); }","preventionTips":["Keep per-mode option sets disjoint when generating table configs.","Clean stale keys when switching start modes."],"tags":["kafka","config","validation","conflicting-options"],"backgroundTag":"conflicting-config-options","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"}