risingwavelabs/risingwave · error · SinkError::GooglePubSub

Google Pub/Sub sink only support append-only mode

Error message

Google Pub/Sub sink only support append-only mode

What it means

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.

Source

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

    fn enforce_secret<'a>(
        prop_iter: impl Iterator<Item = &'a str>,
    ) -> crate::error::ConnectorResult<()> {
        for prop in prop_iter {
            GooglePubSubConfig::enforce_one(prop)?;
        }
        Ok(())
    }
}
impl Sink for GooglePubSubSink {
    type LogSinker = AsyncTruncateLogSinkerOf<GooglePubSubSinkWriter>;

    const SINK_NAME: &'static str = PUBSUB_SINK;

    crate::impl_validate_sink_unknown_fields!();

    async fn validate(&self) -> Result<()> {
        if !self.is_append_only {
            return Err(SinkError::GooglePubSub(anyhow!(
                "Google Pub/Sub sink only support append-only mode"
            )));
        }

        let conf = &self.config;
        if matches!((&conf.emulator_host, &conf.credentials), (None, None)) {
            return Err(SinkError::GooglePubSub(anyhow!(
                "Configure at least one of `pubsub.emulator_host` and `pubsub.credentials` in the Google Pub/Sub sink"
            )));
        }

        Ok(())
    }

    async fn new_log_sinker(&self, _writer_param: SinkWriterParam) -> Result<Self::LogSinker> {
        Ok(GooglePubSubSinkWriter::new(
            self.config.clone(),
            self.schema.clone(),

View on GitHub (pinned to 6469eb736d)

Solutions

1. Declare the sink as append-only: use `FORMAT APPEND ONLY ENCODE ...`
2. Sink from an append-only source (e.g. a source or a filtered MV that only emits inserts)
3. If updates are needed, route through another sink connector that supports upsert (e.g. Kafka upsert, JDBC)

Example fix

// before
CREATE SINK s FROM mv WITH (connector='google_pubsub', FORMAT UPSERT ENCODE JSON);
// after
CREATE SINK s FROM mv WITH (connector='google_pubsub', FORMAT APPEND ONLY ENCODE JSON);

fn ensure_append_only(sink_format: &str) -> Result<(), String> {
    if !sink_format.eq_ignore_ascii_case("append only") {
        return Err("Google Pub/Sub sink only supports FORMAT APPEND ONLY".into());
    }
    Ok(())
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

• only append-only delta lake sink is supported
• `Delete` operation is not supported in `append_only` mode
• `UpdateDelete` operation is not supported in `append_only` m
• File sink only supports append-only mode at present. Please
• `{CHANNEL}` and `{CHANNEL_COLUMN}` only one can be set

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