risingwavelabs/risingwave · error · SinkError::GooglePubSub

Missing emulator_host or credentials in Google Pub/Sub sink

Error message

Missing emulator_host or credentials in Google Pub/Sub sink

What it means

This is a defensive final branch in sink construction: after handling credentials and emulator_host, neither is set, so the environment cannot be built. Normally validate() catches this earlier with a clearer message; this error surfaces when the sink is constructed without that validation having run (or config was mutated in between).

Source

Thrown at src/connector/src/sink/google_pubsub.rs:240

                SinkError::GooglePubSub(
                    anyhow!(e).context("Failed to create Google Cloud Pub/Sub credentials file"),
                )
            })?;
            let provider =
                DefaultTokenSourceProvider::new_with_credentials(auth_config, Box::new(cred_file))
                    .await
                    .map_err(|e| {
                        SinkError::GooglePubSub(
                            anyhow!(e).context(
                                "Failed to create Google Cloud Pub/Sub token source provider",
                            ),
                        )
                    })?;
            Environment::GoogleCloud(Box::new(provider))
        } else if let Some(emu_host) = config.emulator_host {
            Environment::Emulator(emu_host)
        } else {
            return Err(SinkError::GooglePubSub(anyhow!(
                "Missing emulator_host or credentials in Google Pub/Sub sink"
            )));
        };

        let client_config = ClientConfig {
            endpoint: config.endpoint,
            project_id: Some(config.project_id),
            environment,
            ..Default::default()
        };
        let client = Client::new(client_config)
            .await
            .map_err(|e| SinkError::GooglePubSub(anyhow!(e)))?;

        let topic = async {
            let topic = client.topic(&config.topic);
            if !topic.exists(None).await? {
                topic.create(None, None).await?;

View on GitHub (pinned to 6469eb736d)

Solutions

  1. Set `pubsub.credentials` or `pubsub.emulator_host` in the sink properties
  2. Run validate() on the sink before new() so the earlier, clearer message catches the misconfiguration
  3. Double-check the properties map passed to GooglePubSubConfig::from_btreemap contains the intended keys

Example fix

// before
let config = GooglePubSubConfig { emulator_host: None, credentials: None, .. };
// after
let config = GooglePubSubConfig { emulator_host: Some("localhost:8085".into()), credentials: None, .. };
Defensive patterns

Strategy: validation

Validate before calling

fn ensure_env(config: &GooglePubSubConfig) -> Result<(), String> {
    if config.emulator_host.is_none() && config.credentials.is_none() {
        return Err("missing emulator_host or credentials".into());
    }
    Ok(())
}

Type guard

fn has_pubsub_env(c: &GooglePubSubConfig) -> bool {
    c.emulator_host.is_some() || c.credentials.is_some()
}

Try / catch

match sink.validate().await {
    Err(e) => eprintln!("sink misconfigured: {e:#}"),
    Ok(()) => sink.new().await?,
}

Prevention

When it happens

Trigger: Building GooglePubSubSink::new directly with a config lacking both `credentials` and `emulator_host`, bypassing the validate() path.

Common situations: Programmatic sink construction in tests/tools; config built from a properties map that dropped both keys.

Related errors

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