apache/seatunnel · error · PulsarConnectorException

SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED

SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED

Error message

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.

What it means

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.

Source

Thrown at seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSource.java:175

                adminConfig,
                partitionDiscoverer,
                partitionDiscoveryIntervalMs,
                consumerMetadataMap,
                getBoundedness(),
                checkpointState.getAssignedPartitions());
    }

    @Override
    public void setJobContext(JobContext jobContext) {
        if (multiTableConfig.isMultiTable()
                && JobMode.BATCH.equals(jobContext.getJobMode())
                && getBoundedness() == Boundedness.UNBOUNDED) {
            List<String> unboundedTables =
                    consumerMetadataMap.entrySet().stream()
                            .filter(e -> e.getValue().getStopCursor() instanceof NeverStopCursor)
                            .map(e -> e.getKey().toString())
                            .collect(Collectors.toList());
            throw new PulsarConnectorException(
                    SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
                    String.format(
                            "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.",
                            String.join(", ", unboundedTables)));
        }
    }

    private Map<TablePath, PulsarConsumerMetadata> createConsumerMetadata(
            CatalogTable singleCatalogTable) {
        Map<TablePath, PulsarConsumerMetadata> metadataMap = new LinkedHashMap<>();
        for (PulsarTableConfig tableConfig : multiTableConfig.getTableConfigs()) {
            CatalogTable catalogTable =
                    multiTableConfig.isTablesConfigs()
                            ? buildCatalogTable(tableConfig)
                            : CatalogTable.of(
                                    TableIdentifier.of(

View on GitHub (pinned to cf67b549a7)

Solutions

1. Change cursor.stop.mode to LATEST or TIMESTAMP for every table listed in the error.
2. Run the job in STREAMING mode instead of batch if unbounded consumption is intended.
3. Set a global bounded stop-cursor default so all tables inherit a bounded mode.

Example fix

// before
source {
  Pulsar {
    tables_configs {
      tables {
        cursor.stop.mode = NEVER
      }
    }
  }
}
// after
source {
  Pulsar {
    tables_configs {
      tables {
        cursor.stop.mode = LATEST
      }
    }
  }
}

for (TableConfig t : tables) {
    if (t.getStopCursor() instanceof NeverStopCursor && jobMode == JobMode.BATCH) {
        throw new IllegalArgumentException("Table " + t + " unbounded in batch mode");
    }
}

boolean isBounded = cursor -> !(cursor instanceof NeverStopCursor);

try {
    source.setJobContext(jobContext);
} catch (PulsarConnectorException e) {
    if (SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED.equals(e.getSeaTunnelErrorCode())) {
        log.error("Batch mode with unbounded tables: {}", e.getMessage());
    }
    throw e;
}

When it happens

Trigger: Running the job in BATCH mode with multiple Pulsar tables where one or more tables configure cursor.stop.mode = NEVER (PulsarSource.java:175, setJobContext).

Common situations: 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.

Related errors

• Stripe source only supports batch jobs
• CommonErrorCode.UNSUPPORTED_DATA_TYPE
• getProducedCatalogTables method has not been implemented.
• The Factory has not been implemented and the deprecated Plug
• Option 'field_delimiter' cannot be empty

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