{"record":{"id":"dcfb86dba03a0464","repo":"apache/seatunnel","slug":"seatunnelapierrorcode-config-validation-failed-dcfb86","errorCode":"SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED","errorMessage":"Pulsar source does not support unbounded multi-table configuration in batch mode. The following tables use cursor.stop.mode=NEVER (unbounded): %s. Either change them to a bounded mode (LATEST or TIMESTAMP) or use STREAMING mode.","messagePattern":"Pulsar source does not support unbounded multi-table configuration in batch mode\\. The following tables use cursor\\.stop\\.mode=NEVER \\(unbounded\\): (.+?)\\. Either change them to a bounded mode \\(LATEST or TIMESTAMP\\) or use STREAMING mode\\.","errorType":"error_code","errorClass":"PulsarConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSource.java","lineNumber":175,"sourceCode":"                adminConfig,\n                partitionDiscoverer,\n                partitionDiscoveryIntervalMs,\n                consumerMetadataMap,\n                getBoundedness(),\n                checkpointState.getAssignedPartitions());\n    }\n\n    @Override\n    public void setJobContext(JobContext jobContext) {\n        if (multiTableConfig.isMultiTable()\n                && JobMode.BATCH.equals(jobContext.getJobMode())\n                && getBoundedness() == Boundedness.UNBOUNDED) {\n            List<String> unboundedTables =\n                    consumerMetadataMap.entrySet().stream()\n                            .filter(e -> e.getValue().getStopCursor() instanceof NeverStopCursor)\n                            .map(e -> e.getKey().toString())\n                            .collect(Collectors.toList());\n            throw new PulsarConnectorException(\n                    SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,\n                    String.format(\n                            \"Pulsar source does not support unbounded multi-table configuration in batch mode. \"\n                                    + \"The following tables use cursor.stop.mode=NEVER (unbounded): %s. \"\n                                    + \"Either change them to a bounded mode (LATEST or TIMESTAMP) or use STREAMING mode.\",\n                            String.join(\", \", unboundedTables)));\n        }\n    }\n\n    private Map<TablePath, PulsarConsumerMetadata> createConsumerMetadata(\n            CatalogTable singleCatalogTable) {\n        Map<TablePath, PulsarConsumerMetadata> metadataMap = new LinkedHashMap<>();\n        for (PulsarTableConfig tableConfig : multiTableConfig.getTableConfigs()) {\n            CatalogTable catalogTable =\n                    multiTableConfig.isTablesConfigs()\n                            ? buildCatalogTable(tableConfig)\n                            : CatalogTable.of(\n                                    TableIdentifier.of(","sourceCodeStart":157,"sourceCodeEnd":193,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSource.java#L157-L193","documentation":"In batch mode all sources must terminate. For multi-table Pulsar source configs, setJobContext validates that no table uses cursor.stop.mode=NEVER (unbounded streaming cursor); if any does, it rejects the job with CONFIG_VALIDATION_FAILED, listing the offending tables and remediation options.","triggerScenarios":"Running the job in BATCH mode with multiple Pulsar tables where one or more tables configure cursor.stop.mode = NEVER (PulsarSource.java:175, setJobContext).","commonSituations":"Copying a streaming job config and switching to batch mode; defaulting stop cursor to NEVER; forgetting to set LATEST/TIMESTAMP stop mode for every table in tables_configs.","solutions":["Change cursor.stop.mode to LATEST or TIMESTAMP for every table listed in the error.","Run the job in STREAMING mode instead of batch if unbounded consumption is intended.","Set a global bounded stop-cursor default so all tables inherit a bounded mode."],"exampleFix":"// before\nsource {\n  Pulsar {\n    tables_configs {\n      tables {\n        cursor.stop.mode = NEVER\n      }\n    }\n  }\n}\n// after\nsource {\n  Pulsar {\n    tables_configs {\n      tables {\n        cursor.stop.mode = LATEST\n      }\n    }\n  }\n}","handlingStrategy":"validation","validationCode":"for (TableConfig t : tables) {\n    if (t.getStopCursor() instanceof NeverStopCursor && jobMode == JobMode.BATCH) {\n        throw new IllegalArgumentException(\"Table \" + t + \" unbounded in batch mode\");\n    }\n}","typeGuard":"boolean isBounded = cursor -> !(cursor instanceof NeverStopCursor);","tryCatchPattern":"try {\n    source.setJobContext(jobContext);\n} catch (PulsarConnectorException e) {\n    if (SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED.equals(e.getSeaTunnelErrorCode())) {\n        log.error(\"Batch mode with unbounded tables: {}\", e.getMessage());\n    }\n    throw e;\n}","preventionTips":["Set cursor.stop.mode to LATEST/TIMESTAMP for all tables when using batch mode.","Keep separate streaming vs batch config templates.","List job mode alongside tables_configs in config review checklists."],"tags":["pulsar","source","batch-mode","boundedness"],"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"}