{"record":{"id":"9417f65981542d1d","repo":"risingwavelabs/risingwave","slug":"configure-at-least-one-of-pubsub-emulator-host-a","errorCode":null,"errorMessage":"Configure at least one of `pubsub.emulator_host` and `pubsub.credentials` in the Google Pub/Sub sink","messagePattern":"Configure at least one of `pubsub\\.emulator_host` and `pubsub\\.credentials` in the Google Pub/Sub sink","errorType":"validation","errorClass":"SinkError::GooglePubSub","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/google_pubsub.rs","lineNumber":158,"sourceCode":"    }\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(),\n            self.pk_indices.clone(),\n            &self.format_desc,\n            self.db_name.clone(),\n            self.sink_from_name.clone(),\n        )\n        .await?\n        .into_log_sinker(PUBSUB_SEND_FUTURE_BUFFER_MAX_SIZE))","sourceCodeStart":140,"sourceCodeEnd":176,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/google_pubsub.rs#L140-L176","documentation":"The Pub/Sub sink requires authentication configuration: either an emulator host (for local testing) or service-account credentials (for real GCP). validate() checks that at least one of `emulator_host` or `credentials` is present and fails the sink creation otherwise, so misconfigured sinks are caught before runtime.","triggerScenarios":"CREATE SINK with connector='google_pubsub' whose properties include neither `pubsub.emulator_host` nor `pubsub.credentials`.","commonSituations":"First-time users forgetting GCP credentials; removing the credentials line when moving configs between environments; typos in the property key so neither field is populated.","solutions":["Add `pubsub.credentials='<service-account-json>'` for production GCP usage","Set `pubsub.emulator_host='localhost:8085'` when testing against the Pub/Sub emulator","Verify exact option key names against GooglePubSubConfig to avoid typos that leave both fields None"],"exampleFix":"// before\nWITH (connector='google_pubsub', topic='mytopic')\n// after\nWITH (connector='google_pubsub', topic='mytopic', pubsub.credentials='<sa-json>')","handlingStrategy":"validation","validationCode":"fn validate_pubsub_auth(props: &std::collections::BTreeMap<String, String>) -> Result<(), String> {\n    let has_emu = props.contains_key(\"pubsub.emulator_host\");\n    let has_cred = props.contains_key(\"pubsub.credentials\");\n    if !has_emu && !has_cred {\n        return Err(\"Set pubsub.emulator_host or pubsub.credentials\".into());\n    }\n    Ok(())\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Store credentials in a secret manager and template them into sink DDL","Use emulator_host in dev and credentials in prod, checking per environment","Search sink DDL for the exact key names pubsub.emulator_host / pubsub.credentials before applying"],"tags":["pubsub","auth","config"],"backgroundTag":"missing-required-config-field","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"}