risingwavelabs/risingwave · error · ConnectorError

PulsarIcebergReader has already been deprecated

Error message

PulsarIcebergReader has already been deprecated

What it means

The PulsarIcebergReader feature (writing Pulsar messages directly into Iceberg from the source reader) has been removed. Setting pulsar.iceberg_loader.enabled (or the equivalent property) to true now aborts reader construction with this deprecation error.

Solutions

1. Remove the iceberg loader property from the Pulsar source WITH clause.
2. Use a dedicated Iceberg sink connector to write data to Iceberg instead of the deprecated in-source loader.
3. Upgrade any tooling/docs referencing PulsarIcebergReader to the sink-based workflow.

Example fix

// before
WITH (connector = 'pulsar', pulsar.iceberg.loader.enabled = true)
// after
WITH (connector = 'pulsar') -- plus a separate Iceberg sink for the data

if props.iceberg_loader_enabled == Some(true) {
    return Err("pulsar iceberg loader is deprecated; use an Iceberg sink instead");
}

fn iceberg_loader_active(props: &PulsarProperties) -> bool {
    props.iceberg_loader_enabled.unwrap_or(false)
}

match reader.new(props, ...).await {
    Err(e) if e.to_string().contains("deprecated") => migrate_to_iceberg_sink_workflow(),
    other => other,
}

When it happens

Trigger: Constructing a Pulsar source reader with props.iceberg_loader_enabled set to Some(true) — typically via `pulsar.iceberg.loader.enabled = true` (or similarly named legacy property) in the WITH clause.

Common situations: Old source definitions carried forward from a version where the Iceberg loader existed; documentation or tutorials referencing the removed feature; users trying to sink Pulsar data to Iceberg through the source connector.

Understand the failure class

Background: "is deprecated and will be removed" — deprecation warnings for old API names, keywords, and options, and how to migrate before the removal release — this error's family across 29 libraries.

Related errors

• `compaction-interval-sec` must be greater than 0 when…
• `compaction.max_snapshots_num` must be greater than 0, got
• `compaction.target_file_size_mb` must be greater than 0
• `compaction.type` must not be set when `write_mode` is…
• `compaction.write_parquet_compression` must be one of

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/5de98126dfc736be. Report an issue: GitHub.

Appendix: source

Thrown at src/connector/src/source/pulsar/source/reader.rs:102

impl SplitReader for PulsarSplitReader {
    type Properties = PulsarProperties;
    type Split = PulsarSplit;

    async fn new(
        props: PulsarProperties,
        splits: Vec<PulsarSplit>,
        parser_config: ParserConfig,
        source_ctx: SourceContextRef,
        _columns: Option<Vec<Column>>,
    ) -> ConnectorResult<Self> {
        ensure!(splits.len() == 1, "only support single split");
        let split = splits.into_iter().next().unwrap();
        let topic = split.topic.to_string();

        tracing::debug!("creating consumer for pulsar split topic {}", topic,);

        if props.iceberg_loader_enabled.unwrap_or(false) {
            bail!("PulsarIcebergReader has already been deprecated");
        } else {
            Ok(Self::Broker(
                PulsarBrokerReader::new(props, vec![split], parser_config, source_ctx, None)
                    .await?,
            ))
        }
    }

    fn into_stream(self) -> BoxSourceChunkStream {
        match self {
            Self::Broker(reader) => {
                let (parser_config, source_context) =
                    (reader.parser_config.clone(), reader.source_ctx.clone());
                Box::pin(into_chunk_stream(
                    reader.into_data_stream(),
                    parser_config,
                    source_context,
                ))

View on GitHub (pinned to 6469eb736d)