{"record":{"id":"859480e6cca4b97e","repo":"risingwavelabs/risingwave","slug":"google-pub-sub-sink-only-support-append-only-mode","errorCode":null,"errorMessage":"Google Pub/Sub sink only support append-only mode","messagePattern":"Google Pub/Sub sink only support append-only mode","errorType":"validation","errorClass":"SinkError::GooglePubSub","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/google_pubsub.rs","lineNumber":151,"sourceCode":"    fn enforce_secret<'a>(\n        prop_iter: impl Iterator<Item = &'a str>,\n    ) -> crate::error::ConnectorResult<()> {\n        for prop in prop_iter {\n            GooglePubSubConfig::enforce_one(prop)?;\n        }\n        Ok(())\n    }\n}\nimpl Sink for GooglePubSubSink {\n    type LogSinker = AsyncTruncateLogSinkerOf<GooglePubSubSinkWriter>;\n\n    const SINK_NAME: &'static str = PUBSUB_SINK;\n\n    crate::impl_validate_sink_unknown_fields!();\n\n    async fn validate(&self) -> Result<()> {\n        if !self.is_append_only {\n            return Err(SinkError::GooglePubSub(anyhow!(\n                \"Google Pub/Sub sink only support append-only mode\"\n            )));\n        }\n\n        let conf = &self.config;\n        if matches!((&conf.emulator_host, &conf.credentials), (None, None)) {\n            return Err(SinkError::GooglePubSub(anyhow!(\n                \"Configure at least one of `pubsub.emulator_host` and `pubsub.credentials` in the Google Pub/Sub sink\"\n            )));\n        }\n\n        Ok(())\n    }\n\n    async fn new_log_sinker(&self, _writer_param: SinkWriterParam) -> Result<Self::LogSinker> {\n        Ok(GooglePubSubSinkWriter::new(\n            self.config.clone(),\n            self.schema.clone(),","sourceCodeStart":133,"sourceCodeEnd":169,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/google_pubsub.rs#L133-L169","documentation":"Google Pub/Sub publish semantics in this connector are implemented only for append-only streams. If the sink's input contains UPDATE or DELETE records (non-append-only sink_type, e.g. an upsert sink or a sink on a materialized view with updates), validate() rejects it because there is no way to express deletes/updates in the Pub/Sub topic.","triggerScenarios":"Creating a sink with `FORMAT UPSERT` (or any non-append-only sink type) into connector 'google_pubsub'; sinking from a table/MV whose changelog includes updates/deletes.","commonSituations":"Users pointing an upsert sink at Pub/Sub expecting upsert semantics; switching a Kafka upsert sink definition over to google_pubsub without changing the format.","solutions":["Declare the sink as append-only: use `FORMAT APPEND ONLY ENCODE ...`","Sink from an append-only source (e.g. a source or a filtered MV that only emits inserts)","If updates are needed, route through another sink connector that supports upsert (e.g. Kafka upsert, JDBC)"],"exampleFix":"// before\nCREATE SINK s FROM mv WITH (connector='google_pubsub', FORMAT UPSERT ENCODE JSON);\n// after\nCREATE SINK s FROM mv WITH (connector='google_pubsub', FORMAT APPEND ONLY ENCODE JSON);","handlingStrategy":"validation","validationCode":"fn ensure_append_only(sink_format: &str) -> Result<(), String> {\n    if !sink_format.eq_ignore_ascii_case(\"append only\") {\n        return Err(\"Google Pub/Sub sink only supports FORMAT APPEND ONLY\".into());\n    }\n    Ok(())\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always use FORMAT APPEND ONLY with google_pubsub sinks","Do not sink updating MVs/tables to Pub/Sub; use Kafka upsert or JDBC instead","Filter out update/delete records upstream if changelogs are unavoidable"],"tags":["pubsub","sink","append-only"],"backgroundTag":"unsupported-operation","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}