risingwavelabs/risingwave · critical · SinkError::Redshift

Periodic task for sink id

Error message

Periodic task for sink id {} panicked: {}

What it means

The Redshift sink spawns a background periodic task; `commit_data` detects that the task handle has finished and awaits the JoinError to surface the panic. This error reports that the periodic task panicked (not a normal query failure), meaning commits were not being performed in the background.

Solutions

1. Read the JoinError report and preceding task logs to find the underlying panic cause
2. Restart/recreate the sink to respawn the periodic task
3. Verify Redshift availability and credentials; upgrade RisingWave if the panic is an internal bug and file an issue

if handle.is_finished() {
    let h = self.periodic_task_handle.take().unwrap();
    if let Err(join_err) = h.await {
        // JoinError = task panicked; inspect cause(), alert, and restart sink
        log::error!("periodic task panicked: {:?}", join_err);
    }
}

When it happens

Trigger: The spawned periodic task panics (bug in query execution, DB client dropped/closed, unexpected unwrap/assert) and the next `commit_data(epoch, ...)` observes `handle.is_finished()`.

Common situations: Redshift connection closed mid-run causing a panic path; RisingWave bug or OOM in the task; database restart during COPY leaving the task in a panic state.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• (dynamic: channel send error on shutdown signal)
• (dynamic: UTF-8 conversion error from metadata bytes)
• IcebergSinkWriter should be initialized before barrier
• intermediate.table.name is required for append-only sink
• intermediate.table.name is required for non-append-only sink

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

Appendix: source

Thrown at src/connector/src/sink/snowflake_redshift/redshift.rs:650

                &self.client,
                &self.config,
                self.config.s3_inner.as_ref().ok_or_else(|| {
                    SinkError::Config(anyhow!("S3 configuration is required for redshift s3 sink"))
                })?,
                self.is_append_only,
            )
            .await?;
        }
        Ok(())
    }

    async fn commit_data(&mut self, epoch: u64, metadata: Vec<SinkMetadata>) -> Result<()> {
        if let Some(handle) = &self.periodic_task_handle {
            let is_finished = handle.is_finished();
            if is_finished {
                let handle = self.periodic_task_handle.take().unwrap();
                handle.await.map_err(|e| {
                    SinkError::Redshift(anyhow!(
                        "Periodic task for sink id {} panicked: {}",
                        self.sink_id,
                        e.to_report_string()
                    ))
                })?;
            }
        };
        let paths = metadata
            .into_iter()
            .filter(|m| {
                if let Some(sink_metadata::Metadata::Serialized(SerializedMetadata { metadata })) =
                    &m.metadata
                {
                    !metadata.is_empty()
                } else {
                    false
                }
            })

View on GitHub (pinned to 6469eb736d)