{"record":{"id":"5de98126dfc736be","repo":"risingwavelabs/risingwave","slug":"pulsaricebergreader-has-already-been-deprecated","errorCode":null,"errorMessage":"PulsarIcebergReader has already been deprecated","messagePattern":"PulsarIcebergReader has already been deprecated","errorType":"exception","errorClass":"ConnectorError","httpStatus":null,"severity":"error","filePath":"src/connector/src/source/pulsar/source/reader.rs","lineNumber":102,"sourceCode":"impl SplitReader for PulsarSplitReader {\n    type Properties = PulsarProperties;\n    type Split = PulsarSplit;\n\n    async fn new(\n        props: PulsarProperties,\n        splits: Vec<PulsarSplit>,\n        parser_config: ParserConfig,\n        source_ctx: SourceContextRef,\n        _columns: Option<Vec<Column>>,\n    ) -> ConnectorResult<Self> {\n        ensure!(splits.len() == 1, \"only support single split\");\n        let split = splits.into_iter().next().unwrap();\n        let topic = split.topic.to_string();\n\n        tracing::debug!(\"creating consumer for pulsar split topic {}\", topic,);\n\n        if props.iceberg_loader_enabled.unwrap_or(false) {\n            bail!(\"PulsarIcebergReader has already been deprecated\");\n        } else {\n            Ok(Self::Broker(\n                PulsarBrokerReader::new(props, vec![split], parser_config, source_ctx, None)\n                    .await?,\n            ))\n        }\n    }\n\n    fn into_stream(self) -> BoxSourceChunkStream {\n        match self {\n            Self::Broker(reader) => {\n                let (parser_config, source_context) =\n                    (reader.parser_config.clone(), reader.source_ctx.clone());\n                Box::pin(into_chunk_stream(\n                    reader.into_data_stream(),\n                    parser_config,\n                    source_context,\n                ))","sourceCodeStart":84,"sourceCodeEnd":120,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/source/pulsar/source/reader.rs#L84-L120","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Remove the iceberg loader property from the Pulsar source WITH clause.","Use a dedicated Iceberg sink connector to write data to Iceberg instead of the deprecated in-source loader.","Upgrade any tooling/docs referencing PulsarIcebergReader to the sink-based workflow."],"exampleFix":"// before\nWITH (connector = 'pulsar', pulsar.iceberg.loader.enabled = true)\n// after\nWITH (connector = 'pulsar') -- plus a separate Iceberg sink for the data","handlingStrategy":"validation","validationCode":"if props.iceberg_loader_enabled == Some(true) {\n    return Err(\"pulsar iceberg loader is deprecated; use an Iceberg sink instead\");\n}","typeGuard":"fn iceberg_loader_active(props: &PulsarProperties) -> bool {\n    props.iceberg_loader_enabled.unwrap_or(false)\n}","tryCatchPattern":"match reader.new(props, ...).await {\n    Err(e) if e.to_string().contains(\"deprecated\") => migrate_to_iceberg_sink_workflow(),\n    other => other,\n}","preventionTips":["Remove legacy iceberg loader properties from old Pulsar source DDL during upgrades.","Use a dedicated Iceberg sink connector for writing to Iceberg.","Search deployments for 'iceberg_loader' flags after version upgrades."],"tags":["pulsar","deprecation","iceberg","config"],"backgroundTag":"deprecated-api-usage","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}